use super::{
SqlxStore,
entities::{
scim_managed_connection as connection, scim_managed_connection_event as event,
scim_managed_credential as credential,
},
};
use crate::{
model::{self, ActiveRow},
pool::Exec,
sql::Sql,
};
use alibi_core::{
AuthResult, AuthSchema,
store::{CreateScimManagedConnection, ScimManagedCreation, ScimManagedStore},
};
use async_trait::async_trait;
use chrono::SubsecRound;
#[async_trait]
impl<S: AuthSchema> ScimManagedStore for SqlxStore<S> {
async fn find_scim_identity_tombstone(
&self,
connection: &str,
key: &str,
) -> AuthResult<Option<alibi_core::store::ScimIdentityTombstone>> {
let exec = self.exec();
let mut sql = model::select_model::<super::entities::scim_identity_tombstone::Model>(exec);
sql.push(" WHERE ");
sql.compare("scimIdentityTombstone", "connectionId", " = ", connection);
sql.push(" AND ");
sql.compare("scimIdentityTombstone", "externalIdKey", " = ", key);
model::limit_one(&mut sql);
Ok(exec
.fetch_optional::<super::entities::scim_identity_tombstone::Model>(sql)
.await?
.map(|row| alibi_core::store::ScimIdentityTombstone {
id: row.id,
provisioning_domain_id: row.provisioning_domain_id,
user_id: row.user_id,
profile: row.profile,
}))
}
async fn scim_user_domain_stable(&self, connection: &str, domain: &str) -> AuthResult<bool> {
use super::entities::scim_user as user;
let mut sql = model::select_model::<user::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimUser", "connectionId", " = ", connection);
sql.push(" AND ");
sql.compare("scimUser", "provisioningDomainId", " <> ", domain);
model::limit_one(&mut sql);
Ok(self
.exec()
.fetch_optional::<user::Model>(sql)
.await?
.is_none())
}
async fn list_scim_users(
&self,
connection: &str,
filters: &[(String, String)],
offset: i64,
count: i64,
) -> AuthResult<alibi_core::store::ScimUserPage> {
use super::entities::scim_user as user;
let predicates = |sql: &mut Sql| {
sql.push(" WHERE ");
sql.compare("scimUser", "connectionId", " = ", connection);
for (field, value) in filters {
sql.push(" AND ");
if matches!(field.as_str(), "emailValueIndex" | "workEmailValueIndex") {
sql.compare("scimUser", field, " LIKE ", format!("%{value}%"));
} else {
sql.compare("scimUser", field, " = ", value.clone());
}
}
};
let mut total = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
total.ident("scimUser");
predicates(&mut total);
let total_results = self
.exec()
.fetch_scalar::<i64>(total)
.await?
.unwrap_or_default();
let users = if count == 0 {
Vec::new()
} else {
let mut sql = model::select_model::<user::Model>(self.exec());
predicates(&mut sql);
sql.push(" ORDER BY ");
sql.column("scimUser", "orderKey");
sql.push(" ASC LIMIT ");
sql.bind(count);
sql.push(" OFFSET ");
sql.bind(offset);
self.exec().fetch_all::<user::Model>(sql).await?
};
Ok(alibi_core::store::ScimUserPage {
total_results,
users: users
.into_iter()
.map(|row| alibi_core::store::ScimUserRecord {
id: row.id,
connection_id: row.connection_id,
provisioning_domain_id: row.provisioning_domain_id,
user_name: row.user_name,
primary_email: row.primary_email,
display_name: row.display_name,
active: row.active,
external_id: row.external_id,
formatted_name: row.formatted_name,
given_name: row.given_name,
family_name: row.family_name,
serialized_emails: row.serialized_emails,
serialized_attributes: row.serialized_attributes,
created_at: row.created_at,
updated_at: row.updated_at,
})
.collect(),
})
}
async fn scim_group_domain_stable(&self, connection: &str, domain: &str) -> AuthResult<bool> {
use super::entities::scim_group as group;
let mut sql = model::select_model::<group::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimGroup", "connectionId", " = ", connection);
sql.push(" AND ");
sql.compare("scimGroup", "provisioningDomainId", " <> ", domain);
model::limit_one(&mut sql);
Ok(self
.exec()
.fetch_optional::<group::Model>(sql)
.await?
.is_none())
}
async fn list_scim_groups(
&self,
connection: &str,
filters: &[(String, String)],
offset: i64,
count: i64,
) -> AuthResult<alibi_core::store::ScimGroupPage> {
use super::entities::scim_group as group;
let predicates = |sql: &mut Sql| {
sql.push(" WHERE ");
sql.compare("scimGroup", "connectionId", " = ", connection);
for (field, value) in filters {
sql.push(" AND ");
sql.compare("scimGroup", field, " = ", value.clone());
}
};
let mut total = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
total.ident("scimGroup");
predicates(&mut total);
let total_results = self
.exec()
.fetch_scalar::<i64>(total)
.await?
.unwrap_or_default();
let groups = if count == 0 {
Vec::new()
} else {
let mut sql = model::select_model::<group::Model>(self.exec());
predicates(&mut sql);
sql.push(" ORDER BY ");
sql.column("scimGroup", "orderKey");
sql.push(" ASC LIMIT ");
sql.bind(count);
sql.push(" OFFSET ");
sql.bind(offset);
self.exec().fetch_all::<group::Model>(sql).await?
};
Ok(alibi_core::store::ScimGroupPage {
total_results,
groups: groups
.into_iter()
.map(|row| alibi_core::store::ScimGroupRecord {
id: row.id,
connection_id: row.connection_id,
provisioning_domain_id: row.provisioning_domain_id,
display_name: row.display_name,
external_id: row.external_id,
created_at: row.created_at,
updated_at: row.updated_at,
})
.collect(),
})
}
async fn get_scim_user(
&self,
id: &str,
connection: &str,
) -> AuthResult<Option<alibi_core::store::ScimUserRecord>> {
use super::entities::scim_user as user;
let mut sql = model::select_model::<user::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimUser", "id", " = ", id);
sql.push(" AND ");
sql.compare("scimUser", "connectionId", " = ", connection);
model::limit_one(&mut sql);
let row = self.exec().fetch_optional::<user::Model>(sql).await?;
Ok(row.map(|row| alibi_core::store::ScimUserRecord {
id: row.id,
connection_id: row.connection_id,
provisioning_domain_id: row.provisioning_domain_id,
user_name: row.user_name,
primary_email: row.primary_email,
display_name: row.display_name,
active: row.active,
external_id: row.external_id,
formatted_name: row.formatted_name,
given_name: row.given_name,
family_name: row.family_name,
serialized_emails: row.serialized_emails,
serialized_attributes: row.serialized_attributes,
created_at: row.created_at,
updated_at: row.updated_at,
}))
}
async fn find_scim_managed_credential(
&self,
id: &str,
) -> AuthResult<Option<alibi_core::store::ScimManagedCredential>> {
let mut sql = model::select_model::<credential::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimManagedCredential", "credentialId", " = ", id);
model::limit_one(&mut sql);
Ok(self
.exec()
.fetch_optional::<credential::Model>(sql)
.await?
.map(Into::into))
}
async fn find_scim_active_managed_connection(
&self,
id: &str,
) -> AuthResult<Option<alibi_core::store::ScimManagedConnection>> {
let mut sql = model::select_model::<connection::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimManagedConnection", "id", " = ", id);
sql.push(" AND ");
sql.compare("scimManagedConnection", "status", " = ", "active");
model::limit_one(&mut sql);
Ok(self
.exec()
.fetch_optional::<connection::Model>(sql)
.await?
.map(Into::into))
}
async fn record_scim_credential_use(
&self,
id: &str,
previous: Option<chrono::DateTime<chrono::Utc>>,
at: chrono::DateTime<chrono::Utc>,
) -> AuthResult<()> {
let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
sql.ident("scimManagedCredential");
sql.push(" SET ");
sql.ident("lastUsedAt");
sql.push(" = ");
sql.bind(at);
sql.push(" WHERE ");
sql.compare("scimManagedCredential", "id", " = ", id);
sql.push(" AND ");
sql.compare("scimManagedCredential", "status", " = ", "active");
sql.push(" AND ");
if let Some(previous) = previous {
sql.compare("scimManagedCredential", "lastUsedAt", " = ", previous);
} else {
sql.column("scimManagedCredential", "lastUsedAt");
sql.push(" IS NULL");
}
model::returning::<credential::Model>(&mut sql);
_ = self.exec().fetch_optional::<credential::Model>(sql).await?;
Ok(())
}
async fn get_scim_group(
&self,
id: &str,
connection: &str,
) -> AuthResult<Option<alibi_core::store::ScimGroupRecord>> {
use super::entities::scim_group as group;
let mut sql = model::select_model::<group::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimGroup", "id", " = ", id);
sql.push(" AND ");
sql.compare("scimGroup", "connectionId", " = ", connection);
model::limit_one(&mut sql);
let row = self.exec().fetch_optional::<group::Model>(sql).await?;
Ok(row.map(|row| alibi_core::store::ScimGroupRecord {
id: row.id,
connection_id: row.connection_id,
provisioning_domain_id: row.provisioning_domain_id,
display_name: row.display_name,
external_id: row.external_id,
created_at: row.created_at,
updated_at: row.updated_at,
}))
}
async fn scim_group_members(
&self,
id: &str,
connection: &str,
) -> AuthResult<alibi_core::store::ScimGroupMemberships> {
use super::entities::{scim_group_member as membership, scim_user as user};
let mut sql = model::select_model::<membership::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimGroupMember", "groupId", " = ", id);
sql.push(" AND ");
sql.compare("scimGroupMember", "connectionId", " = ", connection);
sql.push(" ORDER BY ");
sql.column("scimGroupMember", "createdAt");
sql.push(" ASC LIMIT 1001");
let memberships = self.exec().fetch_all::<membership::Model>(sql).await?;
let persisted_count = memberships.len();
if persisted_count > 1000 {
return Ok(alibi_core::store::ScimGroupMemberships {
persisted_count,
members: Vec::new(),
});
}
let mut ids = Vec::new();
for membership in &memberships {
if !ids.contains(&membership.scim_user_id) {
ids.push(membership.scim_user_id.clone());
}
}
let mut users = std::collections::HashMap::new();
for chunk in ids.chunks(500) {
let mut sql = model::select_model::<user::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimUser", "connectionId", " = ", connection);
sql.push(" AND ");
sql.column("scimUser", "id");
sql.push(" IN ");
sql.bind_list(chunk.iter().cloned());
sql.push(" LIMIT ");
sql.bind(
i64::try_from(chunk.len())
.map_err(|error| alibi_core::AuthError::internal(error.to_string()))?,
);
let rows = self.exec().fetch_all::<user::Model>(sql).await?;
for row in rows {
_ = users.insert(row.id, row.display_name);
}
}
let members = memberships
.into_iter()
.filter_map(|membership| {
users.get(&membership.scim_user_id).map(|display| {
alibi_core::store::ScimGroupMemberRecord {
id: membership.scim_user_id.clone(),
display_name: display.clone(),
}
})
})
.collect();
Ok(alibi_core::store::ScimGroupMemberships {
persisted_count,
members,
})
}
async fn scim_projection_candidates(
&self,
domain: &str,
cursor: Option<&str>,
) -> AuthResult<Vec<alibi_core::store::ScimProjectionCandidate>> {
use super::entities::scim_user;
let mut sql = model::select_model::<scim_user::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimUser", "provisioningDomainId", " = ", domain);
if let Some(cursor) = cursor {
sql.push(" AND ");
sql.compare("scimUser", "userId", " > ", cursor);
}
sql.push(" ORDER BY ");
sql.ident("userId");
sql.push(" ASC LIMIT 51");
Ok(self
.exec()
.fetch_all::<scim_user::Model>(sql)
.await?
.into_iter()
.map(|row| alibi_core::store::ScimProjectionCandidate {
id: row.id,
user_id: row.user_id,
})
.collect())
}
async fn acquire_scim_decommission_lease(
&self,
mut binding: alibi_core::store::ScimConnectionBinding,
) -> AuthResult<Option<alibi_core::store::ScimConnectionBinding>> {
let lease = uuid::Uuid::new_v4().simple().to_string();
let now = chrono::Utc::now().trunc_subsecs(3);
let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
sql.ident("scimConnectionBinding");
sql.push(" SET ");
sql.ident("decommissionRevision");
sql.push(" = ");
sql.ident("decommissionRevision");
sql.push(" + 1, ");
for (index, (column, value)) in [
("decommissionStatus", "reconciling".to_string()),
("decommissionLeaseId", lease.clone()),
]
.into_iter()
.enumerate()
{
if index > 0 {
sql.push(", ");
}
sql.ident(column);
sql.push(" = ");
sql.bind(value);
}
sql.push(", ");
sql.ident("decommissionedAt");
sql.push(" = ");
sql.bind(binding.decommissioned_at.unwrap_or(now));
sql.push(", ");
sql.ident("decommissionCompletedAt");
sql.push(" = NULL, ");
sql.ident("decommissionLeaseExpiresAt");
sql.push(" = ");
sql.bind(now + chrono::Duration::minutes(5));
sql.push(" WHERE ");
sql.compare("scimConnectionBinding", "id", " = ", binding.id.clone());
sql.push(" AND ");
sql.compare(
"scimConnectionBinding",
"decommissionRevision",
" = ",
binding.decommission_revision,
);
sql.push(" AND ");
sql.compare(
"scimConnectionBinding",
"decommissionStatus",
" = ",
binding.decommission_status.clone(),
);
if self.exec().execute(sql).await? == 0 {
return Ok(None);
}
binding.decommission_revision += 1;
self.get_scim_connection_binding_with_exec(self.exec(), &binding.id)
.await
}
async fn release_scim_decommission_lease(&self, id: &str, lease: &str) -> AuthResult<()> {
for _ in 0..3 {
let mut query =
model::select_model::<super::entities::scim_connection_binding::Model>(self.exec());
query.push(" WHERE ");
query.compare("scimConnectionBinding", "id", " = ", id.to_string());
model::limit_one(&mut query);
let Ok(Some(current)) = self
.exec()
.fetch_optional::<super::entities::scim_connection_binding::Model>(query)
.await
else {
break;
};
if current.decommission_status == "complete"
|| current.decommission_lease_id.as_deref() != Some(lease)
{
break;
}
let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
sql.ident("scimConnectionBinding");
sql.push(" SET ");
sql.ident("decommissionRevision");
sql.push(" = ");
sql.ident("decommissionRevision");
sql.push(" + 1, ");
sql.ident("decommissionLeaseId");
sql.push(" = NULL, ");
sql.ident("decommissionLeaseExpiresAt");
sql.push(" = NULL WHERE ");
sql.compare("scimConnectionBinding", "id", " = ", id.to_string());
sql.push(" AND ");
sql.compare(
"scimConnectionBinding",
"decommissionRevision",
" = ",
current.decommission_revision,
);
sql.push(" AND ");
sql.compare(
"scimConnectionBinding",
"decommissionLeaseId",
" = ",
lease.to_string(),
);
match self.exec().execute(sql).await {
Ok(0) => {}
_ => break,
}
}
Ok(())
}
async fn terminal_scim_connection_binding(
&self,
id: &str,
domain: Option<&str>,
key: &str,
) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
self.ensure_scim_binding(id, domain, key, true).await
}
async fn active_scim_connection_binding(
&self,
id: &str,
domain: &str,
key: &str,
) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
self.ensure_scim_binding(id, Some(domain), key, false).await
}
async fn list_scim_managed_connection_events(
&self,
id: &str,
domain: &str,
) -> AuthResult<Option<Vec<alibi_core::store::ScimManagedConnectionEvent>>> {
let exec = self.exec();
let mut sql = model::select_model::<connection::Model>(exec);
sql.push(" WHERE ");
sql.compare("scimManagedConnection", "connectionId", " = ", id);
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"provisioningDomainId",
" = ",
domain,
);
model::limit_one(&mut sql);
let Some(connection) = exec.fetch_optional::<connection::Model>(sql).await? else {
return Ok(None);
};
let mut sql = model::select_model::<event::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedConnectionEvent",
"connectionRecordId",
" = ",
connection.id,
);
sql.push(" ORDER BY ");
sql.ident("sequence");
sql.push(" DESC LIMIT 100");
let mut events = exec.fetch_all::<event::Model>(sql).await?;
events.reverse();
Ok(Some(
events
.into_iter()
.map(|event| alibi_core::store::ScimManagedConnectionEvent {
sequence: event.sequence,
kind: event.kind,
actor_id: event.actor_id,
credential_id: event.credential_id,
created_at: event.created_at,
})
.collect(),
))
}
async fn decommission_scim_managed_connection(
&self,
data: alibi_core::store::DecommissionScimManagedConnection,
complete: bool,
) -> AuthResult<Option<alibi_core::store::ScimManagedConnection>> {
self.in_transaction(true, async move |tx| {
let exec = Exec::tx(tx);
let mut sql = model::select_model::<connection::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedConnection",
"connectionId",
" = ",
data.connection_id,
);
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"provisioningDomainId",
" = ",
data.provisioning_domain_id,
);
model::limit_one(&mut sql);
if exec.engine() == crate::pool::Engine::Postgres {
sql.push(" FOR UPDATE");
}
let mut connection = exec
.fetch_optional::<connection::Model>(sql)
.await?
.ok_or_else(|| alibi_core::AuthError::Api {
status: 404,
code: None,
message: "Managed SCIM connection not found".into(),
})?;
if connection.status == "decommissioned"
|| (!complete && connection.status == "decommissioning")
{
return Ok(Some(connection.into()));
}
if complete && connection.status != "decommissioning" {
return Err(alibi_core::store::scim::scim_managed_conflict(
"Managed SCIM connection did not begin decommissioning",
));
}
let now = chrono::Utc::now().trunc_subsecs(3);
let expected = if complete {
"decommissioning"
} else {
"active"
};
let target = if complete {
"decommissioned"
} else {
"decommissioning"
};
let mut sql = Sql::with(exec.engine(), "UPDATE ");
sql.ident("scimManagedConnection");
sql.push(" SET ");
sql.ident("revision");
sql.push(" = ");
sql.ident("revision");
sql.push(" + 1, ");
sql.ident("status");
sql.push(" = ");
sql.bind(target);
sql.push(", ");
sql.ident(if complete {
"decommissionedAt"
} else {
"decommissionStartedAt"
});
sql.push(" = ");
sql.bind(now);
sql.push(", ");
sql.ident(if complete {
"decommissionedBy"
} else {
"decommissionStartedBy"
});
sql.push(" = ");
sql.bind(data.actor_id.clone());
sql.push(" WHERE ");
sql.compare("scimManagedConnection", "id", " = ", connection.id.clone());
sql.push(" AND ");
sql.compare("scimManagedConnection", "status", " = ", expected);
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"revision",
" = ",
connection.revision,
);
if exec.execute(sql).await? == 0 {
return Ok(None);
}
connection.revision += 1;
connection.status = target.into();
if complete {
connection.decommissioned_at = Some(now);
connection.decommissioned_by = Some(data.actor_id.clone());
} else {
connection.decommission_started_at = Some(now);
connection.decommission_started_by = Some(data.actor_id.clone());
}
if !complete {
let mut sql = model::select_model::<credential::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedCredential",
"connectionRecordId",
" = ",
connection.id.clone(),
);
sql.push(" AND ");
sql.compare("scimManagedCredential", "status", " = ", "active");
sql.push(" LIMIT 100");
for credential in exec.fetch_all::<credential::Model>(sql).await? {
let mut sql = Sql::with(exec.engine(), "UPDATE ");
sql.ident("scimManagedCredential");
sql.push(" SET ");
sql.ident("status");
sql.push(" = ");
sql.bind("decommissioned");
sql.push(", ");
sql.ident("activeSlotKey");
sql.push(" = ");
sql.bind(format!("{}:inactive", credential.credential_id));
sql.push(", ");
sql.ident("decommissionedAt");
sql.push(" = ");
sql.bind(now);
sql.push(" WHERE ");
sql.compare("scimManagedCredential", "id", " = ", credential.id);
sql.push(" AND ");
sql.compare("scimManagedCredential", "status", " = ", "active");
sql.push(" AND ");
sql.compare(
"scimManagedCredential",
"activeSlotKey",
" = ",
credential.active_slot_key,
);
_ = exec.execute(sql).await?;
}
}
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(
exec,
"scimManagedConnectionEvent",
"scimManagedConnectionEvent",
"id",
)
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionRecordId", connection.id.clone());
row.set(
"eventKey",
format!("{}:{}", connection.id, connection.revision),
);
row.set("sequence", connection.revision);
row.set(
"type",
if complete {
"connection.decommissioned"
} else {
"connection.decommissioning"
},
);
row.set("actorId", data.actor_id);
row.set("credentialId", None::<String>);
row.set("createdAt", now);
_ = model::insert::<event::Model>(exec, &row).await?;
Ok(Some(connection.into()))
})
.await
}
async fn revoke_scim_managed_credential(
&self,
data: alibi_core::store::RevokeScimManagedCredential,
) -> AuthResult<Option<alibi_core::store::ScimManagedConnection>> {
self.in_transaction(true, async move |tx| {
let exec = Exec::tx(tx);
let mut sql = model::select_model::<connection::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedConnection",
"connectionId",
" = ",
data.connection_id,
);
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"provisioningDomainId",
" = ",
data.provisioning_domain_id,
);
model::limit_one(&mut sql);
if exec.engine() == crate::pool::Engine::Postgres {
sql.push(" FOR UPDATE");
}
let mut connection = exec
.fetch_optional::<connection::Model>(sql)
.await?
.ok_or_else(|| alibi_core::AuthError::Api {
status: 404,
code: None,
message: "Managed SCIM connection not found".into(),
})?;
let mut sql = model::select_model::<credential::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedCredential",
"credentialId",
" = ",
data.credential_id.clone(),
);
sql.push(" AND ");
sql.compare(
"scimManagedCredential",
"connectionRecordId",
" = ",
connection.id.clone(),
);
model::limit_one(&mut sql);
let credential = exec
.fetch_optional::<credential::Model>(sql)
.await?
.ok_or_else(|| alibi_core::AuthError::Api {
status: 404,
code: None,
message: "Managed SCIM connection not found".into(),
})?;
if credential.status == "revoked" {
return Ok(Some(connection.into()));
}
if connection.status != "active" || credential.status != "active" {
return Err(alibi_core::store::scim::scim_managed_conflict(
"Managed SCIM credential is not active",
));
}
let mut sql = Sql::with(exec.engine(), "UPDATE ");
sql.ident("scimManagedConnection");
sql.push(" SET ");
sql.ident("revision");
sql.push(" = ");
sql.ident("revision");
sql.push(" + 1 WHERE ");
sql.compare("scimManagedConnection", "id", " = ", connection.id.clone());
sql.push(" AND ");
sql.compare("scimManagedConnection", "status", " = ", "active");
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"revision",
" = ",
connection.revision,
);
if exec.execute(sql).await? == 0 {
return Ok(None);
}
connection.revision += 1;
let revoked_at = chrono::Utc::now().trunc_subsecs(3);
let mut sql = Sql::with(exec.engine(), "UPDATE ");
sql.ident("scimManagedCredential");
sql.push(" SET ");
for (index, (column, value)) in [
("status", "revoked".to_string()),
(
"activeSlotKey",
format!("{}:inactive", credential.credential_id),
),
("revokedBy", data.actor_id.clone()),
]
.into_iter()
.enumerate()
{
if index > 0 {
sql.push(", ");
}
sql.ident(column);
sql.push(" = ");
sql.bind(value);
}
sql.push(", ");
sql.ident("revokedAt");
sql.push(" = ");
sql.bind(revoked_at);
sql.push(" WHERE ");
sql.compare("scimManagedCredential", "id", " = ", credential.id);
sql.push(" AND ");
sql.compare("scimManagedCredential", "status", " = ", "active");
sql.push(" AND ");
sql.compare(
"scimManagedCredential",
"activeSlotKey",
" = ",
credential.active_slot_key,
);
if exec.execute(sql).await? == 0 {
return Ok(None);
}
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(
exec,
"scimManagedConnectionEvent",
"scimManagedConnectionEvent",
"id",
)
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionRecordId", connection.id.clone());
row.set(
"eventKey",
format!("{}:{}", connection.id, connection.revision),
);
row.set("sequence", connection.revision);
row.set("type", "credential.revoked");
row.set("actorId", data.actor_id);
row.set("credentialId", Some(data.credential_id));
row.set("createdAt", revoked_at);
_ = model::insert::<event::Model>(exec, &row).await?;
Ok(Some(connection.into()))
})
.await
}
async fn rotate_scim_managed_credential(
&self,
data: alibi_core::store::RotateScimManagedCredential,
) -> AuthResult<Option<ScimManagedCreation>> {
self.in_transaction(true, async move |tx| {
let exec = Exec::tx(tx);
let mut sql = model::select_model::<connection::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedConnection",
"connectionId",
" = ",
data.connection_id,
);
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"provisioningDomainId",
" = ",
data.provisioning_domain_id,
);
model::limit_one(&mut sql);
if exec.engine() == crate::pool::Engine::Postgres {
sql.push(" FOR UPDATE");
}
let mut connection = exec
.fetch_optional::<connection::Model>(sql)
.await?
.ok_or_else(|| alibi_core::AuthError::Api {
status: 404,
code: None,
message: "Managed SCIM connection not found".into(),
})?;
if connection.status != "active" {
return Err(alibi_core::store::scim::scim_managed_conflict(
"Managed SCIM connection is not active",
));
}
let mut sql = Sql::with(exec.engine(), "UPDATE ");
sql.ident("scimManagedConnection");
sql.push(" SET ");
sql.ident("revision");
sql.push(" = ");
sql.ident("revision");
sql.push(" + 1 WHERE ");
sql.compare("scimManagedConnection", "id", " = ", connection.id.clone());
sql.push(" AND ");
sql.compare("scimManagedConnection", "status", " = ", "active");
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"revision",
" = ",
connection.revision,
);
if exec.execute(sql).await? == 0 {
return Ok(None);
}
connection.revision += 1;
let mut sql = model::select_model::<credential::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedCredential",
"connectionRecordId",
" = ",
connection.id.clone(),
);
sql.push(" AND ");
sql.compare("scimManagedCredential", "status", " = ", "active");
sql.push(" LIMIT 101");
let active = exec.fetch_all::<credential::Model>(sql).await?;
let mut slots = std::collections::HashSet::new();
for credential in active {
if credential.expires_at <= data.created_at {
let mut row = ActiveRow::new();
row.set("id", credential.id);
row.set("status", "expired");
row.set(
"activeSlotKey",
format!("{}:inactive", credential.credential_id),
);
_ = model::update::<credential::Model>(exec, &row).await?;
} else {
_ = slots.insert(credential.active_slot_key);
}
}
if slots.len() >= data.max_active_credentials {
return Err(alibi_core::store::scim::scim_managed_conflict(
"Managed SCIM connection has the maximum number of active credentials",
));
}
let slot = (0..data.max_active_credentials)
.find(|slot| !slots.contains(&format!("{}:active:{slot}", connection.id)))
.ok_or_else(|| {
alibi_core::store::scim::scim_managed_conflict(
"Managed SCIM connection has the maximum number of active credentials",
)
})?;
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(exec, "scimManagedCredential", "scimManagedCredential", "id")
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionRecordId", connection.id.clone());
row.set("credentialId", data.credential_id.clone());
row.set("tokenDigest", data.token_digest);
row.set("hashVersion", "v1");
row.set("activeSlotKey", format!("{}:active:{slot}", connection.id));
row.set("status", "active");
row.set("serializedScopes", data.serialized_scopes);
row.set("expiresAt", data.expires_at);
row.set("createdAt", data.created_at);
row.set("createdBy", data.actor_id.clone());
let credential = model::insert::<credential::Model>(exec, &row).await?;
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(
exec,
"scimManagedConnectionEvent",
"scimManagedConnectionEvent",
"id",
)
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionRecordId", connection.id.clone());
row.set(
"eventKey",
format!("{}:{}", connection.id, connection.revision),
);
row.set("sequence", connection.revision);
row.set("type", "credential.rotated");
row.set("actorId", data.actor_id);
row.set("credentialId", Some(data.credential_id));
row.set("createdAt", data.created_at);
_ = model::insert::<event::Model>(exec, &row).await?;
Ok(Some(ScimManagedCreation {
connection: connection.into(),
credential: credential.into(),
}))
})
.await
}
async fn get_scim_managed_connection(
&self,
id: &str,
domain: &str,
) -> AuthResult<Option<alibi_core::store::ScimManagedConnectionState>> {
let exec = self.exec();
let mut sql = model::select_model::<connection::Model>(exec);
sql.push(" WHERE ");
sql.compare("scimManagedConnection", "connectionId", " = ", id);
sql.push(" AND ");
sql.compare(
"scimManagedConnection",
"provisioningDomainId",
" = ",
domain,
);
model::limit_one(&mut sql);
let Some(connection) = exec.fetch_optional::<connection::Model>(sql).await? else {
return Ok(None);
};
let mut sql = model::select_model::<credential::Model>(exec);
sql.push(" WHERE ");
sql.compare(
"scimManagedCredential",
"connectionRecordId",
" = ",
connection.id.clone(),
);
sql.push(" ORDER BY ");
sql.ident("createdAt");
sql.push(" DESC LIMIT ");
sql.bind(self.find_many_limit());
let credentials = exec
.fetch_all::<credential::Model>(sql)
.await?
.into_iter()
.map(Into::into)
.collect();
Ok(Some(alibi_core::store::ScimManagedConnectionState {
connection: connection.into(),
credentials,
}))
}
async fn list_scim_managed_connections(
&self,
domain: &str,
) -> AuthResult<Vec<alibi_core::store::ScimManagedConnection>> {
let mut sql = model::select_model::<connection::Model>(self.exec());
sql.push(" WHERE ");
sql.compare(
"scimManagedConnection",
"provisioningDomainId",
" = ",
domain,
);
sql.push(" ORDER BY ");
sql.ident("createdAt");
sql.push(" DESC LIMIT ");
sql.bind(self.find_many_limit());
Ok(self
.exec()
.fetch_all::<connection::Model>(sql)
.await?
.into_iter()
.map(Into::into)
.collect())
}
async fn create_scim_managed_connection(
&self,
data: CreateScimManagedConnection,
) -> AuthResult<ScimManagedCreation> {
let request_id = data.creation_request_id.clone();
let result = self
.in_transaction(true, async move |tx| {
let exec = Exec::tx(tx);
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(exec, "scimManagedConnection", "scimManagedConnection", "id")
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("creationRequestId", data.creation_request_id);
row.set("connectionId", data.connection_id);
row.set("provisioningDomainId", data.provisioning_domain_id);
row.set("status", "active");
row.set("revision", 2_i64);
row.set("createdAt", data.created_at);
row.set("createdBy", data.actor_id.clone());
let connection = model::insert::<connection::Model>(exec, &row).await?;
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(exec, "scimManagedCredential", "scimManagedCredential", "id")
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionRecordId", connection.id.clone());
row.set("credentialId", data.credential_id.clone());
row.set("tokenDigest", data.token_digest);
row.set("hashVersion", "v1");
row.set("activeSlotKey", format!("{}:active:0", connection.id));
row.set("status", "active");
row.set("serializedScopes", data.serialized_scopes);
row.set("expiresAt", data.expires_at);
row.set("createdAt", data.created_at);
row.set("createdBy", data.actor_id.clone());
let credential = model::insert::<credential::Model>(exec, &row).await?;
for (sequence, kind) in
[(1_i64, "connection.created"), (2_i64, "credential.issued")]
{
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(
exec,
"scimManagedConnectionEvent",
"scimManagedConnectionEvent",
"id",
)
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionRecordId", connection.id.clone());
row.set("eventKey", format!("{}:{sequence}", connection.id));
row.set("sequence", sequence);
row.set("type", kind);
row.set("actorId", data.actor_id.clone());
row.set(
"credentialId",
if sequence == 2 {
Some(data.credential_id.clone())
} else {
None
},
);
row.set("createdAt", data.created_at);
_ = model::insert::<event::Model>(exec, &row).await?;
}
Ok(ScimManagedCreation {
connection: connection.into(),
credential: credential.into(),
})
})
.await;
if result.is_err() {
let mut sql = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
sql.ident("scimManagedConnection");
sql.push(" WHERE ");
sql.compare(
"scimManagedConnection",
"creationRequestId",
" = ",
request_id,
);
if self.exec().fetch_scalar::<i64>(sql).await?.unwrap_or(0) > 0 {
return Err(alibi_core::store::scim::scim_creation_conflict());
}
}
result
}
}
impl<S: AuthSchema> SqlxStore<S> {
async fn ensure_scim_binding(
&self,
id: &str,
domain: Option<&str>,
key: &str,
terminal: bool,
) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
use super::entities::scim_connection_binding as binding;
let find = async || {
let mut sql = model::select_model::<binding::Model>(self.exec());
sql.push(" WHERE ");
sql.compare("scimConnectionBinding", "connectionKey", " = ", key);
model::limit_one(&mut sql);
self.exec().fetch_optional::<binding::Model>(sql).await
};
if let Some(row) = find().await? {
let binding = row.into();
if terminal {
alibi_core::store::scim::validate_scim_binding_domain(&binding, domain)?;
}
return Ok(binding);
}
let domain = domain.ok_or_else(|| {
alibi_core::AuthError::internal(format!(
"SCIM connection \"{id}\" has no persisted binding."
))
})?;
let now = chrono::Utc::now().trunc_subsecs(3);
let mut row = ActiveRow::new();
row.set(
"id",
self.generated_id(
self.exec(),
"scimConnectionBinding",
"scimConnectionBinding",
"id",
)
.await?
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
);
row.set("connectionId", id);
row.set("connectionKey", key);
row.set("provisioningDomainId", domain);
row.set("createdAt", now);
row.set(
"decommissionStatus",
if terminal { "complete" } else { "active" },
);
if terminal {
row.set("decommissionedAt", now);
row.set("decommissionCompletedAt", now);
}
row.set("decommissionReconciledUserCount", 0_i64);
row.set("decommissionBatchCount", 0_i64);
row.set("decommissionRevision", 0_i64);
match model::insert::<binding::Model>(self.exec(), &row).await {
Ok(row) => Ok(row.into()),
Err(error) => {
if let Some(row) = find().await? {
let binding = row.into();
if terminal {
alibi_core::store::scim::validate_scim_binding_domain(
&binding,
Some(domain),
)?;
}
Ok(binding)
} else {
Err(error)
}
}
}
}
}