Skip to main content

max / makenotwork

11.1 KB · 336 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 /// (ultra-fuzz Run #1 Security LOW: the code was previously marked used before
177 /// any of those checks). The atomic `consume_oauth_code` below is still what
178 /// actually claims the code, so concurrent redemptions remain race-safe, this
179 /// is only a pre-flight read.
180 #[tracing::instrument(skip_all)]
181 pub(crate) async fn peek_oauth_code(pool: &PgPool, code_hash: &str) -> Result<Option<DbOAuthCode>> {
182 let row = sqlx::query_as::<_, DbOAuthCode>(
183 r"
184 SELECT * FROM oauth_authorization_codes
185 WHERE code = $1
186 AND used_at IS NULL
187 AND expires_at > NOW()
188 ",
189 )
190 .bind(code_hash)
191 .fetch_optional(pool)
192 .await?;
193
194 Ok(row)
195 }
196
197 /// Atomically consume an authorization code: mark it used and return it in one step.
198 ///
199 /// `code_hash` is the SHA-256 hex of the presented code. Returns `Some(code)` if
200 /// it was valid and successfully consumed, or `None` if already used, expired, or
201 /// nonexistent. Because this is a single UPDATE with `used_at IS NULL` in the
202 /// WHERE clause, concurrent requests for the same code will never both succeed.
203 #[tracing::instrument(skip_all)]
204 pub(crate) async fn consume_oauth_code(
205 pool: &PgPool,
206 code_hash: &str,
207 ) -> Result<Option<DbOAuthCode>> {
208 let row = sqlx::query_as::<_, DbOAuthCode>(
209 r"
210 UPDATE oauth_authorization_codes
211 SET used_at = NOW()
212 WHERE code = $1
213 AND used_at IS NULL
214 AND expires_at > NOW()
215 RETURNING *
216 ",
217 )
218 .bind(code_hash)
219 .fetch_optional(pool)
220 .await?;
221
222 Ok(row)
223 }
224
225 /// Delete expired or used authorization codes older than 1 hour.
226 /// Called opportunistically from the health monitor loop.
227 #[tracing::instrument(skip_all)]
228 pub(crate) async fn cleanup_expired_oauth_codes(pool: &PgPool) -> Result<u64> {
229 let result = sqlx::query(
230 "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')",
231 )
232 .execute(pool)
233 .await?;
234
235 Ok(result.rows_affected())
236 }
237
238 /// Check if a redirect URI is registered for a given sync app.
239 ///
240 /// Returns `Ok(false)` when the app row doesn't exist or is inactive, never
241 /// surfaces a "no rows" error to the caller. Matching is **exact-string** on
242 /// the registered `redirect_uris` array; trailing slashes are significant
243 /// (`https://x/cb` and `https://x/cb/` are distinct registrations), so apps
244 /// must register every variant they intend to redirect to.
245 #[tracing::instrument(skip_all)]
246 pub(crate) async fn is_registered_redirect_uri(
247 pool: &PgPool,
248 app_id: SyncAppId,
249 uri: &str,
250 ) -> Result<bool> {
251 let row: Option<(bool,)> = sqlx::query_as(
252 "SELECT $2 = ANY(redirect_uris) FROM sync_apps WHERE id = $1 AND is_active = true",
253 )
254 .bind(app_id)
255 .bind(uri)
256 .fetch_optional(pool)
257 .await?;
258
259 Ok(row.is_some_and(|r| r.0))
260 }
261
262 /// Scopes the user has already interactively consented to for this app. Empty
263 /// when the pair has no row (so a first prompt=none with any non-empty scope is
264 /// not a subset and must fall back to interactive consent). Run 6 R6-Sec-L5.
265 #[tracing::instrument(skip_all)]
266 pub(crate) async fn get_granted_scopes(
267 pool: &PgPool,
268 user_id: UserId,
269 app_id: SyncAppId,
270 ) -> Result<crate::oauth_scope::GrantedScopes> {
271 let row = sqlx::query_scalar!(
272 "SELECT scopes FROM oauth_granted_scopes WHERE user_id = $1 AND app_id = $2",
273 user_id as UserId,
274 app_id as SyncAppId,
275 )
276 .fetch_optional(pool)
277 .await?;
278 Ok(row
279 .map(|s| crate::oauth_scope::GrantedScopes::parse(&s))
280 .unwrap_or_default())
281 }
282
283 /// Record an interactive consent: union the freshly-approved scope set into the
284 /// user's standing grant for this app, so a later prompt=none re-auth can
285 /// silently reuse what was approved (Run 6 R6-Sec-L5).
286 #[tracing::instrument(skip_all)]
287 pub(crate) async fn record_granted_scopes(
288 pool: &PgPool,
289 user_id: UserId,
290 app_id: SyncAppId,
291 scope: &crate::oauth_scope::GrantedScopes,
292 ) -> Result<()> {
293 let mut tx = pool.begin().await?;
294
295 // Serialize concurrent consent recordings for this (user, app) so the
296 // read-modify-write union below can't lose a just-granted scope to a lost
297 // update (Sec-M1). Keyed on the pair, so it only contends with this user's
298 // own concurrent consents for this app and auto-releases at commit. Mirrors
299 // the per-reporter lock in db::reports::create_report_within_daily_limit.
300 sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1::text || ':' || $2::text, 0))")
301 .bind(user_id)
302 .bind(app_id)
303 .execute(&mut *tx)
304 .await?;
305
306 // Read the standing grant INSIDE the lock + transaction.
307 let existing: Option<String> = sqlx::query_scalar(
308 "SELECT scopes FROM oauth_granted_scopes WHERE user_id = $1 AND app_id = $2",
309 )
310 .bind(user_id)
311 .bind(app_id)
312 .fetch_optional(&mut *tx)
313 .await?;
314 let mut merged = existing
315 .map(|s| crate::oauth_scope::GrantedScopes::parse(&s))
316 .unwrap_or_default();
317 merged.union_with(scope);
318 let scopes = merged.to_string();
319
320 sqlx::query!(
321 r#"
322 INSERT INTO oauth_granted_scopes (user_id, app_id, scopes)
323 VALUES ($1, $2, $3)
324 ON CONFLICT (user_id, app_id)
325 DO UPDATE SET scopes = EXCLUDED.scopes, updated_at = NOW()
326 "#,
327 user_id as UserId,
328 app_id as SyncAppId,
329 scopes,
330 )
331 .execute(&mut *tx)
332 .await?;
333 tx.commit().await?;
334 Ok(())
335 }
336