src/api/server/session.rs at dcdb8e28875321be58c77534dd73b6e32816751f · lewis.moe/bspds-sandbox

lewis.moe / bspds-sandbox
this repo has no description
bspds-sandbox / src / api / server / session.rs
at dcdb8e28875321be58c77534dd73b6e32816751f 44 kB view raw
   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    ) {
 216        Ok(m) => m,
 217        Err(e) => {
 218            error!("Failed to create access token: {:?}", e);
 219            return ApiError::InternalError(None).into_response();
 220        }
 221    };
 222    let refresh_meta = match crate::auth::create_refresh_token_with_metadata(&row.did, &key_bytes) {
 223        Ok(m) => m,
 224        Err(e) => {
 225            error!("Failed to create refresh token: {:?}", e);
 226            return ApiError::InternalError(None).into_response();
 227        }
 228    };
 229    let did_for_doc = row.did.clone();
 230    let did_resolver = state.did_resolver.clone();
 231    let (insert_result, did_doc) = tokio::join!(
 232        sqlx::query!(
 233            "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)",
 234            row.did,
 235            access_meta.jti,
 236            refresh_meta.jti,
 237            access_meta.expires_at,
 238            refresh_meta.expires_at,
 239            is_legacy_login,
 240            false,
 241            app_password_scopes,
 242            app_password_controller,
 243            app_password_name
 244        )
 245        .execute(&state.db),
 246        did_resolver.resolve_did_document(&did_for_doc)
 247    );
 248    if let Err(e) = insert_result {
 249        error!("Failed to insert session: {:?}", e);
 250        return ApiError::InternalError(None).into_response();
 251    }
 252    if is_legacy_login {
 253        warn!(
 254            did = %row.did,
 255            ip = %client_ip,
 256            "Legacy login on TOTP-enabled account - sending notification"
 257        );
 258        let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
 259        if let Err(e) = crate::comms::queue_legacy_login_notification(
 260            &state.db,
 261            row.id,
 262            &hostname,
 263            &client_ip,
 264            row.preferred_comms_channel,
 265        )
 266        .await
 267        {
 268            error!("Failed to queue legacy login notification: {:?}", e);
 269        }
 270    }
 271    let handle = full_handle(&row.handle, &pds_hostname);
 272    let is_active = account_state.is_active();
 273    let status = account_state.status_for_session().map(String::from);
 274    Json(CreateSessionOutput {
 275        access_jwt: access_meta.token,
 276        refresh_jwt: refresh_meta.token,
 277        handle: handle.into(),
 278        did: row.did.into(),
 279        did_doc,
 280        email: row.email,
 281        email_confirmed: Some(row.email_verified),
 282        active: Some(is_active),
 283        status,
 284    })
 285    .into_response()
 286}
 287
 288pub async fn get_session(
 289    State(state): State<AppState>,
 290    BearerAuthAllowDeactivated(auth_user): BearerAuthAllowDeactivated,
 291) -> Response {
 292    let permissions = auth_user.permissions();
 293    let can_read_email = permissions.allows_email_read();
 294
 295    let did_for_doc = auth_user.did.clone();
 296    let did_resolver = state.did_resolver.clone();
 297    let (db_result, did_doc) = tokio::join!(
 298        sqlx::query!(
 299            r#"SELECT
 300                handle, email, email_verified, is_admin, deactivated_at, takedown_ref, preferred_locale,
 301                preferred_comms_channel as "preferred_channel: crate::comms::CommsChannel",
 302                discord_verified, telegram_verified, signal_verified, migrated_to_pds, migrated_at
 303            FROM users WHERE did = $1"#,
 304            &auth_user.did
 305        )
 306        .fetch_optional(&state.db),
 307        did_resolver.resolve_did_document(&did_for_doc)
 308    );
 309    match db_result {
 310        Ok(Some(row)) => {
 311            let (preferred_channel, preferred_channel_verified) = match row.preferred_channel {
 312                crate::comms::CommsChannel::Email => ("email", row.email_verified),
 313                crate::comms::CommsChannel::Discord => ("discord", row.discord_verified),
 314                crate::comms::CommsChannel::Telegram => ("telegram", row.telegram_verified),
 315                crate::comms::CommsChannel::Signal => ("signal", row.signal_verified),
 316            };
 317            let pds_hostname =
 318                std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
 319            let handle = full_handle(&row.handle, &pds_hostname);
 320            let account_state = AccountState::from_db_fields(
 321                row.deactivated_at,
 322                row.takedown_ref.clone(),
 323                row.migrated_to_pds.clone(),
 324                row.migrated_at,
 325            );
 326            let email_value = if can_read_email {
 327                row.email.clone()
 328            } else {
 329                None
 330            };
 331            let email_confirmed_value = can_read_email && row.email_verified;
 332            let mut response = json!({
 333                "handle": handle,
 334                "did": &auth_user.did,
 335                "active": account_state.is_active(),
 336                "preferredChannel": preferred_channel,
 337                "preferredChannelVerified": preferred_channel_verified,
 338                "preferredLocale": row.preferred_locale,
 339                "isAdmin": row.is_admin
 340            });
 341            if can_read_email {
 342                response["email"] = json!(email_value);
 343                response["emailConfirmed"] = json!(email_confirmed_value);
 344            }
 345            if let Some(status) = account_state.status_for_session() {
 346                response["status"] = json!(status);
 347            }
 348            if let AccountState::Migrated { to_pds, at } = &account_state {
 349                response["migratedToPds"] = json!(to_pds);
 350                response["migratedAt"] = json!(at);
 351            }
 352            if let Some(doc) = did_doc {
 353                response["didDoc"] = doc;
 354            }
 355            Json(response).into_response()
 356        }
 357        Ok(None) => ApiError::AuthenticationFailed(None).into_response(),
 358        Err(e) => {
 359            error!("Database error in get_session: {:?}", e);
 360            ApiError::InternalError(None).into_response()
 361        }
 362    }
 363}
 364
 365pub async fn delete_session(
 366    State(state): State<AppState>,
 367    headers: axum::http::HeaderMap,
 368) -> Response {
 369    let token = match crate::auth::extract_bearer_token_from_header(
 370        headers.get("Authorization").and_then(|h| h.to_str().ok()),
 371    ) {
 372        Some(t) => t,
 373        None => return ApiError::AuthenticationRequired.into_response(),
 374    };
 375    let jti = match crate::auth::get_jti_from_token(&token) {
 376        Ok(jti) => jti,
 377        Err(_) => return ApiError::AuthenticationFailed(None).into_response(),
 378    };
 379    let did = crate::auth::get_did_from_token(&token).ok();
 380    match sqlx::query!("DELETE FROM session_tokens WHERE access_jti = $1", jti)
 381        .execute(&state.db)
 382        .await
 383    {
 384        Ok(res) if res.rows_affected() > 0 => {
 385            if let Some(did) = did {
 386                let session_cache_key = format!("auth:session:{}:{}", did, jti);
 387                let _ = state.cache.delete(&session_cache_key).await;
 388            }
 389            EmptyResponse::ok().into_response()
 390        }
 391        Ok(_) => ApiError::AuthenticationFailed(None).into_response(),
 392        Err(e) => {
 393            error!("Database error in delete_session: {:?}", e);
 394            ApiError::AuthenticationFailed(None).into_response()
 395        }
 396    }
 397}
 398
 399pub async fn refresh_session(
 400    State(state): State<AppState>,
 401    headers: axum::http::HeaderMap,
 402) -> Response {
 403    let client_ip = crate::rate_limit::extract_client_ip(&headers, None);
 404    if !state
 405        .check_rate_limit(RateLimitKind::RefreshSession, &client_ip)
 406        .await
 407    {
 408        tracing::warn!(ip = %client_ip, "Refresh session rate limit exceeded");
 409        return ApiError::RateLimitExceeded(None).into_response();
 410    }
 411    let refresh_token = match crate::auth::extract_bearer_token_from_header(
 412        headers.get("Authorization").and_then(|h| h.to_str().ok()),
 413    ) {
 414        Some(t) => t,
 415        None => return ApiError::AuthenticationRequired.into_response(),
 416    };
 417    let refresh_jti = match crate::auth::get_jti_from_token(&refresh_token) {
 418        Ok(jti) => jti,
 419        Err(_) => {
 420            return ApiError::AuthenticationFailed(Some("Invalid token format".into()))
 421                .into_response();
 422        }
 423    };
 424    let mut tx = match state.db.begin().await {
 425        Ok(tx) => tx,
 426        Err(e) => {
 427            error!("Failed to begin transaction: {:?}", e);
 428            return ApiError::InternalError(None).into_response();
 429        }
 430    };
 431    if let Ok(Some(session_id)) = sqlx::query_scalar!(
 432        "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1 FOR UPDATE",
 433        refresh_jti
 434    )
 435    .fetch_optional(&mut *tx)
 436    .await
 437    {
 438        warn!(
 439            "Refresh token reuse detected! Revoking token family for session_id: {}",
 440            session_id
 441        );
 442        let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_id)
 443            .execute(&mut *tx)
 444            .await;
 445        let _ = tx.commit().await;
 446        return ApiError::AuthenticationFailed(Some(
 447            "Refresh token has been revoked due to suspected compromise".into(),
 448        ))
 449        .into_response();
 450    }
 451    let session_row = match sqlx::query!(
 452        r#"SELECT st.id, st.did, st.scope, st.controller_did, k.key_bytes, k.encryption_version
 453           FROM session_tokens st
 454           JOIN users u ON st.did = u.did
 455           JOIN user_keys k ON u.id = k.user_id
 456           WHERE st.refresh_jti = $1 AND st.refresh_expires_at > NOW()
 457           FOR UPDATE OF st"#,
 458        refresh_jti
 459    )
 460    .fetch_optional(&mut *tx)
 461    .await
 462    {
 463        Ok(Some(row)) => row,
 464        Ok(None) => {
 465            return ApiError::AuthenticationFailed(Some("Invalid refresh token".into()))
 466                .into_response();
 467        }
 468        Err(e) => {
 469            error!("Database error fetching session: {:?}", e);
 470            return ApiError::InternalError(None).into_response();
 471        }
 472    };
 473    let key_bytes =
 474        match crate::config::decrypt_key(&session_row.key_bytes, session_row.encryption_version) {
 475            Ok(k) => k,
 476            Err(e) => {
 477                error!("Failed to decrypt user key: {:?}", e);
 478                return ApiError::InternalError(None).into_response();
 479            }
 480        };
 481    if crate::auth::verify_refresh_token(&refresh_token, &key_bytes).is_err() {
 482        return ApiError::AuthenticationFailed(Some("Invalid refresh token".into())).into_response();
 483    }
 484    let new_access_meta = match crate::auth::create_access_token_with_delegation(
 485        &session_row.did,
 486        &key_bytes,
 487        session_row.scope.as_deref(),
 488        session_row.controller_did.as_deref(),
 489    ) {
 490        Ok(m) => m,
 491        Err(e) => {
 492            error!("Failed to create access token: {:?}", e);
 493            return ApiError::InternalError(None).into_response();
 494        }
 495    };
 496    let new_refresh_meta =
 497        match crate::auth::create_refresh_token_with_metadata(&session_row.did, &key_bytes) {
 498            Ok(m) => m,
 499            Err(e) => {
 500                error!("Failed to create refresh token: {:?}", e);
 501                return ApiError::InternalError(None).into_response();
 502            }
 503        };
 504    match sqlx::query!(
 505        "INSERT INTO used_refresh_tokens (refresh_jti, session_id) VALUES ($1, $2) ON CONFLICT (refresh_jti) DO NOTHING",
 506        refresh_jti,
 507        session_row.id
 508    )
 509    .execute(&mut *tx)
 510    .await
 511    {
 512        Ok(result) if result.rows_affected() == 0 => {
 513            warn!("Concurrent refresh token reuse detected for session_id: {}", session_row.id);
 514            let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_row.id)
 515                .execute(&mut *tx)
 516                .await;
 517            let _ = tx.commit().await;
 518            return ApiError::AuthenticationFailed(Some("Refresh token has been revoked due to suspected compromise".into())).into_response();
 519        }
 520        Err(e) => {
 521            error!("Failed to record used refresh token: {:?}", e);
 522            return ApiError::InternalError(None).into_response();
 523        }
 524        Ok(_) => {}
 525    }
 526    if let Err(e) = sqlx::query!(
 527        "UPDATE session_tokens SET access_jti = $1, refresh_jti = $2, access_expires_at = $3, refresh_expires_at = $4, updated_at = NOW() WHERE id = $5",
 528        new_access_meta.jti,
 529        new_refresh_meta.jti,
 530        new_access_meta.expires_at,
 531        new_refresh_meta.expires_at,
 532        session_row.id
 533    )
 534    .execute(&mut *tx)
 535    .await
 536    {
 537        error!("Database error updating session: {:?}", e);
 538        return ApiError::InternalError(None).into_response();
 539    }
 540    if let Err(e) = tx.commit().await {
 541        error!("Failed to commit transaction: {:?}", e);
 542        return ApiError::InternalError(None).into_response();
 543    }
 544    let did_for_doc = session_row.did.clone();
 545    let did_resolver = state.did_resolver.clone();
 546    let (db_result, did_doc) = tokio::join!(
 547        sqlx::query!(
 548            r#"SELECT
 549                handle, email, email_verified, is_admin, preferred_locale, deactivated_at, takedown_ref,
 550                preferred_comms_channel as "preferred_channel: crate::comms::CommsChannel",
 551                discord_verified, telegram_verified, signal_verified
 552            FROM users WHERE did = $1"#,
 553            session_row.did
 554        )
 555        .fetch_optional(&state.db),
 556        did_resolver.resolve_did_document(&did_for_doc)
 557    );
 558    match db_result {
 559        Ok(Some(u)) => {
 560            let (preferred_channel, preferred_channel_verified) = match u.preferred_channel {
 561                crate::comms::CommsChannel::Email => ("email", u.email_verified),
 562                crate::comms::CommsChannel::Discord => ("discord", u.discord_verified),
 563                crate::comms::CommsChannel::Telegram => ("telegram", u.telegram_verified),
 564                crate::comms::CommsChannel::Signal => ("signal", u.signal_verified),
 565            };
 566            let pds_hostname =
 567                std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
 568            let handle = full_handle(&u.handle, &pds_hostname);
 569            let account_state = AccountState::from_db_fields(
 570                u.deactivated_at,
 571                u.takedown_ref.clone(),
 572                None,
 573                None,
 574            );
 575            let mut response = json!({
 576                "accessJwt": new_access_meta.token,
 577                "refreshJwt": new_refresh_meta.token,
 578                "handle": handle,
 579                "did": session_row.did,
 580                "email": u.email,
 581                "emailConfirmed": u.email_verified,
 582                "preferredChannel": preferred_channel,
 583                "preferredChannelVerified": preferred_channel_verified,
 584                "preferredLocale": u.preferred_locale,
 585                "isAdmin": u.is_admin,
 586                "active": account_state.is_active()
 587            });
 588            if let Some(doc) = did_doc {
 589                response["didDoc"] = doc;
 590            }
 591            if let Some(status) = account_state.status_for_session() {
 592                response["status"] = json!(status);
 593            }
 594            Json(response).into_response()
 595        }
 596        Ok(None) => {
 597            error!("User not found for existing session: {}", session_row.did);
 598            ApiError::InternalError(None).into_response()
 599        }
 600        Err(e) => {
 601            error!("Database error fetching user: {:?}", e);
 602            ApiError::InternalError(None).into_response()
 603        }
 604    }
 605}
 606
 607#[derive(Deserialize)]
 608#[serde(rename_all = "camelCase")]
 609pub struct ConfirmSignupInput {
 610    pub did: Did,
 611    pub verification_code: String,
 612}
 613
 614#[derive(Serialize)]
 615#[serde(rename_all = "camelCase")]
 616pub struct ConfirmSignupOutput {
 617    pub access_jwt: String,
 618    pub refresh_jwt: String,
 619    pub handle: Handle,
 620    pub did: Did,
 621    pub email: Option<String>,
 622    pub email_verified: bool,
 623    pub preferred_channel: String,
 624    pub preferred_channel_verified: bool,
 625}
 626
 627pub async fn confirm_signup(
 628    State(state): State<AppState>,
 629    Json(input): Json<ConfirmSignupInput>,
 630) -> Response {
 631    info!("confirm_signup called for DID: {}", input.did);
 632    let row = match sqlx::query!(
 633        r#"SELECT
 634            u.id, u.did, u.handle, u.email,
 635            u.preferred_comms_channel as "channel: crate::comms::CommsChannel",
 636            u.discord_id, u.telegram_username, u.signal_number,
 637            k.key_bytes, k.encryption_version
 638        FROM users u
 639        JOIN user_keys k ON u.id = k.user_id
 640        WHERE u.did = $1"#,
 641        input.did.as_str()
 642    )
 643    .fetch_optional(&state.db)
 644    .await
 645    {
 646        Ok(Some(row)) => row,
 647        Ok(None) => {
 648            warn!("User not found for confirm_signup: {}", input.did);
 649            return ApiError::InvalidRequest("Invalid DID or verification code".into())
 650                .into_response();
 651        }
 652        Err(e) => {
 653            error!("Database error in confirm_signup: {:?}", e);
 654            return ApiError::InternalError(None).into_response();
 655        }
 656    };
 657
 658    let (channel_str, identifier) = match row.channel {
 659        crate::comms::CommsChannel::Email => ("email", row.email.clone().unwrap_or_default()),
 660        crate::comms::CommsChannel::Discord => {
 661            ("discord", row.discord_id.clone().unwrap_or_default())
 662        }
 663        crate::comms::CommsChannel::Telegram => (
 664            "telegram",
 665            row.telegram_username.clone().unwrap_or_default(),
 666        ),
 667        crate::comms::CommsChannel::Signal => {
 668            ("signal", row.signal_number.clone().unwrap_or_default())
 669        }
 670    };
 671
 672    let normalized_token =
 673        crate::auth::verification_token::normalize_token_input(&input.verification_code);
 674    match crate::auth::verification_token::verify_signup_token(
 675        &normalized_token,
 676        channel_str,
 677        &identifier,
 678    ) {
 679        Ok(token_data) => {
 680            if token_data.did != input.did.as_str() {
 681                warn!(
 682                    "Token DID mismatch for confirm_signup: expected {}, got {}",
 683                    input.did, token_data.did
 684                );
 685                return ApiError::InvalidRequest("Invalid verification code".into())
 686                    .into_response();
 687            }
 688        }
 689        Err(crate::auth::verification_token::VerifyError::Expired) => {
 690            warn!("Verification code expired for user: {}", input.did);
 691            return ApiError::ExpiredToken(Some("Verification code has expired".into()))
 692                .into_response();
 693        }
 694        Err(e) => {
 695            warn!("Invalid verification code for user {}: {:?}", input.did, e);
 696            return ApiError::InvalidRequest("Invalid verification code".into()).into_response();
 697        }
 698    }
 699
 700    let key_bytes = match crate::config::decrypt_key(&row.key_bytes, row.encryption_version) {
 701        Ok(k) => k,
 702        Err(e) => {
 703            error!("Failed to decrypt user key: {:?}", e);
 704            return ApiError::InternalError(None).into_response();
 705        }
 706    };
 707    let verified_column = match row.channel {
 708        crate::comms::CommsChannel::Email => "email_verified",
 709        crate::comms::CommsChannel::Discord => "discord_verified",
 710        crate::comms::CommsChannel::Telegram => "telegram_verified",
 711        crate::comms::CommsChannel::Signal => "signal_verified",
 712    };
 713    let update_query = format!("UPDATE users SET {} = TRUE WHERE did = $1", verified_column);
 714    if let Err(e) = sqlx::query(&update_query)
 715        .bind(input.did.as_str())
 716        .execute(&state.db)
 717        .await
 718    {
 719        error!("Failed to update verification status: {:?}", e);
 720        return ApiError::InternalError(None).into_response();
 721    }
 722
 723    let access_meta = match crate::auth::create_access_token_with_metadata(&row.did, &key_bytes) {
 724        Ok(m) => m,
 725        Err(e) => {
 726            error!("Failed to create access token: {:?}", e);
 727            return ApiError::InternalError(None).into_response();
 728        }
 729    };
 730    let refresh_meta = match crate::auth::create_refresh_token_with_metadata(&row.did, &key_bytes) {
 731        Ok(m) => m,
 732        Err(e) => {
 733            error!("Failed to create refresh token: {:?}", e);
 734            return ApiError::InternalError(None).into_response();
 735        }
 736    };
 737    let no_scope: Option<String> = None;
 738    if let Err(e) = sqlx::query!(
 739        "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)",
 740        row.did,
 741        access_meta.jti,
 742        refresh_meta.jti,
 743        access_meta.expires_at,
 744        refresh_meta.expires_at,
 745        false,
 746        false,
 747        no_scope
 748    )
 749    .execute(&state.db)
 750    .await
 751    {
 752        error!("Failed to insert session: {:?}", e);
 753        return ApiError::InternalError(None).into_response();
 754    }
 755    let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
 756    if let Err(e) = crate::comms::enqueue_welcome(&state.db, row.id, &hostname).await {
 757        warn!("Failed to enqueue welcome notification: {:?}", e);
 758    }
 759    let email_verified = matches!(row.channel, crate::comms::CommsChannel::Email);
 760    let preferred_channel = match row.channel {
 761        crate::comms::CommsChannel::Email => "email",
 762        crate::comms::CommsChannel::Discord => "discord",
 763        crate::comms::CommsChannel::Telegram => "telegram",
 764        crate::comms::CommsChannel::Signal => "signal",
 765    };
 766    Json(ConfirmSignupOutput {
 767        access_jwt: access_meta.token,
 768        refresh_jwt: refresh_meta.token,
 769        handle: row.handle.into(),
 770        did: row.did.into(),
 771        email: row.email,
 772        email_verified,
 773        preferred_channel: preferred_channel.to_string(),
 774        preferred_channel_verified: true,
 775    })
 776    .into_response()
 777}
 778
 779#[derive(Deserialize)]
 780#[serde(rename_all = "camelCase")]
 781pub struct ResendVerificationInput {
 782    pub did: Did,
 783}
 784
 785pub async fn resend_verification(
 786    State(state): State<AppState>,
 787    Json(input): Json<ResendVerificationInput>,
 788) -> Response {
 789    info!("resend_verification called for DID: {}", input.did);
 790    let row = match sqlx::query!(
 791        r#"SELECT
 792            id, handle, email,
 793            preferred_comms_channel as "channel: crate::comms::CommsChannel",
 794            discord_id, telegram_username, signal_number,
 795            email_verified, discord_verified, telegram_verified, signal_verified
 796        FROM users
 797        WHERE did = $1"#,
 798        input.did.as_str()
 799    )
 800    .fetch_optional(&state.db)
 801    .await
 802    {
 803        Ok(Some(row)) => row,
 804        Ok(None) => {
 805            return ApiError::InvalidRequest("User not found".into()).into_response();
 806        }
 807        Err(e) => {
 808            error!("Database error in resend_verification: {:?}", e);
 809            return ApiError::InternalError(None).into_response();
 810        }
 811    };
 812    let is_verified =
 813        row.email_verified || row.discord_verified || row.telegram_verified || row.signal_verified;
 814    if is_verified {
 815        return ApiError::InvalidRequest("Account is already verified".into()).into_response();
 816    }
 817
 818    let (channel_str, recipient) = match row.channel {
 819        crate::comms::CommsChannel::Email => ("email", row.email.clone().unwrap_or_default()),
 820        crate::comms::CommsChannel::Discord => {
 821            ("discord", row.discord_id.clone().unwrap_or_default())
 822        }
 823        crate::comms::CommsChannel::Telegram => (
 824            "telegram",
 825            row.telegram_username.clone().unwrap_or_default(),
 826        ),
 827        crate::comms::CommsChannel::Signal => {
 828            ("signal", row.signal_number.clone().unwrap_or_default())
 829        }
 830    };
 831
 832    let verification_token =
 833        crate::auth::verification_token::generate_signup_token(&input.did, channel_str, &recipient);
 834    let formatted_token =
 835        crate::auth::verification_token::format_token_for_display(&verification_token);
 836
 837    if let Err(e) = crate::comms::enqueue_signup_verification(
 838        &state.db,
 839        row.id,
 840        channel_str,
 841        &recipient,
 842        &formatted_token,
 843        None,
 844    )
 845    .await
 846    {
 847        warn!("Failed to enqueue verification notification: {:?}", e);
 848    }
 849    SuccessResponse::ok().into_response()
 850}
 851
 852#[derive(Serialize)]
 853#[serde(rename_all = "camelCase")]
 854pub struct SessionInfo {
 855    pub id: String,
 856    pub session_type: String,
 857    pub client_name: Option<String>,
 858    pub created_at: String,
 859    pub expires_at: String,
 860    pub is_current: bool,
 861}
 862
 863#[derive(Serialize)]
 864#[serde(rename_all = "camelCase")]
 865pub struct ListSessionsOutput {
 866    pub sessions: Vec<SessionInfo>,
 867}
 868
 869pub async fn list_sessions(
 870    State(state): State<AppState>,
 871    headers: HeaderMap,
 872    auth: BearerAuth,
 873) -> Response {
 874    let current_jti = headers
 875        .get("authorization")
 876        .and_then(|v| v.to_str().ok())
 877        .and_then(|v| v.strip_prefix("Bearer "))
 878        .and_then(|token| crate::auth::get_jti_from_token(token).ok());
 879
 880    let mut sessions: Vec<SessionInfo> = Vec::new();
 881
 882    let jwt_result = sqlx::query_as::<
 883        _,
 884        (
 885            i32,
 886            String,
 887            chrono::DateTime<chrono::Utc>,
 888            chrono::DateTime<chrono::Utc>,
 889        ),
 890    >(
 891        r#"
 892        SELECT id, access_jti, created_at, refresh_expires_at
 893        FROM session_tokens
 894        WHERE did = $1 AND refresh_expires_at > NOW()
 895        ORDER BY created_at DESC
 896        "#,
 897    )
 898    .bind(&auth.0.did)
 899    .fetch_all(&state.db)
 900    .await;
 901
 902    match jwt_result {
 903        Ok(rows) => {
 904            for (id, access_jti, created_at, expires_at) in rows {
 905                sessions.push(SessionInfo {
 906                    id: format!("jwt:{}", id),
 907                    session_type: "legacy".to_string(),
 908                    client_name: None,
 909                    created_at: created_at.to_rfc3339(),
 910                    expires_at: expires_at.to_rfc3339(),
 911                    is_current: current_jti.as_ref() == Some(&access_jti),
 912                });
 913            }
 914        }
 915        Err(e) => {
 916            error!("DB error fetching JWT sessions: {:?}", e);
 917            return ApiError::InternalError(None).into_response();
 918        }
 919    }
 920
 921    let oauth_result = sqlx::query_as::<
 922        _,
 923        (
 924            i32,
 925            String,
 926            chrono::DateTime<chrono::Utc>,
 927            chrono::DateTime<chrono::Utc>,
 928            String,
 929        ),
 930    >(
 931        r#"
 932        SELECT id, token_id, created_at, expires_at, client_id
 933        FROM oauth_token
 934        WHERE did = $1 AND expires_at > NOW()
 935        ORDER BY created_at DESC
 936        "#,
 937    )
 938    .bind(&auth.0.did)
 939    .fetch_all(&state.db)
 940    .await;
 941
 942    match oauth_result {
 943        Ok(rows) => {
 944            for (id, token_id, created_at, expires_at, client_id) in rows {
 945                let client_name = extract_client_name(&client_id);
 946                let is_current_oauth = auth.0.is_oauth && current_jti.as_ref() == Some(&token_id);
 947                sessions.push(SessionInfo {
 948                    id: format!("oauth:{}", id),
 949                    session_type: "oauth".to_string(),
 950                    client_name: Some(client_name),
 951                    created_at: created_at.to_rfc3339(),
 952                    expires_at: expires_at.to_rfc3339(),
 953                    is_current: is_current_oauth,
 954                });
 955            }
 956        }
 957        Err(e) => {
 958            error!("DB error fetching OAuth sessions: {:?}", e);
 959            return ApiError::InternalError(None).into_response();
 960        }
 961    }
 962
 963    sessions.sort_by(|a, b| b.created_at.cmp(&a.created_at));
 964
 965    (StatusCode::OK, Json(ListSessionsOutput { sessions })).into_response()
 966}
 967
 968fn extract_client_name(client_id: &str) -> String {
 969    if client_id.starts_with("http://localhost") || client_id.starts_with("http://127.0.0.1") {
 970        "Localhost App".to_string()
 971    } else if let Ok(parsed) = reqwest::Url::parse(client_id) {
 972        parsed.host_str().unwrap_or("Unknown App").to_string()
 973    } else {
 974        client_id.to_string()
 975    }
 976}
 977
 978#[derive(Deserialize)]
 979#[serde(rename_all = "camelCase")]
 980pub struct RevokeSessionInput {
 981    pub session_id: String,
 982}
 983
 984pub async fn revoke_session(
 985    State(state): State<AppState>,
 986    auth: BearerAuth,
 987    Json(input): Json<RevokeSessionInput>,
 988) -> Response {
 989    if let Some(jwt_id) = input.session_id.strip_prefix("jwt:") {
 990        let Ok(session_id) = jwt_id.parse::<i32>() else {
 991            return ApiError::InvalidRequest("Invalid session ID".into()).into_response();
 992        };
 993        let session = sqlx::query_as::<_, (String,)>(
 994            "SELECT access_jti FROM session_tokens WHERE id = $1 AND did = $2",
 995        )
 996        .bind(session_id)
 997        .bind(&auth.0.did)
 998        .fetch_optional(&state.db)
 999        .await;
1000        let access_jti = match session {
1001            Ok(Some((jti,))) => jti,
1002            Ok(None) => {
1003                return ApiError::SessionNotFound.into_response();
1004            }
1005            Err(e) => {
1006                error!("DB error in revoke_session: {:?}", e);
1007                return ApiError::InternalError(None).into_response();
1008            }
1009        };
1010        if let Err(e) = sqlx::query("DELETE FROM session_tokens WHERE id = $1")
1011            .bind(session_id)
1012            .execute(&state.db)
1013            .await
1014        {
1015            error!("DB error deleting session: {:?}", e);
1016            return ApiError::InternalError(None).into_response();
1017        }
1018        let cache_key = format!("auth:session:{}:{}", &auth.0.did, access_jti);
1019        if let Err(e) = state.cache.delete(&cache_key).await {
1020            warn!("Failed to invalidate session cache: {:?}", e);
1021        }
1022        info!(did = %&auth.0.did, session_id = %session_id, "JWT session revoked");
1023    } else if let Some(oauth_id) = input.session_id.strip_prefix("oauth:") {
1024        let Ok(session_id) = oauth_id.parse::<i32>() else {
1025            return ApiError::InvalidRequest("Invalid session ID".into()).into_response();
1026        };
1027        let result = sqlx::query("DELETE FROM oauth_token WHERE id = $1 AND did = $2")
1028            .bind(session_id)
1029            .bind(&auth.0.did)
1030            .execute(&state.db)
1031            .await;
1032        match result {
1033            Ok(r) if r.rows_affected() == 0 => {
1034                return ApiError::SessionNotFound.into_response();
1035            }
1036            Err(e) => {
1037                error!("DB error deleting OAuth session: {:?}", e);
1038                return ApiError::InternalError(None).into_response();
1039            }
1040            _ => {}
1041        }
1042        info!(did = %&auth.0.did, session_id = %session_id, "OAuth session revoked");
1043    } else {
1044        return ApiError::InvalidRequest("Invalid session ID format".into()).into_response();
1045    }
1046    EmptyResponse::ok().into_response()
1047}
1048
1049pub async fn revoke_all_sessions(
1050    State(state): State<AppState>,
1051    headers: HeaderMap,
1052    auth: BearerAuth,
1053) -> Response {
1054    let current_jti = headers
1055        .get("authorization")
1056        .and_then(|v| v.to_str().ok())
1057        .and_then(|v| v.strip_prefix("Bearer "))
1058        .and_then(|token| crate::auth::get_jti_from_token(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::locale::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}