alibi_plugins/scim/
projection.rs

1use super::ScimPlugin;
2use alibi_core::endpoint::{EndpointCall, EndpointDefinition, EndpointInput, EndpointResponse};
3use alibi_core::store::AuthTransaction;
4pub use alibi_core::store::{
5    ScimAuthorizationSource, ScimIdentitySource, ScimIdentityState, ScimProjectedUserState,
6    ScimRoleExistenceInput, ScimRoleMappingInput,
7};
8use alibi_core::{
9    AuthContext, AuthInitContext, AuthPlugin, AuthRequest, AuthResponse, AuthResult, AuthRoute,
10    AuthSchema,
11};
12use async_trait::async_trait;
13use std::sync::Arc;
14/// Application access reconciliation inside the actual provisioning transaction.
15#[async_trait]
16pub trait ScimProjection<S: AuthSchema>: Send + Sync {
17    async fn reconcile_user(
18        &self,
19        input: &ScimProjectedUserState,
20        database: &dyn AuthTransaction<S>,
21    ) -> AuthResult<()>;
22}
23#[async_trait]
24pub trait ScimRoleProjection<S: AuthSchema>: Send + Sync {
25    async fn map(
26        &self,
27        input: &ScimRoleMappingInput,
28        database: &dyn AuthTransaction<S>,
29    ) -> AuthResult<Option<Vec<String>>>;
30    async fn exists(
31        &self,
32        input: &ScimRoleExistenceInput,
33        database: &dyn AuthTransaction<S>,
34    ) -> AuthResult<bool>;
35}
36/// The application chooses whether provisioning creates or links an authentication User.
37#[derive(Clone)]
38pub enum ScimIdentityResolution {
39    Create,
40    Link {
41        user_id: String,
42        manage_profile: bool,
43    },
44}
45/// Canonical provider attributes supplied to the application identity resolver.
46pub struct ScimIdentityResolutionInput {
47    pub connection_id: String,
48    pub provisioning_domain_id: String,
49    pub resource: serde_json::Value,
50}
51#[async_trait]
52pub trait ScimIdentity<S: AuthSchema>: Send + Sync {
53    /// Resolve a new provisioning source before its mutation transaction begins.
54    async fn resolve_user(
55        &self,
56        _input: &ScimIdentityResolutionInput,
57        _database: &dyn alibi_core::AuthStore<S>,
58    ) -> AuthResult<ScimIdentityResolution> {
59        Ok(ScimIdentityResolution::Create)
60    }
61
62    async fn reconcile_user(
63        &self,
64        input: &ScimIdentityState,
65        database: &dyn AuthTransaction<S>,
66    ) -> AuthResult<()>;
67}
68pub(super) struct ScimIdentityHandle<S: AuthSchema>(pub Arc<dyn ScimIdentity<S>>);
69pub(super) struct ScimProjectionHandle<S: AuthSchema>(
70    pub Arc<dyn ScimProjection<S>>,
71    pub Option<Arc<dyn ScimRoleProjection<S>>>,
72);
73
74/// A SCIM plugin carrying application-owned transactional projection callbacks.
75pub struct ConfiguredScimPlugin<S: AuthSchema> {
76    base: ScimPlugin,
77    projection: Option<Arc<dyn ScimProjection<S>>>,
78    identity: Option<Arc<dyn ScimIdentity<S>>>,
79    roles: Option<Arc<dyn ScimRoleProjection<S>>>,
80    verifier: Option<Arc<dyn super::ScimBearerTokenVerifier<S>>>,
81}
82impl ScimPlugin {
83    #[must_use]
84    pub fn projection<S: AuthSchema>(
85        self,
86        projection: Arc<dyn ScimProjection<S>>,
87    ) -> ConfiguredScimPlugin<S> {
88        ConfiguredScimPlugin {
89            base: self,
90            projection: Some(projection),
91            identity: None,
92            roles: None,
93            verifier: None,
94        }
95    }
96}
97impl ScimPlugin {
98    #[must_use]
99    pub fn identity<S: AuthSchema>(
100        self,
101        identity: Arc<dyn ScimIdentity<S>>,
102    ) -> ConfiguredScimPlugin<S> {
103        ConfiguredScimPlugin {
104            base: self,
105            projection: None,
106            identity: Some(identity),
107            roles: None,
108            verifier: None,
109        }
110    }
111}
112impl ScimPlugin {
113    #[must_use]
114    pub fn authentication<S: AuthSchema>(
115        mut self,
116        verifier: Arc<dyn super::ScimBearerTokenVerifier<S>>,
117    ) -> ConfiguredScimPlugin<S> {
118        self.has_token_verifier = true;
119        ConfiguredScimPlugin {
120            base: self,
121            projection: None,
122            identity: None,
123            roles: None,
124            verifier: Some(verifier),
125        }
126    }
127}
128impl<S: AuthSchema> ConfiguredScimPlugin<S> {
129    #[must_use]
130    pub fn authentication(mut self, verifier: Arc<dyn super::ScimBearerTokenVerifier<S>>) -> Self {
131        self.base.has_token_verifier = true;
132        self.verifier = Some(verifier);
133        self
134    }
135    #[must_use]
136    pub fn identity(mut self, identity: Arc<dyn ScimIdentity<S>>) -> Self {
137        self.identity = Some(identity);
138        self
139    }
140
141    #[must_use]
142    pub fn roles(mut self, roles: Arc<dyn ScimRoleProjection<S>>) -> Self {
143        self.roles = Some(roles);
144        self
145    }
146}
147#[async_trait]
148impl<S: AuthSchema> AuthPlugin<S> for ConfiguredScimPlugin<S> {
149    route_openapi_metadata!(S);
150    fn name(&self) -> &'static str {
151        "scim"
152    }
153    fn allowed_media_types(&self, route: &AuthRoute) -> Vec<&'static str> {
154        <ScimPlugin as AuthPlugin<S>>::allowed_media_types(&self.base, route)
155    }
156    fn routes(&self) -> Vec<AuthRoute> {
157        <ScimPlugin as AuthPlugin<S>>::routes(&self.base)
158    }
159    fn server_endpoints(&self) -> Vec<EndpointDefinition> {
160        <ScimPlugin as AuthPlugin<S>>::server_endpoints(&self.base)
161    }
162    fn validate_endpoint(
163        &self,
164        call: &EndpointCall,
165        ctx: &AuthContext<S>,
166    ) -> AuthResult<EndpointInput> {
167        self.base.validate_endpoint(call, ctx)
168    }
169    async fn on_init(&self, ctx: &mut AuthInitContext<S>) -> AuthResult<()> {
170        self.base.on_init(ctx).await?;
171        if let Some(verifier) = &self.verifier {
172            ctx.extensions
173                .insert(super::authentication::ScimBearerVerifierHandle(
174                    verifier.clone(),
175                ));
176        }
177        if let Some(projection) = &self.projection {
178            ctx.extensions
179                .insert(ScimProjectionHandle(projection.clone(), self.roles.clone()));
180        }
181        if let Some(identity) = &self.identity {
182            ctx.extensions.insert(ScimIdentityHandle(identity.clone()));
183        }
184        Ok(())
185    }
186    async fn on_endpoint(
187        &self,
188        call: &EndpointCall,
189        ctx: &AuthContext<S>,
190    ) -> AuthResult<EndpointResponse> {
191        self.base.on_endpoint(call, ctx).await
192    }
193    async fn on_http_request_action(
194        &self,
195        req: &AuthRequest,
196        ctx: &AuthContext<S>,
197    ) -> AuthResult<Option<alibi_core::HttpRequestAction>> {
198        self.base.on_http_request_action(req, ctx).await
199    }
200    async fn on_request(
201        &self,
202        req: &AuthRequest,
203        ctx: &AuthContext<S>,
204    ) -> AuthResult<Option<AuthResponse>> {
205        self.base.on_request(req, ctx).await
206    }
207}
208
209impl ScimPlugin {
210    pub(super) async fn reconcile_projection<S: AuthSchema>(
211        &self,
212        call: &EndpointCall,
213        ctx: &AuthContext<S>,
214    ) -> AuthResult<EndpointResponse> {
215        use alibi_core::{AuthError, store::transaction, utils::json::JsValue};
216        let projection=ctx.extensions.get::<ScimProjectionHandle<S>>().ok_or_else(||AuthError::config("SCIM projection reconciliation requires projection.reconcileUser to be configured."))?;
217        let domain = call
218            .body()
219            .and_then(|body| body.get("provisioningDomainId"))
220            .and_then(JsValue::as_str)
221            .ok_or_else(|| AuthError::internal("Validated domain missing"))?
222            .to_owned();
223        let mut cursor = None;
224        let mut reconciled = 0;
225        let mut batches = 0;
226        loop {
227            let candidates = ctx
228                .database
229                .scim_projection_candidates(&domain, cursor.as_deref())
230                .await?;
231            if candidates.is_empty() {
232                break;
233            }
234            let more = candidates.len() > 50;
235            let mut by_user = indexmap::IndexMap::new();
236            for candidate in candidates.into_iter().take(50) {
237                cursor = Some(candidate.user_id.clone());
238                _ = by_user.insert(candidate.user_id, candidate.id);
239            }
240            let count = by_user.len();
241            let ids = by_user.into_values().collect::<Vec<_>>();
242            let callback = projection.0.clone();
243            let roles = projection.1.clone();
244            let batch_domain = domain.clone();
245            let result = transaction::<S, (), _>(ctx.database.as_ref(), move |database| {
246                Box::pin(async move {
247                    let states = database
248                        .prepare_scim_projection(&batch_domain, &ids, roles.is_some())
249                        .await?;
250                    for mut prepared in states {
251                        if let Some(roles) = &roles {
252                            apply_roles(database, roles.as_ref(), &mut prepared).await?;
253                        }
254
255                        database
256                            .remove_scim_projection_grants(
257                                &batch_domain,
258                                &prepared.state.user_id,
259                                &prepared.removed_grant_ids,
260                            )
261                            .await?;
262                        callback
263                            .reconcile_user(&prepared.state, database)
264                            .await
265                            .map_err(|error| {
266                                if alibi_core::endpoint::is_endpoint_api_error(&error) {
267                                    error
268                                } else {
269                                    AuthError::CallbackFailure(Box::new(error))
270                                }
271                            })?;
272                    }
273                    Ok(())
274                })
275            })
276            .await;
277            match result {
278                Ok(()) => {}
279                Err(AuthError::CallbackFailure(cause)) => {
280                    let detail = match cause.as_ref() {
281                        AuthError::Api { message, .. }
282                            if matches!(
283                                message.as_str(),
284                                "SCIM role mapping failed"
285                                    | "SCIM role validation failed"
286                                    | "SCIM identity reconciliation failed"
287                            ) =>
288                        {
289                            message.as_str()
290                        }
291                        _ => "SCIM projection reconciliation failed",
292                    };
293                    return Ok(EndpointResponse::error(AuthError::Api{status:500,code:None,message:detail.into()}).with_error_body(JsValue::from(serde_json::json!({"schemas":["urn:ietf:params:scim:api:messages:2.0:Error"],"status":"500","detail":detail}))));
294                }
295                Err(error) => return Err(error),
296            }
297            batches += 1;
298            reconciled += count;
299            if !more {
300                break;
301            }
302        }
303        EndpointResponse::json(
304            &serde_json::json!({"provisioningDomainId":domain,"reconciledUsers":reconciled,"batches":batches}),
305        )
306    }
307}
308
309pub(super) async fn apply_roles<S: AuthSchema>(
310    database: &dyn AuthTransaction<S>,
311    roles: &dyn ScimRoleProjection<S>,
312    prepared: &mut alibi_core::store::ScimProjectionPreparedUser,
313) -> AuthResult<()> {
314    use alibi_core::store::CreateScimProjectionGrant;
315    use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
316    use chrono::SubsecRound;
317    use sha2::{Digest, Sha256};
318    let mut desired = indexmap::IndexMap::new();
319    let mut existence = std::collections::HashMap::new();
320    for input in &prepared.role_sources {
321        let mapped = roles
322            .map(input, database)
323            .await
324            .map_err(|error| application_callback_error(error, "SCIM role mapping failed"))?
325            .unwrap_or_default();
326        let mut unique = std::collections::HashSet::new();
327        for candidate in mapped {
328            let role = alibi_core::utils::javascript::trim(&candidate);
329            if role.is_empty() || !unique.insert(role.to_string()) {
330                continue;
331            }
332            let existence_key = (
333                input.connection_id.clone(),
334                input.provisioning_domain_id.clone(),
335                role.to_string(),
336            );
337            let exists = if let Some(exists) = existence.get(&existence_key) {
338                *exists
339            } else {
340                let exists = roles
341                    .exists(
342                        &ScimRoleExistenceInput {
343                            connection_id: input.connection_id.clone(),
344                            provisioning_domain_id: input.provisioning_domain_id.clone(),
345                            role: role.into(),
346                        },
347                        database,
348                    )
349                    .await
350                    .map_err(|error| {
351                        application_callback_error(error, "SCIM role validation failed")
352                    })?;
353                _ = existence.insert(existence_key, exists);
354                exists
355            };
356            if !exists {
357                continue;
358            }
359            let ScimAuthorizationSource::Group {
360                id,
361                external_id,
362                display_name,
363            } = &input.source;
364            let key = URL_SAFE_NO_PAD.encode(Sha256::digest(serde_json::to_vec(&[
365                "scim-projection-grant",
366                &input.connection_id,
367                &input.scim_user_id,
368                "group",
369                id,
370                role,
371            ])?));
372            _ = desired.insert(
373                key.clone(),
374                (
375                    CreateScimProjectionGrant {
376                        connection_id: input.connection_id.clone(),
377                        provisioning_domain_id: input.provisioning_domain_id.clone(),
378                        scim_user_id: input.scim_user_id.clone(),
379                        user_id: input.user_id.clone(),
380                        source_id: id.clone(),
381                        source_value: external_id.clone().unwrap_or_else(|| display_name.clone()),
382                        role: role.into(),
383                        grant_key: key,
384                        created_at: chrono::Utc::now().trunc_subsecs(3),
385                    },
386                    serde_json::json!({"source":input.source,"role":role}),
387                ),
388            );
389        }
390    }
391    prepared.removed_grant_ids = prepared
392        .existing_grants
393        .iter()
394        .filter(|(_, key)| !desired.contains_key(key))
395        .map(|(id, _)| id.clone())
396        .collect();
397    database
398        .remove_scim_projection_grants(
399            &prepared.state.provisioning_domain_id,
400            &prepared.state.user_id,
401            &prepared.removed_grant_ids,
402        )
403        .await?;
404    prepared.removed_grant_ids.clear();
405    let now = chrono::Utc::now().trunc_subsecs(3);
406    for (key, (data, _)) in &desired {
407        if !prepared
408            .existing_grants
409            .iter()
410            .any(|(_, existing)| existing == key)
411        {
412            let mut data = data.clone();
413            data.created_at = now;
414            database.create_scim_projection_grant(data).await?;
415        }
416    }
417    let mut ordered = desired.into_iter().collect::<Vec<_>>();
418    // The reference orders base64url grant keys with String.localeCompare,
419    // whose default collation puts lowercase letters before later uppercase
420    // letters by alphabetic weight rather than by their ASCII byte values.
421    let collator = icu_collator::Collator::try_new(
422        icu_collator::CollatorPreferences::default(),
423        icu_collator::options::CollatorOptions::default(),
424    )
425    .map_err(|error| alibi_core::AuthError::internal(error.to_string()))?;
426    ordered.sort_by(|left, right| collator.compare(&left.0, &right.0));
427    prepared.state.grants = ordered.into_iter().map(|(_, (_, state))| state).collect();
428    Ok(())
429}
430pub(super) fn application_callback_error(
431    error: alibi_core::AuthError,
432    detail: &str,
433) -> alibi_core::AuthError {
434    if alibi_core::endpoint::is_endpoint_api_error(&error) {
435        error
436    } else {
437        tracing::error!(error=%error,detail,"SCIM application callback failed");
438        alibi_core::AuthError::CallbackFailure(Box::new(alibi_core::AuthError::Api {
439            status: 500,
440            code: None,
441            message: detail.into(),
442        }))
443    }
444}
445impl ScimPlugin {
446    pub(super) async fn decommission_binding<S: AuthSchema>(
447        &self,
448        ctx: &AuthContext<S>,
449        mut binding: alibi_core::store::ScimConnectionBinding,
450    ) -> AuthResult<alibi_core::store::ScimConnectionBinding> {
451        use alibi_core::{AuthError, store::transaction};
452        use chrono::SubsecRound;
453        let active_lease = |binding: &alibi_core::store::ScimConnectionBinding| {
454            binding.decommission_status == "reconciling"
455                && binding
456                    .decommission_lease_id
457                    .as_ref()
458                    .is_some_and(|lease| !lease.is_empty())
459                && binding
460                    .decommission_lease_expires_at
461                    .is_some_and(|expiry| expiry > chrono::Utc::now().trunc_subsecs(3))
462        };
463        if binding.decommission_status == "complete" || active_lease(&binding) {
464            return Ok(binding);
465        }
466        let mut acquired = None;
467        for _ in 0..10 {
468            if let Some(leased) = ctx
469                .database
470                .acquire_scim_decommission_lease(binding.clone())
471                .await?
472            {
473                acquired = Some(leased);
474                break;
475            }
476            binding = ctx
477                .database
478                .terminal_scim_connection_binding(
479                    &binding.connection_id,
480                    Some(&binding.provisioning_domain_id),
481                    &binding.connection_key,
482                )
483                .await?;
484            if binding.decommission_status == "complete" || active_lease(&binding) {
485                return Ok(binding);
486            }
487        }
488        let binding=acquired.ok_or_else(||AuthError::internal(format!("SCIM connection \"{}\" changed repeatedly while decommissioning; retry the operation.",binding.connection_id)))?;
489        let lease = binding
490            .decommission_lease_id
491            .clone()
492            .ok_or_else(|| AuthError::internal("Acquired SCIM lease missing"))?;
493        let projection = ctx.extensions.get::<ScimProjectionHandle<S>>();
494        let identity = ctx.extensions.get::<ScimIdentityHandle<S>>();
495        let result=async {
496            loop {
497                let binding_id=binding.id.clone();let connection_id=binding.connection_id.clone();let lease=lease.clone();let projection=projection.clone();let identity=identity.clone();
498                let checkpoint=transaction::<S,alibi_core::store::ScimConnectionBinding,_>(ctx.database.as_ref(),move|database|Box::pin(async move{
499                    let stored=database.get_scim_connection_binding(&binding_id).await?.ok_or_else(||AuthError::internal(format!("SCIM connection \"{connection_id}\" binding disappeared during decommissioning.")))?;
500                    if stored.decommission_status=="complete"{return Ok(stored);}
501                    if stored.decommission_lease_id.as_deref()!=Some(&lease){return Err(AuthError::internal(format!("SCIM connection \"{connection_id}\" decommission lease was taken over by another worker.")));}
502                    let candidates=database.scim_projection_candidates(&stored.provisioning_domain_id,stored.decommission_cursor_user_id.as_deref()).await?;
503                    let more=candidates.len()>50;let mut cursor=None;let mut by_user=indexmap::IndexMap::new();for candidate in candidates.into_iter().take(50){cursor=Some(candidate.user_id.clone());_=by_user.insert(candidate.user_id,candidate.id);}
504                    let user_ids=by_user.keys().cloned().collect::<Vec<_>>();let ids=by_user.into_values().collect::<Vec<_>>();
505                    let roles=projection.as_ref().and_then(|projection|projection.1.clone());
506                    let states=database.prepare_scim_projection(&stored.provisioning_domain_id,&ids,roles.is_some()).await?;
507                    if !database.renew_scim_decommission_lease(&stored,&lease).await?{return Err(AuthError::internal(format!("SCIM connection \"{connection_id}\" decommission lease changed before reconciliation.")));}
508                    for mut prepared in states {
509                        if let Some(roles)=&roles{apply_roles(database,roles.as_ref(),&mut prepared).await?;}
510                        database.remove_scim_projection_grants(&stored.provisioning_domain_id,&prepared.state.user_id,&prepared.removed_grant_ids).await?;
511                        if let Some(projection)=&projection{projection.0.reconcile_user(&prepared.state,database).await.map_err(|error|if alibi_core::endpoint::is_endpoint_api_error(&error){error}else{AuthError::CallbackFailure(Box::new(error))})?;}
512                    }
513                    for user_id in &user_ids{
514                        let state=database.reconcile_scim_identity(user_id).await?;
515                        if let Some(identity)=&identity{identity.0.reconcile_user(&state,database).await.map_err(|error|application_callback_error(error,"SCIM identity reconciliation failed"))?;}
516                        if !state.active{database.delete_user_sessions(user_id).await?;}
517                    }
518                    database.advance_scim_decommission(&stored,&lease,cursor.as_deref(),user_ids.len(),more).await?.ok_or_else(||AuthError::internal(format!("SCIM connection \"{connection_id}\" decommission checkpoint changed concurrently.")))
519                })).await?;
520                if checkpoint.decommission_status=="complete"{return Ok(checkpoint);}
521            }
522        }.await;
523        if result.is_err() {
524            let _ignored = ctx
525                .database
526                .release_scim_decommission_lease(&binding.id, &lease)
527                .await;
528        }
529        result
530    }
531    pub(super) fn scim_callback_response(cause: &alibi_core::AuthError) -> EndpointResponse {
532        use alibi_core::{AuthError, utils::json::JsValue};
533        let detail = match cause {
534            AuthError::Api { message, .. }
535                if matches!(
536                    message.as_str(),
537                    "SCIM role mapping failed"
538                        | "SCIM role validation failed"
539                        | "SCIM identity reconciliation failed"
540                ) =>
541            {
542                message.as_str()
543            }
544            _ => "SCIM projection reconciliation failed",
545        };
546        EndpointResponse::error(AuthError::Api{status:500,code:None,message:detail.into()}).with_error_body(JsValue::from(serde_json::json!({"schemas":["urn:ietf:params:scim:api:messages:2.0:Error"],"status":"500","detail":detail})))
547    }
548}
alibi_plugins/scim/projection.rs

alibi_plugins/scim/
projection.rs
