1use super::{
2 SqlxStore,
3 entities::{
4 scim_managed_connection as connection, scim_managed_connection_event as event,
5 scim_managed_credential as credential,
6 },
7};
8use crate::{
9 model::{self, ActiveRow},
10 pool::Exec,
11 sql::Sql,
12};
13use alibi_core::{
14 AuthResult, AuthSchema,
15 store::{CreateScimManagedConnection, ScimManagedCreation, ScimManagedStore},
16};
17use async_trait::async_trait;
18use chrono::SubsecRound;
19#[async_trait]
20impl<S: AuthSchema> ScimManagedStore for SqlxStore<S> {
21 async fn find_scim_identity_tombstone(
22 &self,
23 connection: &str,
24 key: &str,
25 ) -> AuthResult<Option<alibi_core::store::ScimIdentityTombstone>> {
26 let exec = self.exec();
27 let mut sql = model::select_model::<super::entities::scim_identity_tombstone::Model>(exec);
28 sql.push(" WHERE ");
29 sql.compare("scimIdentityTombstone", "connectionId", " = ", connection);
30 sql.push(" AND ");
31 sql.compare("scimIdentityTombstone", "externalIdKey", " = ", key);
32 model::limit_one(&mut sql);
33 Ok(exec
34 .fetch_optional::<super::entities::scim_identity_tombstone::Model>(sql)
35 .await?
36 .map(|row| alibi_core::store::ScimIdentityTombstone {
37 id: row.id,
38 provisioning_domain_id: row.provisioning_domain_id,
39 user_id: row.user_id,
40 profile: row.profile,
41 }))
42 }
43 async fn scim_user_domain_stable(&self, connection: &str, domain: &str) -> AuthResult<bool> {
44 use super::entities::scim_user as user;
45 let mut sql = model::select_model::<user::Model>(self.exec());
46 sql.push(" WHERE ");
47 sql.compare("scimUser", "connectionId", " = ", connection);
48 sql.push(" AND ");
49 sql.compare("scimUser", "provisioningDomainId", " <> ", domain);
50 model::limit_one(&mut sql);
51 Ok(self
52 .exec()
53 .fetch_optional::<user::Model>(sql)
54 .await?
55 .is_none())
56 }
57 async fn list_scim_users(
58 &self,
59 connection: &str,
60 filters: &[(String, String)],
61 offset: i64,
62 count: i64,
63 ) -> AuthResult<alibi_core::store::ScimUserPage> {
64 use super::entities::scim_user as user;
65 let predicates = |sql: &mut Sql| {
66 sql.push(" WHERE ");
67 sql.compare("scimUser", "connectionId", " = ", connection);
68 for (field, value) in filters {
69 sql.push(" AND ");
70 if matches!(field.as_str(), "emailValueIndex" | "workEmailValueIndex") {
71 sql.compare("scimUser", field, " LIKE ", format!("%{value}%"));
72 } else {
73 sql.compare("scimUser", field, " = ", value.clone());
74 }
75 }
76 };
77 let mut total = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
78 total.ident("scimUser");
79 predicates(&mut total);
80 let total_results = self
81 .exec()
82 .fetch_scalar::<i64>(total)
83 .await?
84 .unwrap_or_default();
85 let users = if count == 0 {
86 Vec::new()
87 } else {
88 let mut sql = model::select_model::<user::Model>(self.exec());
89 predicates(&mut sql);
90 sql.push(" ORDER BY ");
91 sql.column("scimUser", "orderKey");
92 sql.push(" ASC LIMIT ");
93 sql.bind(count);
94 sql.push(" OFFSET ");
95 sql.bind(offset);
96 self.exec().fetch_all::<user::Model>(sql).await?
97 };
98 Ok(alibi_core::store::ScimUserPage {
99 total_results,
100 users: users
101 .into_iter()
102 .map(|row| alibi_core::store::ScimUserRecord {
103 id: row.id,
104 connection_id: row.connection_id,
105 provisioning_domain_id: row.provisioning_domain_id,
106 user_name: row.user_name,
107 primary_email: row.primary_email,
108 display_name: row.display_name,
109 active: row.active,
110 external_id: row.external_id,
111 formatted_name: row.formatted_name,
112 given_name: row.given_name,
113 family_name: row.family_name,
114 serialized_emails: row.serialized_emails,
115 serialized_attributes: row.serialized_attributes,
116 created_at: row.created_at,
117 updated_at: row.updated_at,
118 })
119 .collect(),
120 })
121 }
122
123 async fn scim_group_domain_stable(&self, connection: &str, domain: &str) -> AuthResult<bool> {
124 use super::entities::scim_group as group;
125 let mut sql = model::select_model::<group::Model>(self.exec());
126 sql.push(" WHERE ");
127 sql.compare("scimGroup", "connectionId", " = ", connection);
128 sql.push(" AND ");
129 sql.compare("scimGroup", "provisioningDomainId", " <> ", domain);
130 model::limit_one(&mut sql);
131 Ok(self
132 .exec()
133 .fetch_optional::<group::Model>(sql)
134 .await?
135 .is_none())
136 }
137 async fn list_scim_groups(
138 &self,
139 connection: &str,
140 filters: &[(String, String)],
141 offset: i64,
142 count: i64,
143 ) -> AuthResult<alibi_core::store::ScimGroupPage> {
144 use super::entities::scim_group as group;
145 let predicates = |sql: &mut Sql| {
146 sql.push(" WHERE ");
147 sql.compare("scimGroup", "connectionId", " = ", connection);
148 for (field, value) in filters {
149 sql.push(" AND ");
150 sql.compare("scimGroup", field, " = ", value.clone());
151 }
152 };
153 let mut total = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
154 total.ident("scimGroup");
155 predicates(&mut total);
156 let total_results = self
157 .exec()
158 .fetch_scalar::<i64>(total)
159 .await?
160 .unwrap_or_default();
161 let groups = if count == 0 {
162 Vec::new()
163 } else {
164 let mut sql = model::select_model::<group::Model>(self.exec());
165 predicates(&mut sql);
166 sql.push(" ORDER BY ");
167 sql.column("scimGroup", "orderKey");
168 sql.push(" ASC LIMIT ");
169 sql.bind(count);
170 sql.push(" OFFSET ");
171 sql.bind(offset);
172 self.exec().fetch_all::<group::Model>(sql).await?
173 };
174 Ok(alibi_core::store::ScimGroupPage {
175 total_results,
176 groups: groups
177 .into_iter()
178 .map(|row| alibi_core::store::ScimGroupRecord {
179 id: row.id,
180 connection_id: row.connection_id,
181 provisioning_domain_id: row.provisioning_domain_id,
182 display_name: row.display_name,
183 external_id: row.external_id,
184 created_at: row.created_at,
185 updated_at: row.updated_at,
186 })
187 .collect(),
188 })
189 }
190
191 async fn get_scim_user(
192 &self,
193 id: &str,
194 connection: &str,
195 ) -> AuthResult<Option<alibi_core::store::ScimUserRecord>> {
196 use super::entities::scim_user as user;
197 let mut sql = model::select_model::<user::Model>(self.exec());
198 sql.push(" WHERE ");
199 sql.compare("scimUser", "id", " = ", id);
200 sql.push(" AND ");
201 sql.compare("scimUser", "connectionId", " = ", connection);
202 model::limit_one(&mut sql);
203 let row = self.exec().fetch_optional::<user::Model>(sql).await?;
204 Ok(row.map(|row| alibi_core::store::ScimUserRecord {
205 id: row.id,
206 connection_id: row.connection_id,
207 provisioning_domain_id: row.provisioning_domain_id,
208 user_name: row.user_name,
209 primary_email: row.primary_email,
210 display_name: row.display_name,
211 active: row.active,
212 external_id: row.external_id,
213 formatted_name: row.formatted_name,
214 given_name: row.given_name,
215 family_name: row.family_name,
216 serialized_emails: row.serialized_emails,
217 serialized_attributes: row.serialized_attributes,
218 created_at: row.created_at,
219 updated_at: row.updated_at,
220 }))
221 }
222
223 async fn find_scim_managed_credential(
224 &self,
225 id: &str,
226 ) -> AuthResult<Option<alibi_core::store::ScimManagedCredential>> {
227 let mut sql = model::select_model::<credential::Model>(self.exec());
228 sql.push(" WHERE ");
229 sql.compare("scimManagedCredential", "credentialId", " = ", id);
230 model::limit_one(&mut sql);
231 Ok(self
232 .exec()
233 .fetch_optional::<credential::Model>(sql)
234 .await?
235 .map(Into::into))
236 }
237 async fn find_scim_active_managed_connection(
238 &self,
239 id: &str,
240 ) -> AuthResult<Option<alibi_core::store::ScimManagedConnection>> {
241 let mut sql = model::select_model::<connection::Model>(self.exec());
242 sql.push(" WHERE ");
243 sql.compare("scimManagedConnection", "id", " = ", id);
244 sql.push(" AND ");
245 sql.compare("scimManagedConnection", "status", " = ", "active");
246 model::limit_one(&mut sql);
247 Ok(self
248 .exec()
249 .fetch_optional::<connection::Model>(sql)
250 .await?
251 .map(Into::into))
252 }
253 async fn record_scim_credential_use(
254 &self,
255 id: &str,
256 previous: Option<chrono::DateTime<chrono::Utc>>,
257 at: chrono::DateTime<chrono::Utc>,
258 ) -> AuthResult<()> {
259 let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
260 sql.ident("scimManagedCredential");
261 sql.push(" SET ");
262 sql.ident("lastUsedAt");
263 sql.push(" = ");
264 sql.bind(at);
265 sql.push(" WHERE ");
266 sql.compare("scimManagedCredential", "id", " = ", id);
267 sql.push(" AND ");
268 sql.compare("scimManagedCredential", "status", " = ", "active");
269 sql.push(" AND ");
270 if let Some(previous) = previous {
271 sql.compare("scimManagedCredential", "lastUsedAt", " = ", previous);
272 } else {
273 sql.column("scimManagedCredential", "lastUsedAt");
274 sql.push(" IS NULL");
275 }
276 model::returning::<credential::Model>(&mut sql);
277 _ = self.exec().fetch_optional::<credential::Model>(sql).await?;
278 Ok(())
279 }
280
281 async fn get_scim_group(
282 &self,
283 id: &str,
284 connection: &str,
285 ) -> AuthResult<Option<alibi_core::store::ScimGroupRecord>> {
286 use super::entities::scim_group as group;
287 let mut sql = model::select_model::<group::Model>(self.exec());
288 sql.push(" WHERE ");
289 sql.compare("scimGroup", "id", " = ", id);
290 sql.push(" AND ");
291 sql.compare("scimGroup", "connectionId", " = ", connection);
292 model::limit_one(&mut sql);
293 let row = self.exec().fetch_optional::<group::Model>(sql).await?;
294 Ok(row.map(|row| alibi_core::store::ScimGroupRecord {
295 id: row.id,
296 connection_id: row.connection_id,
297 provisioning_domain_id: row.provisioning_domain_id,
298 display_name: row.display_name,
299 external_id: row.external_id,
300 created_at: row.created_at,
301 updated_at: row.updated_at,
302 }))
303 }
304 async fn scim_group_members(
305 &self,
306 id: &str,
307 connection: &str,
308 ) -> AuthResult<alibi_core::store::ScimGroupMemberships> {
309 use super::entities::{scim_group_member as membership, scim_user as user};
310 let mut sql = model::select_model::<membership::Model>(self.exec());
311 sql.push(" WHERE ");
312 sql.compare("scimGroupMember", "groupId", " = ", id);
313 sql.push(" AND ");
314 sql.compare("scimGroupMember", "connectionId", " = ", connection);
315 sql.push(" ORDER BY ");
316 sql.column("scimGroupMember", "createdAt");
317 sql.push(" ASC LIMIT 1001");
318 let memberships = self.exec().fetch_all::<membership::Model>(sql).await?;
319 let persisted_count = memberships.len();
320 if persisted_count > 1000 {
321 return Ok(alibi_core::store::ScimGroupMemberships {
322 persisted_count,
323 members: Vec::new(),
324 });
325 }
326 let mut ids = Vec::new();
327 for membership in &memberships {
328 if !ids.contains(&membership.scim_user_id) {
329 ids.push(membership.scim_user_id.clone());
330 }
331 }
332 let mut users = std::collections::HashMap::new();
333 for chunk in ids.chunks(500) {
334 let mut sql = model::select_model::<user::Model>(self.exec());
335 sql.push(" WHERE ");
336 sql.compare("scimUser", "connectionId", " = ", connection);
337 sql.push(" AND ");
338 sql.column("scimUser", "id");
339 sql.push(" IN ");
340 sql.bind_list(chunk.iter().cloned());
341 sql.push(" LIMIT ");
342 sql.bind(
343 i64::try_from(chunk.len())
344 .map_err(|error| alibi_core::AuthError::internal(error.to_string()))?,
345 );
346 let rows = self.exec().fetch_all::<user::Model>(sql).await?;
347 for row in rows {
348 _ = users.insert(row.id, row.display_name);
349 }
350 }
351 let members = memberships
352 .into_iter()
353 .filter_map(|membership| {
354 users.get(&membership.scim_user_id).map(|display| {
355 alibi_core::store::ScimGroupMemberRecord {
356 id: membership.scim_user_id.clone(),
357 display_name: display.clone(),
358 }
359 })
360 })
361 .collect();
362 Ok(alibi_core::store::ScimGroupMemberships {
363 persisted_count,
364 members,
365 })
366 }
367 async fn scim_projection_candidates(
368 &self,
369 domain: &str,
370 cursor: Option<&str>,
371 ) -> AuthResult<Vec<alibi_core::store::ScimProjectionCandidate>> {
372 use super::entities::scim_user;
373 let mut sql = model::select_model::<scim_user::Model>(self.exec());
374 sql.push(" WHERE ");
375 sql.compare("scimUser", "provisioningDomainId", " = ", domain);
376 if let Some(cursor) = cursor {
377 sql.push(" AND ");
378 sql.compare("scimUser", "userId", " > ", cursor);
379 }
380 sql.push(" ORDER BY ");
381 sql.ident("userId");
382 sql.push(" ASC LIMIT 51");
383 Ok(self
384 .exec()
385 .fetch_all::<scim_user::Model>(sql)
386 .await?
387 .into_iter()
388 .map(|row| alibi_core::store::ScimProjectionCandidate {
389 id: row.id,
390 user_id: row.user_id,
391 })
392 .collect())
393 }
394
395 async fn acquire_scim_decommission_lease(
396 &self,
397 mut binding: alibi_core::store::ScimConnectionBinding,
398 ) -> AuthResult<Option<alibi_core::store::ScimConnectionBinding>> {
399 let lease = uuid::Uuid::new_v4().simple().to_string();
400 let now = chrono::Utc::now().trunc_subsecs(3);
401 let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
402 sql.ident("scimConnectionBinding");
403 sql.push(" SET ");
404 sql.ident("decommissionRevision");
405 sql.push(" = ");
406 sql.ident("decommissionRevision");
407 sql.push(" + 1, ");
408 for (index, (column, value)) in [
409 ("decommissionStatus", "reconciling".to_string()),
410 ("decommissionLeaseId", lease.clone()),
411 ]
412 .into_iter()
413 .enumerate()
414 {
415 if index > 0 {
416 sql.push(", ");
417 }
418 sql.ident(column);
419 sql.push(" = ");
420 sql.bind(value);
421 }
422 sql.push(", ");
423 sql.ident("decommissionedAt");
424 sql.push(" = ");
425 sql.bind(binding.decommissioned_at.unwrap_or(now));
426 sql.push(", ");
427 sql.ident("decommissionCompletedAt");
428 sql.push(" = NULL, ");
429 sql.ident("decommissionLeaseExpiresAt");
430 sql.push(" = ");
431 sql.bind(now + chrono::Duration::minutes(5));
432 sql.push(" WHERE ");
433 sql.compare("scimConnectionBinding", "id", " = ", binding.id.clone());
434 sql.push(" AND ");
435 sql.compare(
436 "scimConnectionBinding",
437 "decommissionRevision",
438 " = ",
439 binding.decommission_revision,
440 );
441 sql.push(" AND ");
442 sql.compare(
443 "scimConnectionBinding",
444 "decommissionStatus",
445 " = ",
446 binding.decommission_status.clone(),
447 );
448 if self.exec().execute(sql).await? == 0 {
449 return Ok(None);
450 }
451 binding.decommission_revision += 1;
452 self.get_scim_connection_binding_with_exec(self.exec(), &binding.id)
453 .await
454 }
455 async fn release_scim_decommission_lease(&self, id: &str, lease: &str) -> AuthResult<()> {
456 for _ in 0..3 {
457 let mut query =
458 model::select_model::<super::entities::scim_connection_binding::Model>(self.exec());
459 query.push(" WHERE ");
460 query.compare("scimConnectionBinding", "id", " = ", id.to_string());
461 model::limit_one(&mut query);
462 let Ok(Some(current)) = self
463 .exec()
464 .fetch_optional::<super::entities::scim_connection_binding::Model>(query)
465 .await
466 else {
467 break;
468 };
469 if current.decommission_status == "complete"
470 || current.decommission_lease_id.as_deref() != Some(lease)
471 {
472 break;
473 }
474 let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
475 sql.ident("scimConnectionBinding");
476 sql.push(" SET ");
477 sql.ident("decommissionRevision");
478 sql.push(" = ");
479 sql.ident("decommissionRevision");
480 sql.push(" + 1, ");
481 sql.ident("decommissionLeaseId");
482 sql.push(" = NULL, ");
483 sql.ident("decommissionLeaseExpiresAt");
484 sql.push(" = NULL WHERE ");
485 sql.compare("scimConnectionBinding", "id", " = ", id.to_string());
486 sql.push(" AND ");
487 sql.compare(
488 "scimConnectionBinding",
489 "decommissionRevision",
490 " = ",
491 current.decommission_revision,
492 );
493 sql.push(" AND ");
494 sql.compare(
495 "scimConnectionBinding",
496 "decommissionLeaseId",
497 " = ",
498 lease.to_string(),
499 );
500 match self.exec().execute(sql).await {
501 Ok(0) => {}
502 _ => break,
503 }
504 }
505 Ok(())
506 }
507 async fn terminal_scim_connection_binding(
508 &self,
509 id: &str,
510 domain: Option<&str>,
511 key: &str,
512 ) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
513 self.ensure_scim_binding(id, domain, key, true).await
514 }
515 async fn active_scim_connection_binding(
516 &self,
517 id: &str,
518 domain: &str,
519 key: &str,
520 ) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
521 self.ensure_scim_binding(id, Some(domain), key, false).await
522 }
523
524 async fn list_scim_managed_connection_events(
525 &self,
526 id: &str,
527 domain: &str,
528 ) -> AuthResult<Option<Vec<alibi_core::store::ScimManagedConnectionEvent>>> {
529 let exec = self.exec();
530 let mut sql = model::select_model::<connection::Model>(exec);
531 sql.push(" WHERE ");
532 sql.compare("scimManagedConnection", "connectionId", " = ", id);
533 sql.push(" AND ");
534 sql.compare(
535 "scimManagedConnection",
536 "provisioningDomainId",
537 " = ",
538 domain,
539 );
540 model::limit_one(&mut sql);
541 let Some(connection) = exec.fetch_optional::<connection::Model>(sql).await? else {
542 return Ok(None);
543 };
544 let mut sql = model::select_model::<event::Model>(exec);
545 sql.push(" WHERE ");
546 sql.compare(
547 "scimManagedConnectionEvent",
548 "connectionRecordId",
549 " = ",
550 connection.id,
551 );
552 sql.push(" ORDER BY ");
553 sql.ident("sequence");
554 sql.push(" DESC LIMIT 100");
555 let mut events = exec.fetch_all::<event::Model>(sql).await?;
556 events.reverse();
557 Ok(Some(
558 events
559 .into_iter()
560 .map(|event| alibi_core::store::ScimManagedConnectionEvent {
561 sequence: event.sequence,
562 kind: event.kind,
563 actor_id: event.actor_id,
564 credential_id: event.credential_id,
565 created_at: event.created_at,
566 })
567 .collect(),
568 ))
569 }
570
571 async fn decommission_scim_managed_connection(
572 &self,
573 data: alibi_core::store::DecommissionScimManagedConnection,
574 complete: bool,
575 ) -> AuthResult<Option<alibi_core::store::ScimManagedConnection>> {
576 self.in_transaction(true, async move |tx| {
577 let exec = Exec::tx(tx);
578 let mut sql = model::select_model::<connection::Model>(exec);
579 sql.push(" WHERE ");
580 sql.compare(
581 "scimManagedConnection",
582 "connectionId",
583 " = ",
584 data.connection_id,
585 );
586 sql.push(" AND ");
587 sql.compare(
588 "scimManagedConnection",
589 "provisioningDomainId",
590 " = ",
591 data.provisioning_domain_id,
592 );
593 model::limit_one(&mut sql);
594 if exec.engine() == crate::pool::Engine::Postgres {
595 sql.push(" FOR UPDATE");
596 }
597 let mut connection = exec
598 .fetch_optional::<connection::Model>(sql)
599 .await?
600 .ok_or_else(|| alibi_core::AuthError::Api {
601 status: 404,
602 code: None,
603 message: "Managed SCIM connection not found".into(),
604 })?;
605 if connection.status == "decommissioned"
606 || (!complete && connection.status == "decommissioning")
607 {
608 return Ok(Some(connection.into()));
609 }
610 if complete && connection.status != "decommissioning" {
611 return Err(alibi_core::store::scim::scim_managed_conflict(
612 "Managed SCIM connection did not begin decommissioning",
613 ));
614 }
615 let now = chrono::Utc::now().trunc_subsecs(3);
616 let expected = if complete {
617 "decommissioning"
618 } else {
619 "active"
620 };
621 let target = if complete {
622 "decommissioned"
623 } else {
624 "decommissioning"
625 };
626 let mut sql = Sql::with(exec.engine(), "UPDATE ");
627 sql.ident("scimManagedConnection");
628 sql.push(" SET ");
629 sql.ident("revision");
630 sql.push(" = ");
631 sql.ident("revision");
632 sql.push(" + 1, ");
633 sql.ident("status");
634 sql.push(" = ");
635 sql.bind(target);
636 sql.push(", ");
637 sql.ident(if complete {
638 "decommissionedAt"
639 } else {
640 "decommissionStartedAt"
641 });
642 sql.push(" = ");
643 sql.bind(now);
644 sql.push(", ");
645 sql.ident(if complete {
646 "decommissionedBy"
647 } else {
648 "decommissionStartedBy"
649 });
650 sql.push(" = ");
651 sql.bind(data.actor_id.clone());
652 sql.push(" WHERE ");
653 sql.compare("scimManagedConnection", "id", " = ", connection.id.clone());
654 sql.push(" AND ");
655 sql.compare("scimManagedConnection", "status", " = ", expected);
656 sql.push(" AND ");
657 sql.compare(
658 "scimManagedConnection",
659 "revision",
660 " = ",
661 connection.revision,
662 );
663 if exec.execute(sql).await? == 0 {
664 return Ok(None);
665 }
666 connection.revision += 1;
667 connection.status = target.into();
668 if complete {
669 connection.decommissioned_at = Some(now);
670 connection.decommissioned_by = Some(data.actor_id.clone());
671 } else {
672 connection.decommission_started_at = Some(now);
673 connection.decommission_started_by = Some(data.actor_id.clone());
674 }
675 if !complete {
676 let mut sql = model::select_model::<credential::Model>(exec);
677 sql.push(" WHERE ");
678 sql.compare(
679 "scimManagedCredential",
680 "connectionRecordId",
681 " = ",
682 connection.id.clone(),
683 );
684 sql.push(" AND ");
685 sql.compare("scimManagedCredential", "status", " = ", "active");
686 sql.push(" LIMIT 100");
687 for credential in exec.fetch_all::<credential::Model>(sql).await? {
688 let mut sql = Sql::with(exec.engine(), "UPDATE ");
689 sql.ident("scimManagedCredential");
690 sql.push(" SET ");
691 sql.ident("status");
692 sql.push(" = ");
693 sql.bind("decommissioned");
694 sql.push(", ");
695 sql.ident("activeSlotKey");
696 sql.push(" = ");
697 sql.bind(format!("{}:inactive", credential.credential_id));
698 sql.push(", ");
699 sql.ident("decommissionedAt");
700 sql.push(" = ");
701 sql.bind(now);
702 sql.push(" WHERE ");
703 sql.compare("scimManagedCredential", "id", " = ", credential.id);
704 sql.push(" AND ");
705 sql.compare("scimManagedCredential", "status", " = ", "active");
706 sql.push(" AND ");
707 sql.compare(
708 "scimManagedCredential",
709 "activeSlotKey",
710 " = ",
711 credential.active_slot_key,
712 );
713 _ = exec.execute(sql).await?;
714 }
715 }
716 let mut row = ActiveRow::new();
717 row.set(
718 "id",
719 self.generated_id(
720 exec,
721 "scimManagedConnectionEvent",
722 "scimManagedConnectionEvent",
723 "id",
724 )
725 .await?
726 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
727 );
728 row.set("connectionRecordId", connection.id.clone());
729 row.set(
730 "eventKey",
731 format!("{}:{}", connection.id, connection.revision),
732 );
733 row.set("sequence", connection.revision);
734 row.set(
735 "type",
736 if complete {
737 "connection.decommissioned"
738 } else {
739 "connection.decommissioning"
740 },
741 );
742 row.set("actorId", data.actor_id);
743 row.set("credentialId", None::<String>);
744 row.set("createdAt", now);
745 _ = model::insert::<event::Model>(exec, &row).await?;
746 Ok(Some(connection.into()))
747 })
748 .await
749 }
750
751 async fn revoke_scim_managed_credential(
752 &self,
753 data: alibi_core::store::RevokeScimManagedCredential,
754 ) -> AuthResult<Option<alibi_core::store::ScimManagedConnection>> {
755 self.in_transaction(true, async move |tx| {
756 let exec = Exec::tx(tx);
757 let mut sql = model::select_model::<connection::Model>(exec);
758 sql.push(" WHERE ");
759 sql.compare(
760 "scimManagedConnection",
761 "connectionId",
762 " = ",
763 data.connection_id,
764 );
765 sql.push(" AND ");
766 sql.compare(
767 "scimManagedConnection",
768 "provisioningDomainId",
769 " = ",
770 data.provisioning_domain_id,
771 );
772 model::limit_one(&mut sql);
773 if exec.engine() == crate::pool::Engine::Postgres {
774 sql.push(" FOR UPDATE");
775 }
776 let mut connection = exec
777 .fetch_optional::<connection::Model>(sql)
778 .await?
779 .ok_or_else(|| alibi_core::AuthError::Api {
780 status: 404,
781 code: None,
782 message: "Managed SCIM connection not found".into(),
783 })?;
784 let mut sql = model::select_model::<credential::Model>(exec);
785 sql.push(" WHERE ");
786 sql.compare(
787 "scimManagedCredential",
788 "credentialId",
789 " = ",
790 data.credential_id.clone(),
791 );
792 sql.push(" AND ");
793 sql.compare(
794 "scimManagedCredential",
795 "connectionRecordId",
796 " = ",
797 connection.id.clone(),
798 );
799 model::limit_one(&mut sql);
800 let credential = exec
801 .fetch_optional::<credential::Model>(sql)
802 .await?
803 .ok_or_else(|| alibi_core::AuthError::Api {
804 status: 404,
805 code: None,
806 message: "Managed SCIM connection not found".into(),
807 })?;
808 if credential.status == "revoked" {
809 return Ok(Some(connection.into()));
810 }
811 if connection.status != "active" || credential.status != "active" {
812 return Err(alibi_core::store::scim::scim_managed_conflict(
813 "Managed SCIM credential is not active",
814 ));
815 }
816 let mut sql = Sql::with(exec.engine(), "UPDATE ");
817 sql.ident("scimManagedConnection");
818 sql.push(" SET ");
819 sql.ident("revision");
820 sql.push(" = ");
821 sql.ident("revision");
822 sql.push(" + 1 WHERE ");
823 sql.compare("scimManagedConnection", "id", " = ", connection.id.clone());
824 sql.push(" AND ");
825 sql.compare("scimManagedConnection", "status", " = ", "active");
826 sql.push(" AND ");
827 sql.compare(
828 "scimManagedConnection",
829 "revision",
830 " = ",
831 connection.revision,
832 );
833 if exec.execute(sql).await? == 0 {
834 return Ok(None);
835 }
836 connection.revision += 1;
837 let revoked_at = chrono::Utc::now().trunc_subsecs(3);
838 let mut sql = Sql::with(exec.engine(), "UPDATE ");
839 sql.ident("scimManagedCredential");
840 sql.push(" SET ");
841 for (index, (column, value)) in [
842 ("status", "revoked".to_string()),
843 (
844 "activeSlotKey",
845 format!("{}:inactive", credential.credential_id),
846 ),
847 ("revokedBy", data.actor_id.clone()),
848 ]
849 .into_iter()
850 .enumerate()
851 {
852 if index > 0 {
853 sql.push(", ");
854 }
855 sql.ident(column);
856 sql.push(" = ");
857 sql.bind(value);
858 }
859 sql.push(", ");
860 sql.ident("revokedAt");
861 sql.push(" = ");
862 sql.bind(revoked_at);
863 sql.push(" WHERE ");
864 sql.compare("scimManagedCredential", "id", " = ", credential.id);
865 sql.push(" AND ");
866 sql.compare("scimManagedCredential", "status", " = ", "active");
867 sql.push(" AND ");
868 sql.compare(
869 "scimManagedCredential",
870 "activeSlotKey",
871 " = ",
872 credential.active_slot_key,
873 );
874 if exec.execute(sql).await? == 0 {
875 return Ok(None);
876 }
877 let mut row = ActiveRow::new();
878 row.set(
879 "id",
880 self.generated_id(
881 exec,
882 "scimManagedConnectionEvent",
883 "scimManagedConnectionEvent",
884 "id",
885 )
886 .await?
887 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
888 );
889 row.set("connectionRecordId", connection.id.clone());
890 row.set(
891 "eventKey",
892 format!("{}:{}", connection.id, connection.revision),
893 );
894 row.set("sequence", connection.revision);
895 row.set("type", "credential.revoked");
896 row.set("actorId", data.actor_id);
897 row.set("credentialId", Some(data.credential_id));
898 row.set("createdAt", revoked_at);
899 _ = model::insert::<event::Model>(exec, &row).await?;
900 Ok(Some(connection.into()))
901 })
902 .await
903 }
904
905 async fn rotate_scim_managed_credential(
906 &self,
907 data: alibi_core::store::RotateScimManagedCredential,
908 ) -> AuthResult<Option<ScimManagedCreation>> {
909 self.in_transaction(true, async move |tx| {
910 let exec = Exec::tx(tx);
911 let mut sql = model::select_model::<connection::Model>(exec);
912 sql.push(" WHERE ");
913 sql.compare(
914 "scimManagedConnection",
915 "connectionId",
916 " = ",
917 data.connection_id,
918 );
919 sql.push(" AND ");
920 sql.compare(
921 "scimManagedConnection",
922 "provisioningDomainId",
923 " = ",
924 data.provisioning_domain_id,
925 );
926 model::limit_one(&mut sql);
927 if exec.engine() == crate::pool::Engine::Postgres {
928 sql.push(" FOR UPDATE");
929 }
930 let mut connection = exec
931 .fetch_optional::<connection::Model>(sql)
932 .await?
933 .ok_or_else(|| alibi_core::AuthError::Api {
934 status: 404,
935 code: None,
936 message: "Managed SCIM connection not found".into(),
937 })?;
938 if connection.status != "active" {
939 return Err(alibi_core::store::scim::scim_managed_conflict(
940 "Managed SCIM connection is not active",
941 ));
942 }
943 let mut sql = Sql::with(exec.engine(), "UPDATE ");
944 sql.ident("scimManagedConnection");
945 sql.push(" SET ");
946 sql.ident("revision");
947 sql.push(" = ");
948 sql.ident("revision");
949 sql.push(" + 1 WHERE ");
950 sql.compare("scimManagedConnection", "id", " = ", connection.id.clone());
951 sql.push(" AND ");
952 sql.compare("scimManagedConnection", "status", " = ", "active");
953 sql.push(" AND ");
954 sql.compare(
955 "scimManagedConnection",
956 "revision",
957 " = ",
958 connection.revision,
959 );
960 if exec.execute(sql).await? == 0 {
961 return Ok(None);
962 }
963 connection.revision += 1;
964 let mut sql = model::select_model::<credential::Model>(exec);
965 sql.push(" WHERE ");
966 sql.compare(
967 "scimManagedCredential",
968 "connectionRecordId",
969 " = ",
970 connection.id.clone(),
971 );
972 sql.push(" AND ");
973 sql.compare("scimManagedCredential", "status", " = ", "active");
974 sql.push(" LIMIT 101");
975 let active = exec.fetch_all::<credential::Model>(sql).await?;
976 let mut slots = std::collections::HashSet::new();
977 for credential in active {
978 if credential.expires_at <= data.created_at {
979 let mut row = ActiveRow::new();
980 row.set("id", credential.id);
981 row.set("status", "expired");
982 row.set(
983 "activeSlotKey",
984 format!("{}:inactive", credential.credential_id),
985 );
986 _ = model::update::<credential::Model>(exec, &row).await?;
987 } else {
988 _ = slots.insert(credential.active_slot_key);
989 }
990 }
991 if slots.len() >= data.max_active_credentials {
992 return Err(alibi_core::store::scim::scim_managed_conflict(
993 "Managed SCIM connection has the maximum number of active credentials",
994 ));
995 }
996 let slot = (0..data.max_active_credentials)
997 .find(|slot| !slots.contains(&format!("{}:active:{slot}", connection.id)))
998 .ok_or_else(|| {
999 alibi_core::store::scim::scim_managed_conflict(
1000 "Managed SCIM connection has the maximum number of active credentials",
1001 )
1002 })?;
1003 let mut row = ActiveRow::new();
1004 row.set(
1005 "id",
1006 self.generated_id(exec, "scimManagedCredential", "scimManagedCredential", "id")
1007 .await?
1008 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1009 );
1010 row.set("connectionRecordId", connection.id.clone());
1011 row.set("credentialId", data.credential_id.clone());
1012 row.set("tokenDigest", data.token_digest);
1013 row.set("hashVersion", "v1");
1014 row.set("activeSlotKey", format!("{}:active:{slot}", connection.id));
1015 row.set("status", "active");
1016 row.set("serializedScopes", data.serialized_scopes);
1017 row.set("expiresAt", data.expires_at);
1018 row.set("createdAt", data.created_at);
1019 row.set("createdBy", data.actor_id.clone());
1020 let credential = model::insert::<credential::Model>(exec, &row).await?;
1021 let mut row = ActiveRow::new();
1022 row.set(
1023 "id",
1024 self.generated_id(
1025 exec,
1026 "scimManagedConnectionEvent",
1027 "scimManagedConnectionEvent",
1028 "id",
1029 )
1030 .await?
1031 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1032 );
1033 row.set("connectionRecordId", connection.id.clone());
1034 row.set(
1035 "eventKey",
1036 format!("{}:{}", connection.id, connection.revision),
1037 );
1038 row.set("sequence", connection.revision);
1039 row.set("type", "credential.rotated");
1040 row.set("actorId", data.actor_id);
1041 row.set("credentialId", Some(data.credential_id));
1042 row.set("createdAt", data.created_at);
1043 _ = model::insert::<event::Model>(exec, &row).await?;
1044 Ok(Some(ScimManagedCreation {
1045 connection: connection.into(),
1046 credential: credential.into(),
1047 }))
1048 })
1049 .await
1050 }
1051
1052 async fn get_scim_managed_connection(
1053 &self,
1054 id: &str,
1055 domain: &str,
1056 ) -> AuthResult<Option<alibi_core::store::ScimManagedConnectionState>> {
1057 let exec = self.exec();
1058 let mut sql = model::select_model::<connection::Model>(exec);
1059 sql.push(" WHERE ");
1060 sql.compare("scimManagedConnection", "connectionId", " = ", id);
1061 sql.push(" AND ");
1062 sql.compare(
1063 "scimManagedConnection",
1064 "provisioningDomainId",
1065 " = ",
1066 domain,
1067 );
1068 model::limit_one(&mut sql);
1069 let Some(connection) = exec.fetch_optional::<connection::Model>(sql).await? else {
1070 return Ok(None);
1071 };
1072 let mut sql = model::select_model::<credential::Model>(exec);
1073 sql.push(" WHERE ");
1074 sql.compare(
1075 "scimManagedCredential",
1076 "connectionRecordId",
1077 " = ",
1078 connection.id.clone(),
1079 );
1080 sql.push(" ORDER BY ");
1081 sql.ident("createdAt");
1082 sql.push(" DESC LIMIT ");
1083 sql.bind(self.find_many_limit());
1084 let credentials = exec
1085 .fetch_all::<credential::Model>(sql)
1086 .await?
1087 .into_iter()
1088 .map(Into::into)
1089 .collect();
1090 Ok(Some(alibi_core::store::ScimManagedConnectionState {
1091 connection: connection.into(),
1092 credentials,
1093 }))
1094 }
1095
1096 async fn list_scim_managed_connections(
1097 &self,
1098 domain: &str,
1099 ) -> AuthResult<Vec<alibi_core::store::ScimManagedConnection>> {
1100 let mut sql = model::select_model::<connection::Model>(self.exec());
1101 sql.push(" WHERE ");
1102 sql.compare(
1103 "scimManagedConnection",
1104 "provisioningDomainId",
1105 " = ",
1106 domain,
1107 );
1108 sql.push(" ORDER BY ");
1109 sql.ident("createdAt");
1110 sql.push(" DESC LIMIT ");
1111 sql.bind(self.find_many_limit());
1112 Ok(self
1113 .exec()
1114 .fetch_all::<connection::Model>(sql)
1115 .await?
1116 .into_iter()
1117 .map(Into::into)
1118 .collect())
1119 }
1120
1121 async fn create_scim_managed_connection(
1122 &self,
1123 data: CreateScimManagedConnection,
1124 ) -> AuthResult<ScimManagedCreation> {
1125 let request_id = data.creation_request_id.clone();
1126 let result = self
1127 .in_transaction(true, async move |tx| {
1128 let exec = Exec::tx(tx);
1129 let mut row = ActiveRow::new();
1130 row.set(
1131 "id",
1132 self.generated_id(exec, "scimManagedConnection", "scimManagedConnection", "id")
1133 .await?
1134 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1135 );
1136 row.set("creationRequestId", data.creation_request_id);
1137 row.set("connectionId", data.connection_id);
1138 row.set("provisioningDomainId", data.provisioning_domain_id);
1139 row.set("status", "active");
1140 row.set("revision", 2_i64);
1141 row.set("createdAt", data.created_at);
1142 row.set("createdBy", data.actor_id.clone());
1143 let connection = model::insert::<connection::Model>(exec, &row).await?;
1144 let mut row = ActiveRow::new();
1145 row.set(
1146 "id",
1147 self.generated_id(exec, "scimManagedCredential", "scimManagedCredential", "id")
1148 .await?
1149 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1150 );
1151 row.set("connectionRecordId", connection.id.clone());
1152 row.set("credentialId", data.credential_id.clone());
1153 row.set("tokenDigest", data.token_digest);
1154 row.set("hashVersion", "v1");
1155 row.set("activeSlotKey", format!("{}:active:0", connection.id));
1156 row.set("status", "active");
1157 row.set("serializedScopes", data.serialized_scopes);
1158 row.set("expiresAt", data.expires_at);
1159 row.set("createdAt", data.created_at);
1160 row.set("createdBy", data.actor_id.clone());
1161 let credential = model::insert::<credential::Model>(exec, &row).await?;
1162 for (sequence, kind) in
1163 [(1_i64, "connection.created"), (2_i64, "credential.issued")]
1164 {
1165 let mut row = ActiveRow::new();
1166 row.set(
1167 "id",
1168 self.generated_id(
1169 exec,
1170 "scimManagedConnectionEvent",
1171 "scimManagedConnectionEvent",
1172 "id",
1173 )
1174 .await?
1175 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1176 );
1177 row.set("connectionRecordId", connection.id.clone());
1178 row.set("eventKey", format!("{}:{sequence}", connection.id));
1179 row.set("sequence", sequence);
1180 row.set("type", kind);
1181 row.set("actorId", data.actor_id.clone());
1182 row.set(
1183 "credentialId",
1184 if sequence == 2 {
1185 Some(data.credential_id.clone())
1186 } else {
1187 None
1188 },
1189 );
1190 row.set("createdAt", data.created_at);
1191 _ = model::insert::<event::Model>(exec, &row).await?;
1192 }
1193 Ok(ScimManagedCreation {
1194 connection: connection.into(),
1195 credential: credential.into(),
1196 })
1197 })
1198 .await;
1199 if result.is_err() {
1200 let mut sql = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
1201 sql.ident("scimManagedConnection");
1202 sql.push(" WHERE ");
1203 sql.compare(
1204 "scimManagedConnection",
1205 "creationRequestId",
1206 " = ",
1207 request_id,
1208 );
1209 if self.exec().fetch_scalar::<i64>(sql).await?.unwrap_or(0) > 0 {
1210 return Err(alibi_core::store::scim::scim_creation_conflict());
1211 }
1212 }
1213 result
1214 }
1215}
1216
1217impl<S: AuthSchema> SqlxStore<S> {
1218 async fn ensure_scim_binding(
1219 &self,
1220 id: &str,
1221 domain: Option<&str>,
1222 key: &str,
1223 terminal: bool,
1224 ) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
1225 use super::entities::scim_connection_binding as binding;
1226 let find = async || {
1227 let mut sql = model::select_model::<binding::Model>(self.exec());
1228 sql.push(" WHERE ");
1229 sql.compare("scimConnectionBinding", "connectionKey", " = ", key);
1230 model::limit_one(&mut sql);
1231 self.exec().fetch_optional::<binding::Model>(sql).await
1232 };
1233 if let Some(row) = find().await? {
1234 let binding = row.into();
1235 if terminal {
1236 alibi_core::store::scim::validate_scim_binding_domain(&binding, domain)?;
1237 }
1238 return Ok(binding);
1239 }
1240 let domain = domain.ok_or_else(|| {
1241 alibi_core::AuthError::internal(format!(
1242 "SCIM connection \"{id}\" has no persisted binding."
1243 ))
1244 })?;
1245 let now = chrono::Utc::now().trunc_subsecs(3);
1246 let mut row = ActiveRow::new();
1247 row.set(
1248 "id",
1249 self.generated_id(
1250 self.exec(),
1251 "scimConnectionBinding",
1252 "scimConnectionBinding",
1253 "id",
1254 )
1255 .await?
1256 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1257 );
1258 row.set("connectionId", id);
1259 row.set("connectionKey", key);
1260 row.set("provisioningDomainId", domain);
1261 row.set("createdAt", now);
1262 row.set(
1263 "decommissionStatus",
1264 if terminal { "complete" } else { "active" },
1265 );
1266 if terminal {
1267 row.set("decommissionedAt", now);
1268 row.set("decommissionCompletedAt", now);
1269 }
1270 row.set("decommissionReconciledUserCount", 0_i64);
1271 row.set("decommissionBatchCount", 0_i64);
1272 row.set("decommissionRevision", 0_i64);
1273 match model::insert::<binding::Model>(self.exec(), &row).await {
1274 Ok(row) => Ok(row.into()),
1275 Err(error) => {
1276 if let Some(row) = find().await? {
1277 let binding = row.into();
1278 if terminal {
1279 alibi_core::store::scim::validate_scim_binding_domain(
1280 &binding,
1281 Some(domain),
1282 )?;
1283 }
1284 Ok(binding)
1285 } else {
1286 Err(error)
1287 }
1288 }
1289 }
1290 }
1291}