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|model.*(?:does not exist|was not found in the schema)",
103        )
104        .ok()
105    });
106    PATTERN
107        .as_ref()
108        .is_some_and(|pattern| pattern.is_match(&error.to_string()))
109}
110
111/// Missing tables are expected before migrations and stay quiet; any other
112/// startup seed failure is an error worth seeing.
113pub(super) fn log_startup_seed_failure(error: &AuthError) {
114    if missing_table(error) {
115        tracing::debug!(
116            "oauth-provider: oauthResource table not yet created; deferring resource seed to first access."
117        );
118    } else {
119        tracing::error!(
120            %error,
121            "oauth-provider: could not seed `resources` at startup; retrying on first resource access."
122        );
123    }
124}
125
126fn seed_row(input: &serde_json::Map<String, Value>, identifier: &str) -> Value {
127    let mut row = serde_json::Map::new();
128    for &field in FIELDS {
129        let default = match field {
130            "name" => json!(identifier),
131            "disabled" | "dpopBoundAccessTokensRequired" => json!(false),
132            _ => Value::Null,
133        };
134        _ = row.insert(
135            field.into(),
136            input
137                .get(field)
138                .filter(|value| !value.is_null())
139                .cloned()
140                .unwrap_or(default),
141        );
142    }
143    _ = row.insert("policyVersion".into(), json!(1));
144    Value::Object(row)
145}
146
147impl OAuthProviderPlugin {
148    pub(super) async fn seed_resources<S: AuthSchema>(
149        &self,
150        store: &dyn AuthStore<S>,
151    ) -> AuthResult<()> {
152        for raw in &self.config.resources {
153            let mut input = if let Some(identifier) = raw.as_str() {
154                serde_json::Map::from_iter([("identifier".into(), json!(identifier))])
155            } else {
156                raw.as_object().cloned().ok_or_else(|| {
157                    AuthError::config("Resource seed must be an identifier or policy object")
158                })?
159            };
160            let identifier = input
161                .get("identifier")
162                .and_then(Value::as_str)
163                .ok_or_else(|| AuthError::config("Resource seed requires an identifier"))?
164                .to_owned();
165            if self.identifier_error(&identifier).await?.is_some() {
166                tracing::warn!(%identifier, "Skipping invalid configured OAuth resource");
167                continue;
168            }
169            if input.get("signingAlgorithm").is_some_and(|value| {
170                !value.is_null()
171                    && !value
172                        .as_str()
173                        .is_some_and(|algorithm| super::resources::ALGORITHMS.contains(&algorithm))
174            }) {
175                tracing::warn!(%identifier, "Dropping unsupported configured resource signingAlgorithm");
176                _ = input.remove("signingAlgorithm");
177            }
178            // Storage errors, including a missing table, propagate so the lazy
179            // seed resets and the next resource access retries.
180            let mut existing = store.get_oauth_resource(&identifier).await?;
181            let now =
182                chrono::DateTime::from_timestamp_millis(chrono::Utc::now().timestamp_millis())
183                    .ok_or_else(|| AuthError::internal("Invalid resource seed timestamp"))?;
184            if existing.is_none() {
185                match store
186                    .create_oauth_resource(alibi_core::store::CreateOAuthResource {
187                        identifier: identifier.clone(),
188                        metadata: seed_row(&input, &identifier),
189                        created_at: now,
190                    })
191                    .await
192                {
193                    Ok(_) => continue,
194                    // A concurrent boot inserted the row between the lookup and
195                    // the insert. Re-read it so merge and overwrite still apply;
196                    // without a row, the insert error is the one to report.
197                    Err(error) => {
198                        existing = store.get_oauth_resource(&identifier).await.ok().flatten();
199                        if existing.is_none() {
200                            return Err(error);
201                        }
202                    }
203                }
204            }
205            if let Some(existing) = existing {
206                if self.config.resource_seed_mode == OAuthResourceSeedMode::InsertOnly {
207                    continue;
208                }
209                let row = if self.config.resource_seed_mode == OAuthResourceSeedMode::Overwrite {
210                    seed_row(&input, &identifier)
211                } else {
212                    let mut row =
213                        existing.metadata.as_object().cloned().ok_or_else(|| {
214                            AuthError::internal("Invalid persisted resource policy")
215                        })?;
216                    for &field in FIELDS {
217                        if let Some(value) = input.get(field) {
218                            _ = row.insert(field.into(), value.clone());
219                        }
220                    }
221                    Value::Object(row)
222                };
223                _ = store.update_oauth_resource(&identifier, row, now).await?;
224            }
225        }
226        Ok(())
227    }
228
229    pub(super) async fn get_resource<S: AuthSchema>(
230        &self,
231        ctx: &AuthContext<S>,
232        identifier: &str,
233    ) -> AuthResult<Option<alibi_core::store::OAuthResource>> {
234        // Init deliberately does not complete this state. A real first lookup
235        // re-applies merge/overwrite, including edits made since initialization.
236        if let Some(state) = ctx.extensions.get::<SeedState>() {
237            let work = {
238                let mut lifecycle = state.0.lock().await;
239                if lifecycle.completed {
240                    None
241                } else {
242                    let future = lifecycle
243                        .pending
244                        .get_or_insert_with(|| {
245                            let provider = self.clone();
246                            let store = ctx.database.clone();
247                            async move {
248                                provider
249                                    .seed_resources(store.as_ref())
250                                    .await
251                                    .map_err(|error| SeedFailure::capture(&error))
252                            }
253                            .boxed()
254                            .shared()
255                        })
256                        .clone();
257                    Some((lifecycle.attempt, future))
258                }
259            };
260            if let Some((attempt, future)) = work {
261                let result = future.await;
262                let mut lifecycle = state.0.lock().await;
263                if lifecycle.attempt == attempt {
264                    lifecycle.pending = None;
265                    lifecycle.completed = result.is_ok();
266                    lifecycle.attempt = lifecycle.attempt.wrapping_add(1);
267                }
268                result.map_err(SeedFailure::into_error)?;
269            }
270        }
271        super::resource_cache::get(
272            ctx,
273            identifier,
274            self.config.cached_resources.contains(identifier),
275        )
276        .await
277    }
278}
279
280/// Restart policy for configured resource rows. The pinned replacement mode is `overwrite`.
281#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
282#[serde(rename_all = "camelCase")]
283pub enum OAuthResourceSeedMode {
284    #[default]
285    InsertOnly,
286    Merge,
287    Overwrite,
288}
289
290/// Application identifier validation replaces the default absolute-URI/fragment check.
291#[async_trait::async_trait]
292pub trait OAuthResourceIdentifierValidator: Send + Sync + std::fmt::Debug {
293    async fn validate(&self, identifier: &str) -> AuthResult<bool>;
294}
alibi_plugins/oauth_provider/resource_seed.rs

alibi_plugins/oauth_provider/
resource_seed.rs
