Skip to main content

max / makenotwork

10.9 KB · 329 lines History Blame Raw
1 //! Session tracking for remote revocation.
2
3 use sqlx::PgPool;
4
5 use super::{DbUserSession, UserId, UserSessionId};
6 use crate::error::Result;
7
8 /// Insert a tracked session row and return its ID.
9 #[tracing::instrument(skip_all)]
10 pub async fn create_user_session(
11 pool: &PgPool,
12 user_id: UserId,
13 user_agent: Option<&str>,
14 ip_address: Option<&str>,
15 ) -> Result<UserSessionId> {
16 let row = sqlx::query_scalar!(
17 r#"INSERT INTO user_sessions (user_id, user_agent, ip_address) VALUES ($1, $2, $3) RETURNING id AS "id: UserSessionId""#,
18 user_id as UserId,
19 user_agent,
20 ip_address,
21 )
22 .fetch_one(pool)
23 .await?;
24
25 Ok(row)
26 }
27
28 /// Cap a user's active sessions to the newest `max`, deleting older ones.
29 ///
30 /// Bounds unbounded `user_sessions` growth from repeated logins (or an automated
31 /// loop). Only `kind='active'` rows are counted and pruned, `pending_2fa`
32 /// intermediates are never touched. Ordered by `last_active_at DESC`, so the
33 /// freshly-created current session (newest) is always kept and the stalest are
34 /// evicted. Returns the number of rows pruned.
35 ///
36 /// The keep-set is ordered `last_active_at DESC, id DESC`, the `id` tie-break
37 /// makes the eviction deterministic when several sessions share a timestamp,
38 /// instead of leaving the choice to Postgres' undefined row order (Run 11 Sec LOW).
39 #[tracing::instrument(skip_all)]
40 pub async fn prune_user_sessions_over_cap(pool: &PgPool, user_id: UserId, max: i64) -> Result<u64> {
41 let result = sqlx::query!(
42 r#"
43 DELETE FROM user_sessions
44 WHERE user_id = $1 AND kind = 'active'
45 AND id NOT IN (
46 SELECT id FROM user_sessions
47 WHERE user_id = $1 AND kind = 'active'
48 ORDER BY last_active_at DESC, id DESC
49 LIMIT $2
50 )
51 "#,
52 user_id as UserId,
53 max,
54 )
55 .execute(pool)
56 .await?;
57
58 Ok(result.rows_affected())
59 }
60
61 /// Insert a `kind='pending_2fa'` row for an intermediate session held between
62 /// the password step and the TOTP/backup-code step. Exposed to
63 /// `delete_all_sessions_for_user` so "log out everywhere" sweeps a phisher
64 /// who has the password but not the second factor.
65 #[tracing::instrument(skip_all)]
66 pub async fn create_pending_2fa_session(
67 pool: &PgPool,
68 user_id: UserId,
69 user_agent: Option<&str>,
70 ip_address: Option<&str>,
71 ) -> Result<UserSessionId> {
72 let row = sqlx::query_scalar!(
73 r#"INSERT INTO user_sessions (user_id, user_agent, ip_address, kind)
74 VALUES ($1, $2, $3, 'pending_2fa') RETURNING id AS "id: UserSessionId""#,
75 user_id as UserId,
76 user_agent,
77 ip_address,
78 )
79 .fetch_one(pool)
80 .await?;
81
82 Ok(row)
83 }
84
85 /// Confirm the pending_2fa tracking row is still present (i.e. wasn't swept
86 /// by `delete_all_sessions_for_user` while the user was at the TOTP prompt).
87 #[tracing::instrument(skip_all)]
88 pub async fn pending_2fa_session_exists(
89 pool: &PgPool,
90 id: UserSessionId,
91 user_id: UserId,
92 ) -> Result<bool> {
93 let exists = sqlx::query_scalar!(
94 r#"SELECT EXISTS(SELECT 1 FROM user_sessions WHERE id = $1 AND user_id = $2 AND kind = 'pending_2fa') AS "exists!""#,
95 id as UserSessionId,
96 user_id as UserId,
97 )
98 .fetch_one(pool)
99 .await?;
100 Ok(exists)
101 }
102
103 /// Delete a pending_2fa tracking row. Called when 2FA succeeds (the caller
104 /// then `track_session`s a fresh 'active' row) or when the pending state is
105 /// cleared (expiry, account lockout, navigation away).
106 ///
107 /// Scoped by `user_id` (not id alone) so a guessed/enumerated session id can't
108 /// delete another user's pending_2fa row (ultra-fuzz Run 11 Sec LOW).
109 #[tracing::instrument(skip_all)]
110 pub async fn delete_pending_2fa_session(
111 pool: &PgPool,
112 id: UserSessionId,
113 user_id: UserId,
114 ) -> Result<()> {
115 sqlx::query!(
116 "DELETE FROM user_sessions WHERE id = $1 AND user_id = $2 AND kind = 'pending_2fa'",
117 id as UserSessionId,
118 user_id as UserId,
119 )
120 .execute(pool)
121 .await?;
122 Ok(())
123 }
124
125 /// Result of touching a session: whether it exists and the user's current
126 /// suspended status (live from the `users` table, not cached in the session).
127 pub struct TouchResult {
128 /// `false` if the session row was deleted (revoked).
129 pub valid: bool,
130 /// Current `suspended_at IS NOT NULL` from the users table.
131 /// Only meaningful when `valid` is `true`.
132 pub suspended: bool,
133 /// Current `can_create_projects` from the users table.
134 /// Only meaningful when `valid` is `true`.
135 pub can_create_projects: bool,
136 /// Whether the user has an active Fan+ subscription.
137 pub is_fan_plus: bool,
138 /// Active creator tier name (e.g. "SmallFiles"), or None.
139 pub creator_tier: Option<String>,
140 }
141
142 /// Update `last_active_at`, confirm the session still exists, and return
143 /// the user's current `suspended` status from the `users` table.
144 ///
145 /// This ensures suspension takes effect immediately even if the session
146 /// was created before the admin suspended the user.
147 #[tracing::instrument(skip_all)]
148 pub async fn touch_session(pool: &PgPool, session_id: UserSessionId) -> Result<TouchResult> {
149 // Single query: update last_active_at, join users for live status, and check
150 // fan_plus + creator_tier via subqueries (avoids 2 extra round-trips in auth extractor).
151 let row = sqlx::query!(
152 r#"
153 UPDATE user_sessions us
154 SET last_active_at = NOW()
155 FROM users u
156 WHERE us.id = $1 AND u.id = us.user_id
157 RETURNING
158 u.suspended_at IS NOT NULL AS "suspended!",
159 u.can_create_projects AS "can_create_projects!",
160 EXISTS(SELECT 1 FROM fan_plus_subscriptions fps WHERE fps.user_id = u.id AND fps.status = 'active') AS "is_fan_plus!",
161 (SELECT cs.tier FROM creator_subscriptions cs WHERE cs.user_id = u.id AND cs.status = 'active') AS "creator_tier"
162 "#,
163 session_id as UserSessionId,
164 )
165 .map(|r| (r.suspended, r.can_create_projects, r.is_fan_plus, r.creator_tier))
166 .fetch_optional(pool)
167 .await?;
168
169 match row {
170 Some((suspended, can_create_projects, is_fan_plus, creator_tier)) => Ok(TouchResult {
171 valid: true,
172 suspended,
173 can_create_projects,
174 is_fan_plus,
175 creator_tier,
176 }),
177 None => Ok(TouchResult {
178 valid: false,
179 suspended: false,
180 can_create_projects: false,
181 is_fan_plus: false,
182 creator_tier: None,
183 }),
184 }
185 }
186
187 /// List all active sessions for a user, newest first.
188 #[tracing::instrument(skip_all)]
189 pub async fn get_user_sessions(pool: &PgPool, user_id: UserId) -> Result<Vec<DbUserSession>> {
190 let sessions = sqlx::query_as!(
191 DbUserSession,
192 r#"SELECT id AS "id: UserSessionId", user_id AS "user_id: UserId",
193 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
194 last_active_at AS "last_active_at: chrono::DateTime<chrono::Utc>",
195 user_agent, ip_address
196 FROM user_sessions
197 WHERE user_id = $1
198 ORDER BY last_active_at DESC
199 LIMIT 100"#,
200 user_id as UserId,
201 )
202 .fetch_all(pool)
203 .await?;
204
205 Ok(sessions)
206 }
207
208 /// Count active sessions for a user.
209 #[tracing::instrument(skip_all)]
210 pub async fn count_user_sessions(pool: &PgPool, user_id: UserId) -> Result<i64> {
211 let count = sqlx::query_scalar!(
212 r#"SELECT COUNT(*) AS "count!" FROM user_sessions WHERE user_id = $1"#,
213 user_id as UserId,
214 )
215 .fetch_one(pool)
216 .await?;
217
218 Ok(count)
219 }
220
221 /// Delete a single session, scoped to the owning user. Returns `true` if deleted.
222 #[tracing::instrument(skip_all)]
223 pub async fn delete_user_session(
224 pool: &PgPool,
225 session_id: UserSessionId,
226 user_id: UserId,
227 ) -> Result<bool> {
228 let rows = sqlx::query!(
229 "DELETE FROM user_sessions WHERE id = $1 AND user_id = $2",
230 session_id as UserSessionId,
231 user_id as UserId,
232 )
233 .execute(pool)
234 .await?;
235
236 Ok(rows.rows_affected() > 0)
237 }
238
239 /// Delete a session row, scoped to a specific user.
240 ///
241 /// The user scoping isn't strictly required for correctness in the current
242 /// caller (logout reads its own tracking ID out of the session and we
243 /// trust that), but the unscoped signature was an easy footgun, anyone
244 /// who later wired this up with an attacker-controllable session_id could
245 /// delete arbitrary rows. Requiring user_id in the signature keeps the
246 /// SQL pinned to "this user, this row" so that misuse fails fast.
247 #[tracing::instrument(skip_all)]
248 pub async fn delete_session_by_id(
249 pool: &PgPool,
250 session_id: UserSessionId,
251 user_id: UserId,
252 ) -> Result<bool> {
253 let rows = sqlx::query!(
254 "DELETE FROM user_sessions WHERE id = $1 AND user_id = $2",
255 session_id as UserSessionId,
256 user_id as UserId,
257 )
258 .execute(pool)
259 .await?;
260
261 Ok(rows.rows_affected() > 0)
262 }
263
264 /// Delete expired session records (inactive longer than the given threshold).
265 /// Returns the number of rows removed.
266 #[tracing::instrument(skip_all)]
267 pub async fn prune_expired_sessions(
268 pool: &PgPool,
269 stale_threshold: chrono::DateTime<chrono::Utc>,
270 ) -> Result<u64> {
271 // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
272 let result = sqlx::query("DELETE FROM user_sessions WHERE last_active_at < $1")
273 .bind(stale_threshold)
274 .execute(pool)
275 .await?;
276
277 Ok(result.rows_affected())
278 }
279
280 /// Delete all sessions for a user except the current one. Returns count deleted.
281 #[tracing::instrument(skip_all)]
282 pub async fn delete_other_sessions(
283 pool: &PgPool,
284 current_session_id: UserSessionId,
285 user_id: UserId,
286 ) -> Result<Vec<UserSessionId>> {
287 let ids = sqlx::query_scalar!(
288 r#"DELETE FROM user_sessions WHERE user_id = $1 AND id != $2 RETURNING id AS "id: UserSessionId""#,
289 user_id as UserId,
290 current_session_id as UserSessionId,
291 )
292 .fetch_all(pool)
293 .await?;
294
295 Ok(ids)
296 }
297
298 /// Delete ALL sessions for a user. Returns the deleted session IDs (for cache eviction).
299 ///
300 /// Also bumps `users.jwt_invalidated_at`, without this, a stolen SyncKit JWT
301 /// would survive a "log out everywhere" until its natural expiry. Both writes
302 /// run in a single transaction so a partial failure can't leave the JWTs alive
303 /// after the session rows are gone.
304 #[tracing::instrument(skip_all)]
305 pub async fn delete_all_sessions_for_user(
306 pool: &PgPool,
307 user_id: UserId,
308 ) -> Result<Vec<UserSessionId>> {
309 let mut tx = pool.begin().await?;
310
311 let ids = sqlx::query_scalar!(
312 r#"DELETE FROM user_sessions WHERE user_id = $1 RETURNING id AS "id: UserSessionId""#,
313 user_id as UserId,
314 )
315 .fetch_all(&mut *tx)
316 .await?;
317
318 sqlx::query!(
319 "UPDATE users SET jwt_invalidated_at = NOW() WHERE id = $1",
320 user_id as UserId,
321 )
322 .execute(&mut *tx)
323 .await?;
324
325 tx.commit().await?;
326
327 Ok(ids)
328 }
329