Skip to main content

max / makenotwork

10.9 KB · 334 lines History Blame Raw
1 //! OAuth 2.0 authorization code storage and retrieval.
2
3 use chrono::{DateTime, Utc};
4 use sqlx::PgPool;
5 use uuid::Uuid;
6
7 use super::models::{DbOAuthCode, DbOAuthRefreshToken};
8 use super::{SyncAppId, UserId};
9 use crate::error::Result;
10
11 /// Store a new OAuth authorization code, carrying the granted scope.
12 ///
13 /// `code_hash` is the SHA-256 hex of the opaque code, never the plaintext,
14 /// the `code` column holds the hash, looked up by hash in [`peek_oauth_code`] /
15 /// [`consume_oauth_code`], so a DB read never exposes a live, redeemable code.
16 /// Same at-rest contract as `oauth_refresh_tokens.token_hash`.
17 #[allow(clippy::too_many_arguments)]
18 #[tracing::instrument(skip_all)]
19 pub(crate) async fn create_oauth_code(
20 pool: &PgPool,
21 code_hash: &str,
22 app_id: SyncAppId,
23 user_id: UserId,
24 code_challenge: &str,
25 code_challenge_method: &str,
26 redirect_uri: &str,
27 scope: &str,
28 expires_at: DateTime<Utc>,
29 ) -> Result<DbOAuthCode> {
30 let row = sqlx::query_as::<_, DbOAuthCode>(
31 r"
32 INSERT INTO oauth_authorization_codes
33 (code, app_id, user_id, code_challenge, code_challenge_method, redirect_uri, scope, expires_at)
34 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
35 RETURNING *
36 ",
37 )
38 .bind(code_hash)
39 .bind(app_id)
40 .bind(user_id)
41 .bind(code_challenge)
42 .bind(code_challenge_method)
43 .bind(redirect_uri)
44 .bind(scope)
45 .bind(expires_at)
46 .fetch_one(pool)
47 .await?;
48
49 Ok(row)
50 }
51
52 /// Outcome of presenting a refresh token for rotation.
53 pub(crate) enum RefreshRotateOutcome {
54 /// The token was valid and has just been consumed (marked used). The caller
55 /// must now mint a replacement in the same `chain_id`.
56 Valid(Box<DbOAuthRefreshToken>),
57 /// The token exists but was already used (rotated). This is a reuse/theft
58 /// signal: the caller must revoke the whole `chain_id`.
59 Reused { chain_id: Uuid },
60 /// Unknown, expired, or already-revoked token, reject without side effects.
61 Invalid,
62 }
63
64 /// Store a new refresh token (hashed). Used both on the authorization-code
65 /// grant (first issuance) and on every rotation.
66 #[allow(clippy::too_many_arguments)]
67 #[tracing::instrument(skip_all)]
68 pub(crate) async fn create_refresh_token(
69 pool: &PgPool,
70 token_hash: &str,
71 app_id: SyncAppId,
72 user_id: UserId,
73 key: &str,
74 scope: &str,
75 chain_id: Uuid,
76 expires_at: DateTime<Utc>,
77 ) -> Result<DbOAuthRefreshToken> {
78 let row = sqlx::query_as::<_, DbOAuthRefreshToken>(
79 r"
80 INSERT INTO oauth_refresh_tokens
81 (token_hash, app_id, user_id, key, scope, chain_id, expires_at)
82 VALUES ($1, $2, $3, $4, $5, $6, $7)
83 RETURNING *
84 ",
85 )
86 .bind(token_hash)
87 .bind(app_id)
88 .bind(user_id)
89 .bind(key)
90 .bind(scope)
91 .bind(chain_id)
92 .bind(expires_at)
93 .fetch_one(pool)
94 .await?;
95
96 Ok(row)
97 }
98
99 /// Atomically consume a refresh token by its hash, classifying the outcome.
100 ///
101 /// The single `UPDATE ... WHERE used_at IS NULL ...` makes consumption
102 /// race-free (mirrors [`consume_oauth_code`]). A miss is then disambiguated:
103 /// an already-used row is a reuse/theft signal (caller revokes the chain),
104 /// anything else is invalid.
105 #[tracing::instrument(skip_all)]
106 pub(crate) async fn rotate_refresh_token(
107 pool: &PgPool,
108 token_hash: &str,
109 ) -> Result<RefreshRotateOutcome> {
110 let consumed = sqlx::query_as::<_, DbOAuthRefreshToken>(
111 r"
112 UPDATE oauth_refresh_tokens
113 SET used_at = NOW()
114 WHERE token_hash = $1
115 AND used_at IS NULL
116 AND revoked_at IS NULL
117 AND expires_at > NOW()
118 RETURNING *
119 ",
120 )
121 .bind(token_hash)
122 .fetch_optional(pool)
123 .await?;
124
125 if let Some(row) = consumed {
126 return Ok(RefreshRotateOutcome::Valid(Box::new(row)));
127 }
128
129 // No row consumed, was it a replay of an already-rotated token?
130 let existing: Option<(Uuid, Option<DateTime<Utc>>)> =
131 sqlx::query_as("SELECT chain_id, used_at FROM oauth_refresh_tokens WHERE token_hash = $1")
132 .bind(token_hash)
133 .fetch_optional(pool)
134 .await?;
135
136 match existing {
137 Some((chain_id, Some(_used))) => Ok(RefreshRotateOutcome::Reused { chain_id }),
138 _ => Ok(RefreshRotateOutcome::Invalid),
139 }
140 }
141
142 /// Revoke every refresh token in a chain, the response to a reuse/theft signal.
143 #[tracing::instrument(skip_all)]
144 pub(crate) async fn revoke_refresh_chain(pool: &PgPool, chain_id: Uuid) -> Result<()> {
145 sqlx::query(
146 "UPDATE oauth_refresh_tokens SET revoked_at = NOW() WHERE chain_id = $1 AND revoked_at IS NULL",
147 )
148 .bind(chain_id)
149 .execute(pool)
150 .await?;
151 Ok(())
152 }
153
154 /// Delete expired or long-used refresh tokens. Called opportunistically from
155 /// the health monitor loop alongside [`cleanup_expired_oauth_codes`].
156 #[tracing::instrument(skip_all)]
157 pub(crate) async fn cleanup_expired_refresh_tokens(pool: &PgPool) -> Result<u64> {
158 let result = sqlx::query(
159 "DELETE FROM oauth_refresh_tokens
160 WHERE expires_at < NOW()
161 OR (used_at IS NOT NULL AND used_at < NOW() - INTERVAL '1 day')
162 OR (revoked_at IS NOT NULL AND revoked_at < NOW() - INTERVAL '1 day')",
163 )
164 .execute(pool)
165 .await?;
166
167 Ok(result.rows_affected())
168 }
169
170 /// Fetch a still-valid authorization code WITHOUT consuming it.
171 ///
172 /// `code_hash` is the SHA-256 hex of the presented code (the column stores the
173 /// hash, not the plaintext). Lets the token handler validate client_id /
174 /// redirect_uri / PKCE against the code's stored values before burning it, so a
175 /// failed validation leaves the code usable for the legitimate client's retry.
176 /// The atomic `consume_oauth_code` below is what actually claims the code, so
177 /// concurrent redemptions remain race-safe; this is only a pre-flight read.
178 #[tracing::instrument(skip_all)]
179 pub(crate) async fn peek_oauth_code(pool: &PgPool, code_hash: &str) -> Result<Option<DbOAuthCode>> {
180 let row = sqlx::query_as::<_, DbOAuthCode>(
181 r"
182 SELECT * FROM oauth_authorization_codes
183 WHERE code = $1
184 AND used_at IS NULL
185 AND expires_at > NOW()
186 ",
187 )
188 .bind(code_hash)
189 .fetch_optional(pool)
190 .await?;
191
192 Ok(row)
193 }
194
195 /// Atomically consume an authorization code: mark it used and return it in one step.
196 ///
197 /// `code_hash` is the SHA-256 hex of the presented code. Returns `Some(code)` if
198 /// it was valid and successfully consumed, or `None` if already used, expired, or
199 /// nonexistent. Because this is a single UPDATE with `used_at IS NULL` in the
200 /// WHERE clause, concurrent requests for the same code will never both succeed.
201 #[tracing::instrument(skip_all)]
202 pub(crate) async fn consume_oauth_code(
203 pool: &PgPool,
204 code_hash: &str,
205 ) -> Result<Option<DbOAuthCode>> {
206 let row = sqlx::query_as::<_, DbOAuthCode>(
207 r"
208 UPDATE oauth_authorization_codes
209 SET used_at = NOW()
210 WHERE code = $1
211 AND used_at IS NULL
212 AND expires_at > NOW()
213 RETURNING *
214 ",
215 )
216 .bind(code_hash)
217 .fetch_optional(pool)
218 .await?;
219
220 Ok(row)
221 }
222
223 /// Delete expired or used authorization codes older than 1 hour.
224 /// Called opportunistically from the health monitor loop.
225 #[tracing::instrument(skip_all)]
226 pub(crate) async fn cleanup_expired_oauth_codes(pool: &PgPool) -> Result<u64> {
227 let result = sqlx::query(
228 "DELETE FROM oauth_authorization_codes WHERE expires_at < NOW() - INTERVAL '1 hour' OR (used_at IS NOT NULL AND used_at < NOW() - INTERVAL '1 hour')",
229 )
230 .execute(pool)
231 .await?;
232
233 Ok(result.rows_affected())
234 }
235
236 /// Check if a redirect URI is registered for a given sync app.
237 ///
238 /// Returns `Ok(false)` when the app row doesn't exist or is inactive, never
239 /// surfaces a "no rows" error to the caller. Matching is **exact-string** on
240 /// the registered `redirect_uris` array; trailing slashes are significant
241 /// (`https://x/cb` and `https://x/cb/` are distinct registrations), so apps
242 /// must register every variant they intend to redirect to.
243 #[tracing::instrument(skip_all)]
244 pub(crate) async fn is_registered_redirect_uri(
245 pool: &PgPool,
246 app_id: SyncAppId,
247 uri: &str,
248 ) -> Result<bool> {
249 let row: Option<(bool,)> = sqlx::query_as(
250 "SELECT $2 = ANY(redirect_uris) FROM sync_apps WHERE id = $1 AND is_active = true",
251 )
252 .bind(app_id)
253 .bind(uri)
254 .fetch_optional(pool)
255 .await?;
256
257 Ok(row.is_some_and(|r| r.0))
258 }
259
260 /// Scopes the user has already interactively consented to for this app. Empty
261 /// when the pair has no row (so a first prompt=none with any non-empty scope is
262 /// not a subset and must fall back to interactive consent).
263 #[tracing::instrument(skip_all)]
264 pub(crate) async fn get_granted_scopes(
265 pool: &PgPool,
266 user_id: UserId,
267 app_id: SyncAppId,
268 ) -> Result<crate::oauth_scope::GrantedScopes> {
269 let row = sqlx::query_scalar!(
270 "SELECT scopes FROM oauth_granted_scopes WHERE user_id = $1 AND app_id = $2",
271 user_id as UserId,
272 app_id as SyncAppId,
273 )
274 .fetch_optional(pool)
275 .await?;
276 Ok(row
277 .map(|s| crate::oauth_scope::GrantedScopes::parse(&s))
278 .unwrap_or_default())
279 }
280
281 /// Record an interactive consent: union the freshly-approved scope set into the
282 /// user's standing grant for this app, so a later prompt=none re-auth can
283 /// silently reuse what was approved.
284 #[tracing::instrument(skip_all)]
285 pub(crate) async fn record_granted_scopes(
286 pool: &PgPool,
287 user_id: UserId,
288 app_id: SyncAppId,
289 scope: &crate::oauth_scope::GrantedScopes,
290 ) -> Result<()> {
291 let mut tx = pool.begin().await?;
292
293 // Serialize concurrent consent recordings for this (user, app) so the
294 // read-modify-write union below can't lose a just-granted scope to a lost
295 // update (Sec-M1). Keyed on the pair, so it only contends with this user's
296 // own concurrent consents for this app and auto-releases at commit. Mirrors
297 // the per-reporter lock in db::reports::create_report_within_daily_limit.
298 sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1::text || ':' || $2::text, 0))")
299 .bind(user_id)
300 .bind(app_id)
301 .execute(&mut *tx)
302 .await?;
303
304 // Read the standing grant INSIDE the lock + transaction.
305 let existing: Option<String> = sqlx::query_scalar(
306 "SELECT scopes FROM oauth_granted_scopes WHERE user_id = $1 AND app_id = $2",
307 )
308 .bind(user_id)
309 .bind(app_id)
310 .fetch_optional(&mut *tx)
311 .await?;
312 let mut merged = existing
313 .map(|s| crate::oauth_scope::GrantedScopes::parse(&s))
314 .unwrap_or_default();
315 merged.union_with(scope);
316 let scopes = merged.to_string();
317
318 sqlx::query!(
319 r#"
320 INSERT INTO oauth_granted_scopes (user_id, app_id, scopes)
321 VALUES ($1, $2, $3)
322 ON CONFLICT (user_id, app_id)
323 DO UPDATE SET scopes = EXCLUDED.scopes, updated_at = NOW()
324 "#,
325 user_id as UserId,
326 app_id as SyncAppId,
327 scopes,
328 )
329 .execute(&mut *tx)
330 .await?;
331 tx.commit().await?;
332 Ok(())
333 }
334