this repo has no description
lewis.moe
1use crate::api::error::ApiError;
2use crate::api::{EmptyResponse, SuccessResponse};
3use crate::auth::{BearerAuth, BearerAuthAllowDeactivated};
4use crate::state::{AppState, RateLimitKind};
5use crate::types::{AccountState, Did, Handle, PlainPassword};
6use axum::{
7 Json,
8 extract::State,
9 http::{HeaderMap, StatusCode},
10 response::{IntoResponse, Response},
11};
12use bcrypt::verify;
13use serde::{Deserialize, Serialize};
14use serde_json::json;
15use tracing::{error, info, warn};
16
17fn extract_client_ip(headers: &HeaderMap) -> String {
18 if let Some(forwarded) = headers.get("x-forwarded-for")
19 && let Ok(value) = forwarded.to_str()
20 && let Some(first_ip) = value.split(',').next()
21 {
22 return first_ip.trim().to_string();
23 }
24 if let Some(real_ip) = headers.get("x-real-ip")
25 && let Ok(value) = real_ip.to_str()
26 {
27 return value.trim().to_string();
28 }
29 "unknown".to_string()
30}
31
32fn normalize_handle(identifier: &str, pds_hostname: &str) -> String {
33 let identifier = identifier.trim();
34 if identifier.contains('@') || identifier.starts_with("did:") {
35 identifier.to_string()
36 } else if !identifier.contains('.') {
37 format!("{}.{}", identifier.to_lowercase(), pds_hostname)
38 } else {
39 identifier.to_lowercase()
40 }
41}
42
43fn full_handle(stored_handle: &str, _pds_hostname: &str) -> String {
44 stored_handle.to_string()
45}
46
47#[derive(Deserialize)]
48#[serde(rename_all = "camelCase")]
49pub struct CreateSessionInput {
50 pub identifier: String,
51 pub password: PlainPassword,
52 #[serde(default)]
53 pub allow_takendown: bool,
54}
55
56#[derive(Serialize)]
57#[serde(rename_all = "camelCase")]
58pub struct CreateSessionOutput {
59 pub access_jwt: String,
60 pub refresh_jwt: String,
61 pub handle: Handle,
62 pub did: Did,
63 #[serde(skip_serializing_if = "Option::is_none")]
64 pub did_doc: Option<serde_json::Value>,
65 #[serde(skip_serializing_if = "Option::is_none")]
66 pub email: Option<String>,
67 #[serde(skip_serializing_if = "Option::is_none")]
68 pub email_confirmed: Option<bool>,
69 #[serde(skip_serializing_if = "Option::is_none")]
70 pub active: Option<bool>,
71 #[serde(skip_serializing_if = "Option::is_none")]
72 pub status: Option<String>,
73}
74
75pub async fn create_session(
76 State(state): State<AppState>,
77 headers: HeaderMap,
78 Json(input): Json<CreateSessionInput>,
79) -> Response {
80 info!(
81 "create_session called with identifier: {}",
82 input.identifier
83 );
84 let client_ip = extract_client_ip(&headers);
85 if !state
86 .check_rate_limit(RateLimitKind::Login, &client_ip)
87 .await
88 {
89 warn!(ip = %client_ip, "Login rate limit exceeded");
90 return ApiError::RateLimitExceeded(None).into_response();
91 }
92 let pds_hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
93 let normalized_identifier = normalize_handle(&input.identifier, &pds_hostname);
94 info!(
95 "Normalized identifier: {} -> {}",
96 input.identifier, normalized_identifier
97 );
98 let row = match sqlx::query!(
99 r#"SELECT
100 u.id, u.did, u.handle, u.password_hash, u.email, u.deactivated_at, u.takedown_ref,
101 u.email_verified, u.discord_verified, u.telegram_verified, u.signal_verified,
102 u.allow_legacy_login, u.migrated_to_pds,
103 u.preferred_comms_channel as "preferred_comms_channel: crate::comms::CommsChannel",
104 k.key_bytes, k.encryption_version,
105 (SELECT verified FROM user_totp WHERE did = u.did) as totp_enabled
106 FROM users u
107 JOIN user_keys k ON u.id = k.user_id
108 WHERE u.handle = $1 OR u.email = $1 OR u.did = $1"#,
109 normalized_identifier
110 )
111 .fetch_optional(&state.db)
112 .await
113 {
114 Ok(Some(row)) => row,
115 Ok(None) => {
116 let _ = verify(
117 &input.password,
118 "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYw1ZzQKZqmK",
119 );
120 warn!("User not found for login attempt");
121 return ApiError::AuthenticationFailed(Some("Invalid identifier or password".into()))
122 .into_response();
123 }
124 Err(e) => {
125 error!("Database error fetching user: {:?}", e);
126 return ApiError::InternalError(None).into_response();
127 }
128 };
129 let key_bytes = match crate::config::decrypt_key(&row.key_bytes, row.encryption_version) {
130 Ok(k) => k,
131 Err(e) => {
132 error!("Failed to decrypt user key: {:?}", e);
133 return ApiError::InternalError(None).into_response();
134 }
135 };
136 let (password_valid, app_password_name, app_password_scopes, app_password_controller) = if row
137 .password_hash
138 .as_ref()
139 .map(|h| verify(&input.password, h).unwrap_or(false))
140 .unwrap_or(false)
141 {
142 (true, None, None, None)
143 } else {
144 let app_passwords = sqlx::query!(
145 "SELECT name, password_hash, scopes, created_by_controller_did FROM app_passwords WHERE user_id = $1 ORDER BY created_at DESC LIMIT 20",
146 row.id
147 )
148 .fetch_all(&state.db)
149 .await
150 .unwrap_or_default();
151 let matched = app_passwords
152 .iter()
153 .find(|app| verify(&input.password, &app.password_hash).unwrap_or(false));
154 match matched {
155 Some(app) => (
156 true,
157 Some(app.name.clone()),
158 app.scopes.clone(),
159 app.created_by_controller_did.clone(),
160 ),
161 None => (false, None, None, None),
162 }
163 };
164 if !password_valid {
165 warn!("Password verification failed for login attempt");
166 return ApiError::AuthenticationFailed(Some("Invalid identifier or password".into()))
167 .into_response();
168 }
169 let account_state = AccountState::from_db_fields(
170 row.deactivated_at,
171 row.takedown_ref.clone(),
172 row.migrated_to_pds.clone(),
173 None,
174 );
175 if account_state.is_takendown() && !input.allow_takendown {
176 warn!("Login attempt for takendown account: {}", row.did);
177 return ApiError::AccountTakedown.into_response();
178 }
179 let is_verified =
180 row.email_verified || row.discord_verified || row.telegram_verified || row.signal_verified;
181 let is_delegated = crate::delegation::is_delegated_account(&state.db, &row.did)
182 .await
183 .unwrap_or(false);
184 if !is_verified && !is_delegated {
185 warn!("Login attempt for unverified account: {}", row.did);
186 return (
187 StatusCode::FORBIDDEN,
188 Json(json!({
189 "error": "AccountNotVerified",
190 "message": "Please verify your account before logging in",
191 "did": row.did
192 })),
193 )
194 .into_response();
195 }
196 let has_totp = row.totp_enabled.unwrap_or(false);
197 let is_legacy_login = has_totp;
198 if has_totp && !row.allow_legacy_login {
199 warn!("Legacy login blocked for TOTP-enabled account: {}", row.did);
200 return (
201 StatusCode::FORBIDDEN,
202 Json(json!({
203 "error": "MfaRequired",
204 "message": "This account requires MFA. Please use an OAuth client that supports TOTP verification.",
205 "did": row.did
206 })),
207 )
208 .into_response();
209 }
210 let access_meta = match crate::auth::create_access_token_with_delegation(
211 &row.did,
212 &key_bytes,
213 app_password_scopes.as_deref(),
214 app_password_controller.as_deref(),
215 None,
216 ) {
217 Ok(m) => m,
218 Err(e) => {
219 error!("Failed to create access token: {:?}", e);
220 return ApiError::InternalError(None).into_response();
221 }
222 };
223 let refresh_meta = match crate::auth::create_refresh_token_with_metadata(&row.did, &key_bytes) {
224 Ok(m) => m,
225 Err(e) => {
226 error!("Failed to create refresh token: {:?}", e);
227 return ApiError::InternalError(None).into_response();
228 }
229 };
230 let did_for_doc = row.did.clone();
231 let did_resolver = state.did_resolver.clone();
232 let (insert_result, did_doc) = tokio::join!(
233 sqlx::query!(
234 "INSERT INTO session_tokens (did, access_jti, refresh_jti, access_expires_at, refresh_expires_at, legacy_login, mfa_verified, scope, controller_did, app_password_name) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
235 row.did,
236 access_meta.jti,
237 refresh_meta.jti,
238 access_meta.expires_at,
239 refresh_meta.expires_at,
240 is_legacy_login,
241 false,
242 app_password_scopes,
243 app_password_controller,
244 app_password_name
245 )
246 .execute(&state.db),
247 did_resolver.resolve_did_document(&did_for_doc)
248 );
249 if let Err(e) = insert_result {
250 error!("Failed to insert session: {:?}", e);
251 return ApiError::InternalError(None).into_response();
252 }
253 if is_legacy_login {
254 warn!(
255 did = %row.did,
256 ip = %client_ip,
257 "Legacy login on TOTP-enabled account - sending notification"
258 );
259 let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
260 if let Err(e) = crate::comms::queue_legacy_login_notification(
261 &state.db,
262 row.id,
263 &hostname,
264 &client_ip,
265 row.preferred_comms_channel,
266 )
267 .await
268 {
269 error!("Failed to queue legacy login notification: {:?}", e);
270 }
271 }
272 let handle = full_handle(&row.handle, &pds_hostname);
273 let is_active = account_state.is_active();
274 let status = account_state.status_for_session().map(String::from);
275 Json(CreateSessionOutput {
276 access_jwt: access_meta.token,
277 refresh_jwt: refresh_meta.token,
278 handle: handle.into(),
279 did: row.did.into(),
280 did_doc,
281 email: row.email,
282 email_confirmed: Some(row.email_verified),
283 active: Some(is_active),
284 status,
285 })
286 .into_response()
287}
288
289pub async fn get_session(
290 State(state): State<AppState>,
291 BearerAuthAllowDeactivated(auth_user): BearerAuthAllowDeactivated,
292) -> Response {
293 let permissions = auth_user.permissions();
294 let can_read_email = permissions.allows_email_read();
295
296 let did_for_doc = auth_user.did.clone();
297 let did_resolver = state.did_resolver.clone();
298 let (db_result, did_doc) = tokio::join!(
299 sqlx::query!(
300 r#"SELECT
301 handle, email, email_verified, is_admin, deactivated_at, takedown_ref, preferred_locale,
302 preferred_comms_channel as "preferred_channel: crate::comms::CommsChannel",
303 discord_verified, telegram_verified, signal_verified, migrated_to_pds, migrated_at
304 FROM users WHERE did = $1"#,
305 &auth_user.did
306 )
307 .fetch_optional(&state.db),
308 did_resolver.resolve_did_document(&did_for_doc)
309 );
310 match db_result {
311 Ok(Some(row)) => {
312 let (preferred_channel, preferred_channel_verified) = match row.preferred_channel {
313 crate::comms::CommsChannel::Email => ("email", row.email_verified),
314 crate::comms::CommsChannel::Discord => ("discord", row.discord_verified),
315 crate::comms::CommsChannel::Telegram => ("telegram", row.telegram_verified),
316 crate::comms::CommsChannel::Signal => ("signal", row.signal_verified),
317 };
318 let pds_hostname =
319 std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
320 let handle = full_handle(&row.handle, &pds_hostname);
321 let account_state = AccountState::from_db_fields(
322 row.deactivated_at,
323 row.takedown_ref.clone(),
324 row.migrated_to_pds.clone(),
325 row.migrated_at,
326 );
327 let email_value = if can_read_email {
328 row.email.clone()
329 } else {
330 None
331 };
332 let email_confirmed_value = can_read_email && row.email_verified;
333 let mut response = json!({
334 "handle": handle,
335 "did": &auth_user.did,
336 "active": account_state.is_active(),
337 "preferredChannel": preferred_channel,
338 "preferredChannelVerified": preferred_channel_verified,
339 "preferredLocale": row.preferred_locale,
340 "isAdmin": row.is_admin
341 });
342 if can_read_email {
343 response["email"] = json!(email_value);
344 response["emailConfirmed"] = json!(email_confirmed_value);
345 }
346 if let Some(status) = account_state.status_for_session() {
347 response["status"] = json!(status);
348 }
349 if let AccountState::Migrated { to_pds, at } = &account_state {
350 response["migratedToPds"] = json!(to_pds);
351 response["migratedAt"] = json!(at);
352 }
353 if let Some(doc) = did_doc {
354 response["didDoc"] = doc;
355 }
356 Json(response).into_response()
357 }
358 Ok(None) => ApiError::AuthenticationFailed(None).into_response(),
359 Err(e) => {
360 error!("Database error in get_session: {:?}", e);
361 ApiError::InternalError(None).into_response()
362 }
363 }
364}
365
366pub async fn delete_session(
367 State(state): State<AppState>,
368 headers: axum::http::HeaderMap,
369 _auth: BearerAuth,
370) -> Response {
371 let extracted = match crate::auth::extract_auth_token_from_header(
372 headers.get("Authorization").and_then(|h| h.to_str().ok()),
373 ) {
374 Some(t) => t,
375 None => return ApiError::AuthenticationRequired.into_response(),
376 };
377 let jti = match crate::auth::get_jti_from_token(&extracted.token) {
378 Ok(jti) => jti,
379 Err(_) => return ApiError::AuthenticationFailed(None).into_response(),
380 };
381 let did = crate::auth::get_did_from_token(&extracted.token).ok();
382 match sqlx::query!("DELETE FROM session_tokens WHERE access_jti = $1", jti)
383 .execute(&state.db)
384 .await
385 {
386 Ok(res) if res.rows_affected() > 0 => {
387 if let Some(did) = did {
388 let session_cache_key = format!("auth:session:{}:{}", did, jti);
389 let _ = state.cache.delete(&session_cache_key).await;
390 }
391 EmptyResponse::ok().into_response()
392 }
393 Ok(_) => ApiError::AuthenticationFailed(None).into_response(),
394 Err(e) => {
395 error!("Database error in delete_session: {:?}", e);
396 ApiError::AuthenticationFailed(None).into_response()
397 }
398 }
399}
400
401pub async fn refresh_session(
402 State(state): State<AppState>,
403 headers: axum::http::HeaderMap,
404) -> Response {
405 let client_ip = crate::rate_limit::extract_client_ip(&headers, None);
406 if !state
407 .check_rate_limit(RateLimitKind::RefreshSession, &client_ip)
408 .await
409 {
410 tracing::warn!(ip = %client_ip, "Refresh session rate limit exceeded");
411 return ApiError::RateLimitExceeded(None).into_response();
412 }
413 let extracted = match crate::auth::extract_auth_token_from_header(
414 headers.get("Authorization").and_then(|h| h.to_str().ok()),
415 ) {
416 Some(t) => t,
417 None => return ApiError::AuthenticationRequired.into_response(),
418 };
419 let refresh_token = extracted.token;
420 let refresh_jti = match crate::auth::get_jti_from_token(&refresh_token) {
421 Ok(jti) => jti,
422 Err(_) => {
423 return ApiError::AuthenticationFailed(Some("Invalid token format".into()))
424 .into_response();
425 }
426 };
427 let mut tx = match state.db.begin().await {
428 Ok(tx) => tx,
429 Err(e) => {
430 error!("Failed to begin transaction: {:?}", e);
431 return ApiError::InternalError(None).into_response();
432 }
433 };
434 if let Ok(Some(session_id)) = sqlx::query_scalar!(
435 "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1 FOR UPDATE",
436 refresh_jti
437 )
438 .fetch_optional(&mut *tx)
439 .await
440 {
441 warn!(
442 "Refresh token reuse detected! Revoking token family for session_id: {}",
443 session_id
444 );
445 let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_id)
446 .execute(&mut *tx)
447 .await;
448 let _ = tx.commit().await;
449 return ApiError::AuthenticationFailed(Some(
450 "Refresh token has been revoked due to suspected compromise".into(),
451 ))
452 .into_response();
453 }
454 let session_row = match sqlx::query!(
455 r#"SELECT st.id, st.did, st.scope, st.controller_did, k.key_bytes, k.encryption_version
456 FROM session_tokens st
457 JOIN users u ON st.did = u.did
458 JOIN user_keys k ON u.id = k.user_id
459 WHERE st.refresh_jti = $1 AND st.refresh_expires_at > NOW()
460 FOR UPDATE OF st"#,
461 refresh_jti
462 )
463 .fetch_optional(&mut *tx)
464 .await
465 {
466 Ok(Some(row)) => row,
467 Ok(None) => {
468 return ApiError::AuthenticationFailed(Some("Invalid refresh token".into()))
469 .into_response();
470 }
471 Err(e) => {
472 error!("Database error fetching session: {:?}", e);
473 return ApiError::InternalError(None).into_response();
474 }
475 };
476 let key_bytes =
477 match crate::config::decrypt_key(&session_row.key_bytes, session_row.encryption_version) {
478 Ok(k) => k,
479 Err(e) => {
480 error!("Failed to decrypt user key: {:?}", e);
481 return ApiError::InternalError(None).into_response();
482 }
483 };
484 if crate::auth::verify_refresh_token(&refresh_token, &key_bytes).is_err() {
485 return ApiError::AuthenticationFailed(Some("Invalid refresh token".into()))
486 .into_response();
487 }
488 let new_access_meta = match crate::auth::create_access_token_with_delegation(
489 &session_row.did,
490 &key_bytes,
491 session_row.scope.as_deref(),
492 session_row.controller_did.as_deref(),
493 None,
494 ) {
495 Ok(m) => m,
496 Err(e) => {
497 error!("Failed to create access token: {:?}", e);
498 return ApiError::InternalError(None).into_response();
499 }
500 };
501 let new_refresh_meta =
502 match crate::auth::create_refresh_token_with_metadata(&session_row.did, &key_bytes) {
503 Ok(m) => m,
504 Err(e) => {
505 error!("Failed to create refresh token: {:?}", e);
506 return ApiError::InternalError(None).into_response();
507 }
508 };
509 match sqlx::query!(
510 "INSERT INTO used_refresh_tokens (refresh_jti, session_id) VALUES ($1, $2) ON CONFLICT (refresh_jti) DO NOTHING",
511 refresh_jti,
512 session_row.id
513 )
514 .execute(&mut *tx)
515 .await
516 {
517 Ok(result) if result.rows_affected() == 0 => {
518 warn!("Concurrent refresh token reuse detected for session_id: {}", session_row.id);
519 let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_row.id)
520 .execute(&mut *tx)
521 .await;
522 let _ = tx.commit().await;
523 return ApiError::AuthenticationFailed(Some("Refresh token has been revoked due to suspected compromise".into())).into_response();
524 }
525 Err(e) => {
526 error!("Failed to record used refresh token: {:?}", e);
527 return ApiError::InternalError(None).into_response();
528 }
529 Ok(_) => {}
530 }
531 if let Err(e) = sqlx::query!(
532 "UPDATE session_tokens SET access_jti = $1, refresh_jti = $2, access_expires_at = $3, refresh_expires_at = $4, updated_at = NOW() WHERE id = $5",
533 new_access_meta.jti,
534 new_refresh_meta.jti,
535 new_access_meta.expires_at,
536 new_refresh_meta.expires_at,
537 session_row.id
538 )
539 .execute(&mut *tx)
540 .await
541 {
542 error!("Database error updating session: {:?}", e);
543 return ApiError::InternalError(None).into_response();
544 }
545 if let Err(e) = tx.commit().await {
546 error!("Failed to commit transaction: {:?}", e);
547 return ApiError::InternalError(None).into_response();
548 }
549 let did_for_doc = session_row.did.clone();
550 let did_resolver = state.did_resolver.clone();
551 let (db_result, did_doc) = tokio::join!(
552 sqlx::query!(
553 r#"SELECT
554 handle, email, email_verified, is_admin, preferred_locale, deactivated_at, takedown_ref,
555 preferred_comms_channel as "preferred_channel: crate::comms::CommsChannel",
556 discord_verified, telegram_verified, signal_verified
557 FROM users WHERE did = $1"#,
558 session_row.did
559 )
560 .fetch_optional(&state.db),
561 did_resolver.resolve_did_document(&did_for_doc)
562 );
563 match db_result {
564 Ok(Some(u)) => {
565 let (preferred_channel, preferred_channel_verified) = match u.preferred_channel {
566 crate::comms::CommsChannel::Email => ("email", u.email_verified),
567 crate::comms::CommsChannel::Discord => ("discord", u.discord_verified),
568 crate::comms::CommsChannel::Telegram => ("telegram", u.telegram_verified),
569 crate::comms::CommsChannel::Signal => ("signal", u.signal_verified),
570 };
571 let pds_hostname =
572 std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
573 let handle = full_handle(&u.handle, &pds_hostname);
574 let account_state =
575 AccountState::from_db_fields(u.deactivated_at, u.takedown_ref.clone(), None, None);
576 let mut response = json!({
577 "accessJwt": new_access_meta.token,
578 "refreshJwt": new_refresh_meta.token,
579 "handle": handle,
580 "did": session_row.did,
581 "email": u.email,
582 "emailConfirmed": u.email_verified,
583 "preferredChannel": preferred_channel,
584 "preferredChannelVerified": preferred_channel_verified,
585 "preferredLocale": u.preferred_locale,
586 "isAdmin": u.is_admin,
587 "active": account_state.is_active()
588 });
589 if let Some(doc) = did_doc {
590 response["didDoc"] = doc;
591 }
592 if let Some(status) = account_state.status_for_session() {
593 response["status"] = json!(status);
594 }
595 Json(response).into_response()
596 }
597 Ok(None) => {
598 error!("User not found for existing session: {}", session_row.did);
599 ApiError::InternalError(None).into_response()
600 }
601 Err(e) => {
602 error!("Database error fetching user: {:?}", e);
603 ApiError::InternalError(None).into_response()
604 }
605 }
606}
607
608#[derive(Deserialize)]
609#[serde(rename_all = "camelCase")]
610pub struct ConfirmSignupInput {
611 pub did: Did,
612 pub verification_code: String,
613}
614
615#[derive(Serialize)]
616#[serde(rename_all = "camelCase")]
617pub struct ConfirmSignupOutput {
618 pub access_jwt: String,
619 pub refresh_jwt: String,
620 pub handle: Handle,
621 pub did: Did,
622 pub email: Option<String>,
623 pub email_verified: bool,
624 pub preferred_channel: String,
625 pub preferred_channel_verified: bool,
626}
627
628pub async fn confirm_signup(
629 State(state): State<AppState>,
630 Json(input): Json<ConfirmSignupInput>,
631) -> Response {
632 info!("confirm_signup called for DID: {}", input.did);
633 let row = match sqlx::query!(
634 r#"SELECT
635 u.id, u.did, u.handle, u.email,
636 u.preferred_comms_channel as "channel: crate::comms::CommsChannel",
637 u.discord_id, u.telegram_username, u.signal_number,
638 k.key_bytes, k.encryption_version
639 FROM users u
640 JOIN user_keys k ON u.id = k.user_id
641 WHERE u.did = $1"#,
642 input.did.as_str()
643 )
644 .fetch_optional(&state.db)
645 .await
646 {
647 Ok(Some(row)) => row,
648 Ok(None) => {
649 warn!("User not found for confirm_signup: {}", input.did);
650 return ApiError::InvalidRequest("Invalid DID or verification code".into())
651 .into_response();
652 }
653 Err(e) => {
654 error!("Database error in confirm_signup: {:?}", e);
655 return ApiError::InternalError(None).into_response();
656 }
657 };
658
659 let (channel_str, identifier) = match row.channel {
660 crate::comms::CommsChannel::Email => ("email", row.email.clone().unwrap_or_default()),
661 crate::comms::CommsChannel::Discord => {
662 ("discord", row.discord_id.clone().unwrap_or_default())
663 }
664 crate::comms::CommsChannel::Telegram => (
665 "telegram",
666 row.telegram_username.clone().unwrap_or_default(),
667 ),
668 crate::comms::CommsChannel::Signal => {
669 ("signal", row.signal_number.clone().unwrap_or_default())
670 }
671 };
672
673 let normalized_token =
674 crate::auth::verification_token::normalize_token_input(&input.verification_code);
675 match crate::auth::verification_token::verify_signup_token(
676 &normalized_token,
677 channel_str,
678 &identifier,
679 ) {
680 Ok(token_data) => {
681 if token_data.did != input.did.as_str() {
682 warn!(
683 "Token DID mismatch for confirm_signup: expected {}, got {}",
684 input.did, token_data.did
685 );
686 return ApiError::InvalidRequest("Invalid verification code".into())
687 .into_response();
688 }
689 }
690 Err(crate::auth::verification_token::VerifyError::Expired) => {
691 warn!("Verification code expired for user: {}", input.did);
692 return ApiError::ExpiredToken(Some("Verification code has expired".into()))
693 .into_response();
694 }
695 Err(e) => {
696 warn!("Invalid verification code for user {}: {:?}", input.did, e);
697 return ApiError::InvalidRequest("Invalid verification code".into()).into_response();
698 }
699 }
700
701 let key_bytes = match crate::config::decrypt_key(&row.key_bytes, row.encryption_version) {
702 Ok(k) => k,
703 Err(e) => {
704 error!("Failed to decrypt user key: {:?}", e);
705 return ApiError::InternalError(None).into_response();
706 }
707 };
708 let verified_column = match row.channel {
709 crate::comms::CommsChannel::Email => "email_verified",
710 crate::comms::CommsChannel::Discord => "discord_verified",
711 crate::comms::CommsChannel::Telegram => "telegram_verified",
712 crate::comms::CommsChannel::Signal => "signal_verified",
713 };
714 let update_query = format!("UPDATE users SET {} = TRUE WHERE did = $1", verified_column);
715 if let Err(e) = sqlx::query(&update_query)
716 .bind(input.did.as_str())
717 .execute(&state.db)
718 .await
719 {
720 error!("Failed to update verification status: {:?}", e);
721 return ApiError::InternalError(None).into_response();
722 }
723
724 let access_meta = match crate::auth::create_access_token_with_metadata(&row.did, &key_bytes) {
725 Ok(m) => m,
726 Err(e) => {
727 error!("Failed to create access token: {:?}", e);
728 return ApiError::InternalError(None).into_response();
729 }
730 };
731 let refresh_meta = match crate::auth::create_refresh_token_with_metadata(&row.did, &key_bytes) {
732 Ok(m) => m,
733 Err(e) => {
734 error!("Failed to create refresh token: {:?}", e);
735 return ApiError::InternalError(None).into_response();
736 }
737 };
738 let no_scope: Option<String> = None;
739 if let Err(e) = sqlx::query!(
740 "INSERT INTO session_tokens (did, access_jti, refresh_jti, access_expires_at, refresh_expires_at, legacy_login, mfa_verified, scope) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
741 row.did,
742 access_meta.jti,
743 refresh_meta.jti,
744 access_meta.expires_at,
745 refresh_meta.expires_at,
746 false,
747 false,
748 no_scope
749 )
750 .execute(&state.db)
751 .await
752 {
753 error!("Failed to insert session: {:?}", e);
754 return ApiError::InternalError(None).into_response();
755 }
756 let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
757 if let Err(e) = crate::comms::enqueue_welcome(&state.db, row.id, &hostname).await {
758 warn!("Failed to enqueue welcome notification: {:?}", e);
759 }
760 let email_verified = matches!(row.channel, crate::comms::CommsChannel::Email);
761 let preferred_channel = match row.channel {
762 crate::comms::CommsChannel::Email => "email",
763 crate::comms::CommsChannel::Discord => "discord",
764 crate::comms::CommsChannel::Telegram => "telegram",
765 crate::comms::CommsChannel::Signal => "signal",
766 };
767 Json(ConfirmSignupOutput {
768 access_jwt: access_meta.token,
769 refresh_jwt: refresh_meta.token,
770 handle: row.handle.into(),
771 did: row.did.into(),
772 email: row.email,
773 email_verified,
774 preferred_channel: preferred_channel.to_string(),
775 preferred_channel_verified: true,
776 })
777 .into_response()
778}
779
780#[derive(Deserialize)]
781#[serde(rename_all = "camelCase")]
782pub struct ResendVerificationInput {
783 pub did: Did,
784}
785
786pub async fn resend_verification(
787 State(state): State<AppState>,
788 Json(input): Json<ResendVerificationInput>,
789) -> Response {
790 info!("resend_verification called for DID: {}", input.did);
791 let row = match sqlx::query!(
792 r#"SELECT
793 id, handle, email,
794 preferred_comms_channel as "channel: crate::comms::CommsChannel",
795 discord_id, telegram_username, signal_number,
796 email_verified, discord_verified, telegram_verified, signal_verified
797 FROM users
798 WHERE did = $1"#,
799 input.did.as_str()
800 )
801 .fetch_optional(&state.db)
802 .await
803 {
804 Ok(Some(row)) => row,
805 Ok(None) => {
806 return ApiError::InvalidRequest("User not found".into()).into_response();
807 }
808 Err(e) => {
809 error!("Database error in resend_verification: {:?}", e);
810 return ApiError::InternalError(None).into_response();
811 }
812 };
813 let is_verified =
814 row.email_verified || row.discord_verified || row.telegram_verified || row.signal_verified;
815 if is_verified {
816 return ApiError::InvalidRequest("Account is already verified".into()).into_response();
817 }
818
819 let (channel_str, recipient) = match row.channel {
820 crate::comms::CommsChannel::Email => ("email", row.email.clone().unwrap_or_default()),
821 crate::comms::CommsChannel::Discord => {
822 ("discord", row.discord_id.clone().unwrap_or_default())
823 }
824 crate::comms::CommsChannel::Telegram => (
825 "telegram",
826 row.telegram_username.clone().unwrap_or_default(),
827 ),
828 crate::comms::CommsChannel::Signal => {
829 ("signal", row.signal_number.clone().unwrap_or_default())
830 }
831 };
832
833 let verification_token =
834 crate::auth::verification_token::generate_signup_token(&input.did, channel_str, &recipient);
835 let formatted_token =
836 crate::auth::verification_token::format_token_for_display(&verification_token);
837
838 if let Err(e) = crate::comms::enqueue_signup_verification(
839 &state.db,
840 row.id,
841 channel_str,
842 &recipient,
843 &formatted_token,
844 None,
845 )
846 .await
847 {
848 warn!("Failed to enqueue verification notification: {:?}", e);
849 }
850 SuccessResponse::ok().into_response()
851}
852
853#[derive(Serialize)]
854#[serde(rename_all = "camelCase")]
855pub struct SessionInfo {
856 pub id: String,
857 pub session_type: String,
858 pub client_name: Option<String>,
859 pub created_at: String,
860 pub expires_at: String,
861 pub is_current: bool,
862}
863
864#[derive(Serialize)]
865#[serde(rename_all = "camelCase")]
866pub struct ListSessionsOutput {
867 pub sessions: Vec<SessionInfo>,
868}
869
870pub async fn list_sessions(
871 State(state): State<AppState>,
872 headers: HeaderMap,
873 auth: BearerAuth,
874) -> Response {
875 let current_jti = headers
876 .get("authorization")
877 .and_then(|v| v.to_str().ok())
878 .and_then(|v| v.strip_prefix("Bearer "))
879 .and_then(|token| crate::auth::get_jti_from_token(token).ok());
880
881 let mut sessions: Vec<SessionInfo> = Vec::new();
882
883 let jwt_result = sqlx::query_as::<
884 _,
885 (
886 i32,
887 String,
888 chrono::DateTime<chrono::Utc>,
889 chrono::DateTime<chrono::Utc>,
890 ),
891 >(
892 r#"
893 SELECT id, access_jti, created_at, refresh_expires_at
894 FROM session_tokens
895 WHERE did = $1 AND refresh_expires_at > NOW()
896 ORDER BY created_at DESC
897 "#,
898 )
899 .bind(&auth.0.did)
900 .fetch_all(&state.db)
901 .await;
902
903 match jwt_result {
904 Ok(rows) => {
905 for (id, access_jti, created_at, expires_at) in rows {
906 sessions.push(SessionInfo {
907 id: format!("jwt:{}", id),
908 session_type: "legacy".to_string(),
909 client_name: None,
910 created_at: created_at.to_rfc3339(),
911 expires_at: expires_at.to_rfc3339(),
912 is_current: current_jti.as_ref() == Some(&access_jti),
913 });
914 }
915 }
916 Err(e) => {
917 error!("DB error fetching JWT sessions: {:?}", e);
918 return ApiError::InternalError(None).into_response();
919 }
920 }
921
922 let oauth_result = sqlx::query_as::<
923 _,
924 (
925 i32,
926 String,
927 chrono::DateTime<chrono::Utc>,
928 chrono::DateTime<chrono::Utc>,
929 String,
930 ),
931 >(
932 r#"
933 SELECT id, token_id, created_at, expires_at, client_id
934 FROM oauth_token
935 WHERE did = $1 AND expires_at > NOW()
936 ORDER BY created_at DESC
937 "#,
938 )
939 .bind(&auth.0.did)
940 .fetch_all(&state.db)
941 .await;
942
943 match oauth_result {
944 Ok(rows) => {
945 for (id, token_id, created_at, expires_at, client_id) in rows {
946 let client_name = extract_client_name(&client_id);
947 let is_current_oauth = auth.0.is_oauth && current_jti.as_ref() == Some(&token_id);
948 sessions.push(SessionInfo {
949 id: format!("oauth:{}", id),
950 session_type: "oauth".to_string(),
951 client_name: Some(client_name),
952 created_at: created_at.to_rfc3339(),
953 expires_at: expires_at.to_rfc3339(),
954 is_current: is_current_oauth,
955 });
956 }
957 }
958 Err(e) => {
959 error!("DB error fetching OAuth sessions: {:?}", e);
960 return ApiError::InternalError(None).into_response();
961 }
962 }
963
964 sessions.sort_by(|a, b| b.created_at.cmp(&a.created_at));
965
966 (StatusCode::OK, Json(ListSessionsOutput { sessions })).into_response()
967}
968
969fn extract_client_name(client_id: &str) -> String {
970 if client_id.starts_with("http://localhost") || client_id.starts_with("http://127.0.0.1") {
971 "Localhost App".to_string()
972 } else if let Ok(parsed) = reqwest::Url::parse(client_id) {
973 parsed.host_str().unwrap_or("Unknown App").to_string()
974 } else {
975 client_id.to_string()
976 }
977}
978
979#[derive(Deserialize)]
980#[serde(rename_all = "camelCase")]
981pub struct RevokeSessionInput {
982 pub session_id: String,
983}
984
985pub async fn revoke_session(
986 State(state): State<AppState>,
987 auth: BearerAuth,
988 Json(input): Json<RevokeSessionInput>,
989) -> Response {
990 if let Some(jwt_id) = input.session_id.strip_prefix("jwt:") {
991 let Ok(session_id) = jwt_id.parse::<i32>() else {
992 return ApiError::InvalidRequest("Invalid session ID".into()).into_response();
993 };
994 let session = sqlx::query_as::<_, (String,)>(
995 "SELECT access_jti FROM session_tokens WHERE id = $1 AND did = $2",
996 )
997 .bind(session_id)
998 .bind(&auth.0.did)
999 .fetch_optional(&state.db)
1000 .await;
1001 let access_jti = match session {
1002 Ok(Some((jti,))) => jti,
1003 Ok(None) => {
1004 return ApiError::SessionNotFound.into_response();
1005 }
1006 Err(e) => {
1007 error!("DB error in revoke_session: {:?}", e);
1008 return ApiError::InternalError(None).into_response();
1009 }
1010 };
1011 if let Err(e) = sqlx::query("DELETE FROM session_tokens WHERE id = $1")
1012 .bind(session_id)
1013 .execute(&state.db)
1014 .await
1015 {
1016 error!("DB error deleting session: {:?}", e);
1017 return ApiError::InternalError(None).into_response();
1018 }
1019 let cache_key = format!("auth:session:{}:{}", &auth.0.did, access_jti);
1020 if let Err(e) = state.cache.delete(&cache_key).await {
1021 warn!("Failed to invalidate session cache: {:?}", e);
1022 }
1023 info!(did = %&auth.0.did, session_id = %session_id, "JWT session revoked");
1024 } else if let Some(oauth_id) = input.session_id.strip_prefix("oauth:") {
1025 let Ok(session_id) = oauth_id.parse::<i32>() else {
1026 return ApiError::InvalidRequest("Invalid session ID".into()).into_response();
1027 };
1028 let result = sqlx::query("DELETE FROM oauth_token WHERE id = $1 AND did = $2")
1029 .bind(session_id)
1030 .bind(&auth.0.did)
1031 .execute(&state.db)
1032 .await;
1033 match result {
1034 Ok(r) if r.rows_affected() == 0 => {
1035 return ApiError::SessionNotFound.into_response();
1036 }
1037 Err(e) => {
1038 error!("DB error deleting OAuth session: {:?}", e);
1039 return ApiError::InternalError(None).into_response();
1040 }
1041 _ => {}
1042 }
1043 info!(did = %&auth.0.did, session_id = %session_id, "OAuth session revoked");
1044 } else {
1045 return ApiError::InvalidRequest("Invalid session ID format".into()).into_response();
1046 }
1047 EmptyResponse::ok().into_response()
1048}
1049
1050pub async fn revoke_all_sessions(
1051 State(state): State<AppState>,
1052 headers: HeaderMap,
1053 auth: BearerAuth,
1054) -> Response {
1055 let current_jti = crate::auth::extract_auth_token_from_header(
1056 headers.get("authorization").and_then(|v| v.to_str().ok()),
1057 )
1058 .and_then(|extracted| crate::auth::get_jti_from_token(&extracted.token).ok());
1059
1060 let Some(ref jti) = current_jti else {
1061 return ApiError::InvalidToken(None).into_response();
1062 };
1063
1064 if auth.0.is_oauth {
1065 if let Err(e) = sqlx::query("DELETE FROM session_tokens WHERE did = $1")
1066 .bind(&auth.0.did)
1067 .execute(&state.db)
1068 .await
1069 {
1070 error!("DB error revoking JWT sessions: {:?}", e);
1071 return ApiError::InternalError(None).into_response();
1072 }
1073 if let Err(e) = sqlx::query("DELETE FROM oauth_token WHERE did = $1 AND token_id != $2")
1074 .bind(&auth.0.did)
1075 .bind(jti)
1076 .execute(&state.db)
1077 .await
1078 {
1079 error!("DB error revoking OAuth sessions: {:?}", e);
1080 return ApiError::InternalError(None).into_response();
1081 }
1082 } else {
1083 if let Err(e) =
1084 sqlx::query("DELETE FROM session_tokens WHERE did = $1 AND access_jti != $2")
1085 .bind(&auth.0.did)
1086 .bind(jti)
1087 .execute(&state.db)
1088 .await
1089 {
1090 error!("DB error revoking JWT sessions: {:?}", e);
1091 return ApiError::InternalError(None).into_response();
1092 }
1093 if let Err(e) = sqlx::query("DELETE FROM oauth_token WHERE did = $1")
1094 .bind(&auth.0.did)
1095 .execute(&state.db)
1096 .await
1097 {
1098 error!("DB error revoking OAuth sessions: {:?}", e);
1099 return ApiError::InternalError(None).into_response();
1100 }
1101 }
1102
1103 info!(did = %&auth.0.did, "All other sessions revoked");
1104 SuccessResponse::ok().into_response()
1105}
1106
1107#[derive(Serialize)]
1108#[serde(rename_all = "camelCase")]
1109pub struct LegacyLoginPreferenceOutput {
1110 pub allow_legacy_login: bool,
1111 pub has_mfa: bool,
1112}
1113
1114pub async fn get_legacy_login_preference(
1115 State(state): State<AppState>,
1116 auth: BearerAuth,
1117) -> Response {
1118 let result = sqlx::query!(
1119 r#"SELECT
1120 u.allow_legacy_login,
1121 (EXISTS(SELECT 1 FROM user_totp t WHERE t.did = u.did AND t.verified = TRUE) OR
1122 EXISTS(SELECT 1 FROM passkeys p WHERE p.did = u.did)) as "has_mfa!"
1123 FROM users u WHERE u.did = $1"#,
1124 &auth.0.did
1125 )
1126 .fetch_optional(&state.db)
1127 .await;
1128
1129 match result {
1130 Ok(Some(row)) => Json(LegacyLoginPreferenceOutput {
1131 allow_legacy_login: row.allow_legacy_login,
1132 has_mfa: row.has_mfa,
1133 })
1134 .into_response(),
1135 Ok(None) => ApiError::AccountNotFound.into_response(),
1136 Err(e) => {
1137 error!("DB error: {:?}", e);
1138 ApiError::InternalError(None).into_response()
1139 }
1140 }
1141}
1142
1143#[derive(Deserialize)]
1144#[serde(rename_all = "camelCase")]
1145pub struct UpdateLegacyLoginInput {
1146 pub allow_legacy_login: bool,
1147}
1148
1149pub async fn update_legacy_login_preference(
1150 State(state): State<AppState>,
1151 auth: BearerAuth,
1152 Json(input): Json<UpdateLegacyLoginInput>,
1153) -> Response {
1154 if !crate::api::server::reauth::check_legacy_session_mfa(&state.db, &auth.0.did).await {
1155 return crate::api::server::reauth::legacy_mfa_required_response(&state.db, &auth.0.did)
1156 .await;
1157 }
1158
1159 if crate::api::server::reauth::check_reauth_required(&state.db, &auth.0.did).await {
1160 return crate::api::server::reauth::reauth_required_response(&state.db, &auth.0.did).await;
1161 }
1162
1163 let result = sqlx::query!(
1164 "UPDATE users SET allow_legacy_login = $1 WHERE did = $2 RETURNING did",
1165 input.allow_legacy_login,
1166 &auth.0.did
1167 )
1168 .fetch_optional(&state.db)
1169 .await;
1170
1171 match result {
1172 Ok(Some(_)) => {
1173 info!(
1174 did = %&auth.0.did,
1175 allow_legacy_login = input.allow_legacy_login,
1176 "Legacy login preference updated"
1177 );
1178 Json(json!({
1179 "allowLegacyLogin": input.allow_legacy_login
1180 }))
1181 .into_response()
1182 }
1183 Ok(None) => ApiError::AccountNotFound.into_response(),
1184 Err(e) => {
1185 error!("DB error: {:?}", e);
1186 ApiError::InternalError(None).into_response()
1187 }
1188 }
1189}
1190
1191use crate::comms::VALID_LOCALES;
1192
1193#[derive(Deserialize)]
1194#[serde(rename_all = "camelCase")]
1195pub struct UpdateLocaleInput {
1196 pub preferred_locale: String,
1197}
1198
1199pub async fn update_locale(
1200 State(state): State<AppState>,
1201 auth: BearerAuth,
1202 Json(input): Json<UpdateLocaleInput>,
1203) -> Response {
1204 if !VALID_LOCALES.contains(&input.preferred_locale.as_str()) {
1205 return ApiError::InvalidRequest(format!(
1206 "Invalid locale. Valid options: {}",
1207 VALID_LOCALES.join(", ")
1208 ))
1209 .into_response();
1210 }
1211
1212 let result = sqlx::query!(
1213 "UPDATE users SET preferred_locale = $1 WHERE did = $2 RETURNING did",
1214 input.preferred_locale,
1215 &auth.0.did
1216 )
1217 .fetch_optional(&state.db)
1218 .await;
1219
1220 match result {
1221 Ok(Some(_)) => {
1222 info!(
1223 did = %&auth.0.did,
1224 locale = %input.preferred_locale,
1225 "User locale preference updated"
1226 );
1227 Json(json!({
1228 "preferredLocale": input.preferred_locale
1229 }))
1230 .into_response()
1231 }
1232 Ok(None) => ApiError::AccountNotFound.into_response(),
1233 Err(e) => {
1234 error!("DB error updating locale: {:?}", e);
1235 ApiError::InternalError(None).into_response()
1236 }
1237 }
1238}