alibi_plugins/oauth_provider/
resource_seed.rs

1//! Configured resource persistence, separate from resource administration and caching.
2use super::OAuthProviderPlugin;
3use alibi_core::{AuthContext, AuthError, AuthResult, AuthSchema, AuthStore};
4use futures_util::{
5    FutureExt as _,
6    future::{BoxFuture, Shared},
7};
8use serde_json::{Value, json};
9
10const FIELDS: &[&str] = &[
11    "name",
12    "accessTokenTtl",
13    "refreshTokenTtl",
14    "signingAlgorithm",
15    "signingKeyId",
16    "allowedScopes",
17    "customClaims",
18    "dpopBoundAccessTokensRequired",
19    "disabled",
20    "metadata",
21];
22
23/// Scoped to an initialized store/context, not a plugin clone shared across instances.
24#[derive(Default)]
25pub(super) struct SeedState(tokio::sync::Mutex<SeedLifecycle>);
26
27type SeedFuture = Shared<BoxFuture<'static, Result<(), SeedFailure>>>;
28#[derive(Default)]
29struct SeedLifecycle {
30    completed: bool,
31    attempt: u64,
32    pending: Option<SeedFuture>,
33}
34
35// Shared attempts must return the same public callback failure to all waiters.
36// AuthError itself is not Clone (it can own database/serialization errors).
37#[derive(Clone)]
38enum SeedFailure {
39    Api {
40        status: u16,
41        code: Option<String>,
42        message: String,
43    },
44    RateLimited {
45        retry_after: Option<f64>,
46    },
47    Private {
48        message: String,
49        empty: bool,
50    },
51}
52impl SeedFailure {
53    fn capture(error: &AuthError) -> Self {
54        if let AuthError::RateLimited { retry_after } = error {
55            return Self::RateLimited {
56                retry_after: *retry_after,
57            };
58        }
59        if alibi_core::endpoint::is_endpoint_api_error(error) {
60            let (status, code, message) = error.error_payload();
61            return Self::Api {
62                status,
63                code,
64                message,
65            };
66        }
67        Self::Private {
68            empty: matches!(
69                &error,
70                AuthError::CallbackFailure(_) | AuthError::Encryption(_)
71            ),
72            message: error.to_string(),
73        }
74    }
75    fn into_error(self) -> AuthError {
76        match self {
77            Self::Api {
78                status,
79                code,
80                message,
81            } => AuthError::Api {
82                status,
83                code,
84                message,
85            },
86            Self::RateLimited { retry_after } => AuthError::RateLimited { retry_after },
87            Self::Private { message, empty } => {
88                let error = AuthError::internal(message);
89                if empty {
90                    AuthError::CallbackFailure(Box::new(error))
91                } else {
92                    error
93                }
94            }
95        }
96    }
97}
98
99fn missing_table(error: &AuthError) -> bool {
100    static PATTERN: std::sync::LazyLock<Option<regex::Regex>> = std::sync::LazyLock::new(|| {
101        regex::Regex::new(
102            r"(?i)no such table|relation.*does not exist|table.*does(?: not|n[''’]?t) exist",
103        )
104        .ok()
105    });
106    PATTERN
107        .as_ref()
108        .is_some_and(|pattern| pattern.is_match(&error.to_string()))
109}
110
111fn seed_row(input: &serde_json::Map<String, Value>, identifier: &str) -> Value {
112    let mut row = serde_json::Map::new();
113    for &field in FIELDS {
114        let default = match field {
115            "name" => json!(identifier),
116            "disabled" | "dpopBoundAccessTokensRequired" => json!(false),
117            _ => Value::Null,
118        };
119        _ = row.insert(
120            field.into(),
121            input
122                .get(field)
123                .filter(|value| !value.is_null())
124                .cloned()
125                .unwrap_or(default),
126        );
127    }
128    _ = row.insert("policyVersion".into(), json!(1));
129    Value::Object(row)
130}
131
132impl OAuthProviderPlugin {
133    pub(super) async fn seed_resources<S: AuthSchema>(
134        &self,
135        store: &dyn AuthStore<S>,
136    ) -> AuthResult<()> {
137        for raw in &self.config.resources {
138            let mut input = if let Some(identifier) = raw.as_str() {
139                serde_json::Map::from_iter([("identifier".into(), json!(identifier))])
140            } else {
141                raw.as_object().cloned().ok_or_else(|| {
142                    AuthError::config("Resource seed must be an identifier or policy object")
143                })?
144            };
145            let identifier = input
146                .get("identifier")
147                .and_then(Value::as_str)
148                .ok_or_else(|| AuthError::config("Resource seed requires an identifier"))?
149                .to_owned();
150            let valid = if let Some(validator) = &self.config.identifier_validator {
151                validator.validate(&identifier).await?
152            } else {
153                url::Url::parse(&identifier)
154                    .is_ok_and(|url| url.fragment().is_none_or(str::is_empty))
155            };
156            if !valid {
157                tracing::warn!(%identifier, "Skipping invalid configured OAuth resource");
158                continue;
159            }
160            if input.get("signingAlgorithm").is_some_and(|value| {
161                !value.is_null()
162                    && !value
163                        .as_str()
164                        .is_some_and(|algorithm| super::resources::ALGORITHMS.contains(&algorithm))
165            }) {
166                tracing::warn!(%identifier, "Dropping unsupported configured resource signingAlgorithm");
167                _ = input.remove("signingAlgorithm");
168            }
169            let existing = match store.get_oauth_resource(&identifier).await {
170                Ok(existing) => existing,
171                Err(error) if missing_table(&error) => return Ok(()),
172                Err(error) => return Err(error),
173            };
174            let now =
175                chrono::DateTime::from_timestamp_millis(chrono::Utc::now().timestamp_millis())
176                    .ok_or_else(|| AuthError::internal("Invalid resource seed timestamp"))?;
177            if let Some(existing) = existing {
178                if self.config.resource_seed_mode == OAuthResourceSeedMode::InsertOnly {
179                    continue;
180                }
181                let row = if self.config.resource_seed_mode == OAuthResourceSeedMode::Overwrite {
182                    seed_row(&input, &identifier)
183                } else {
184                    let mut row =
185                        existing.metadata.as_object().cloned().ok_or_else(|| {
186                            AuthError::internal("Invalid persisted resource policy")
187                        })?;
188                    for &field in FIELDS {
189                        if let Some(value) = input.get(field) {
190                            _ = row.insert(field.into(), value.clone());
191                        }
192                    }
193                    Value::Object(row)
194                };
195                _ = store.update_oauth_resource(&identifier, row, now).await?;
196            } else if let Err(error) = store
197                .create_oauth_resource(alibi_core::store::CreateOAuthResource {
198                    identifier: identifier.clone(),
199                    metadata: seed_row(&input, &identifier),
200                    created_at: now,
201                })
202                .await
203            {
204                let message = error.to_string().to_lowercase();
205                if message.contains("unique") || message.contains("duplicate") {
206                    continue;
207                }
208                if missing_table(&error) {
209                    return Ok(());
210                }
211                return Err(error);
212            }
213        }
214        Ok(())
215    }
216
217    pub(super) async fn get_resource<S: AuthSchema>(
218        &self,
219        ctx: &AuthContext<S>,
220        identifier: &str,
221    ) -> AuthResult<Option<alibi_core::store::OAuthResource>> {
222        // Init deliberately does not complete this state. A real first lookup
223        // re-applies merge/overwrite, including edits made since initialization.
224        if let Some(state) = ctx.extensions.get::<SeedState>() {
225            let work = {
226                let mut lifecycle = state.0.lock().await;
227                if lifecycle.completed {
228                    None
229                } else {
230                    let future = lifecycle
231                        .pending
232                        .get_or_insert_with(|| {
233                            let provider = self.clone();
234                            let store = ctx.database.clone();
235                            async move {
236                                provider
237                                    .seed_resources(store.as_ref())
238                                    .await
239                                    .map_err(|error| SeedFailure::capture(&error))
240                            }
241                            .boxed()
242                            .shared()
243                        })
244                        .clone();
245                    Some((lifecycle.attempt, future))
246                }
247            };
248            if let Some((attempt, future)) = work {
249                let result = future.await;
250                let mut lifecycle = state.0.lock().await;
251                if lifecycle.attempt == attempt {
252                    lifecycle.pending = None;
253                    lifecycle.completed = result.is_ok();
254                    lifecycle.attempt = lifecycle.attempt.wrapping_add(1);
255                }
256                result.map_err(SeedFailure::into_error)?;
257            }
258        }
259        super::resource_cache::get(
260            ctx,
261            identifier,
262            self.config.cached_resources.contains(identifier),
263        )
264        .await
265    }
266}
267
268/// Restart policy for configured resource rows. The pinned replacement mode is `overwrite`.
269#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
270#[serde(rename_all = "camelCase")]
271pub enum OAuthResourceSeedMode {
272    #[default]
273    InsertOnly,
274    Merge,
275    Overwrite,
276}
277
278/// Application identifier validation replaces the default absolute-URI/fragment check.
279#[async_trait::async_trait]
280pub trait OAuthResourceIdentifierValidator: Send + Sync + std::fmt::Debug {
281    async fn validate(&self, identifier: &str) -> AuthResult<bool>;
282}
alibi_plugins/oauth_provider/resource_seed.rs

alibi_plugins/oauth_provider/
resource_seed.rs
