Something went wrong. Try again.
Our Personal Data Server from scratch!
Something went wrong. Try again.
47 kB · 1364 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365use axum::{ Json, extract::State, http::{HeaderMap, StatusCode}, response::{IntoResponse, Response},};use bcrypt::verify;use serde::{Deserialize, Serialize};use serde_json::json;use tracing::{error, info, warn};use tranquil_db_traits::{ChannelVerificationStatus, SessionId, TokenFamilyId};use tranquil_pds::api::error::{ApiError, DbResultExt};use tranquil_pds::api::{EmptyResponse, PreferredLocaleOutput, SuccessResponse};use tranquil_pds::auth::{ Active, Auth, NormalizedLoginIdentifier, Permissive, require_legacy_session_mfa, require_reauth_window,};use tranquil_pds::rate_limit::{ LoginLimit, RateLimited, RefreshSessionLimit, TotpVerifyLimit, check_user_rate_limit_with_message,};use tranquil_pds::state::AppState;use tranquil_pds::types::{AccountState, AtIdentifier, Did, Handle, PlainPassword};use tranquil_types::TokenId;
pub fn verification_blocks_login(channel_verification: &ChannelVerificationStatus) -> bool { !tranquil_config::get() .server .disable_account_verification_gate && !channel_verification.has_any_verified()}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct CreateSessionInput { pub identifier: String, pub password: PlainPassword, #[serde(default)] pub allow_takendown: bool, pub auth_factor_token: Option<String>,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct CreateSessionOutput { pub access_jwt: String, pub refresh_jwt: String, pub handle: Handle, pub did: Did, #[serde(skip_serializing_if = "Option::is_none")] pub did_doc: Option<serde_json::Value>, #[serde(skip_serializing_if = "Option::is_none")] pub email: Option<String>, #[serde(skip_serializing_if = "Option::is_none")] pub email_confirmed: Option<bool>, #[serde(skip_serializing_if = "Option::is_none")] pub email_auth_factor: Option<bool>, #[serde(skip_serializing_if = "Option::is_none")] pub active: Option<bool>, #[serde(skip_serializing_if = "Option::is_none")] pub status: Option<String>,}
pub async fn create_session( State(state): State<AppState>, rate_limit: RateLimited<LoginLimit>, Json(input): Json<CreateSessionInput>,) -> Result<Response, ApiError> { let client_ip = rate_limit.client_ip(); info!( "create_session called with identifier: {}", input.identifier ); let hostname_for_handles = tranquil_config::get().server.hostname_without_port(); let normalized_identifier = NormalizedLoginIdentifier::normalize(&input.identifier, hostname_for_handles); info!( "Normalized identifier: {} -> {}", input.identifier, normalized_identifier ); let Ok(login_identifier) = AtIdentifier::new(normalized_identifier.as_str()) else { let _ = verify( &input.password, "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYw1ZzQKZqmK", ); warn!("Login identifier is not a valid handle or DID"); return Err(ApiError::AuthenticationFailed(Some( "Invalid identifier or password".into(), ))); }; let row = match state .repos .user .get_login_full_by_identifier(&login_identifier) .await { Ok(Some(row)) => row, Ok(None) => { let _ = verify( &input.password, "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYw1ZzQKZqmK", ); warn!("User not found for login attempt"); return Err(ApiError::AuthenticationFailed(Some( "Invalid identifier or password".into(), ))); } Err(e) => { error!("Database error fetching user: {:?}", e); return Err(ApiError::InternalError(None)); } }; let key_bytes = match tranquil_pds::config::decrypt_key(&row.key_bytes, row.encryption_version) { Ok(k) => k, Err(e) => { error!("Failed to decrypt user key: {:?}", e); return Err(ApiError::InternalError(None)); } }; let credential = crate::common::verify_credential( state.repos.session.as_ref(), row.id, &input.password, row.password_hash.as_ref(), ) .await; let (app_password_name, app_password_scopes, app_password_controller) = match credential { Some(crate::common::CredentialMatch::MainPassword) => (None, None, None), Some(crate::common::CredentialMatch::AppPassword { name, scopes, controller_did, }) => (Some(name), scopes, controller_did), None => { warn!("Password verification failed for login attempt"); return Err(ApiError::AuthenticationFailed(Some( "Invalid identifier or password".into(), ))); } }; let account_state = AccountState::from_db_fields( row.deactivated_at, row.takedown_ref.clone(), row.migrated_to_pds.clone(), None, ); if account_state.is_takendown() && !input.allow_takendown { warn!("Login attempt for takendown account: {}", row.did); return Err(ApiError::AccountTakedown); } let is_delegated = state .repos .delegation .is_delegated_account(&row.did) .await .unwrap_or(false); if verification_blocks_login(&row.channel_verification) && !is_delegated { warn!("Login attempt for unverified account: {}", row.did); let resend_info = auto_resend_verification(&state, &row.did).await; let handle = resend_info .as_ref() .map(|r| r.handle.to_string()) .unwrap_or_else(|| row.handle.to_string()); let channel = resend_info .as_ref() .map(|r| r.channel.as_str()) .unwrap_or(row.preferred_comms_channel.as_str()); return Ok(( StatusCode::FORBIDDEN, Json(json!({ "error": "account_not_verified", "message": "Please verify your account before logging in", "did": row.did, "handle": handle, "channel": channel })), ) .into_response()); } let has_totp = row.totp_enabled; let email_2fa_enabled = row.email_2fa_enabled; let is_legacy_login = has_totp || email_2fa_enabled; let used_totp_factor = tranquil_pds::auth::legacy_2fa::used_totp_factor( has_totp, input.auth_factor_token.as_deref(), ); let twofa_ctx = tranquil_pds::auth::legacy_2fa::Legacy2faContext { is_app_password: app_password_name.is_some(), email_2fa_enabled, has_totp, allow_legacy_login: row.allow_legacy_login, }; if let Some(token) = input.auth_factor_token.as_deref() && has_totp && tranquil_pds::auth::legacy_2fa::looks_like_totp_token(token) { check_user_rate_limit_with_message::<TotpVerifyLimit>( &state, row.did.as_str(), "Too many verification attempts. Please try again in a few minutes.", ) .await?; } match tranquil_pds::auth::legacy_2fa::process_legacy_2fa( state.cache.as_ref(), &row.did, &twofa_ctx, input.auth_factor_token.as_deref(), async |t: &str| { crate::server::totp::verify_totp_or_backup_for_user(&state, &row.did, t).await }, ) .await { Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::NotRequired) => {} Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::Blocked) => { warn!("Legacy login blocked for TOTP-enabled account: {}", row.did); return Err(ApiError::LegacyLoginBlocked); } Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::ChallengeSent(code)) => { let hostname = &tranquil_config::get().server.hostname; match tranquil_pds::comms::comms_repo::enqueue_notice( state.repos.user.as_ref(), state.repos.infra.as_ref(), row.id, tranquil_pds::comms::Notice::TwoFactorCode { code: code.as_str(), }, hostname, ) .await { Ok(Some(_)) => {} Ok(None) => { tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did) .await; return Err(ApiError::InvalidRequest( "We couldn't deliver the verification code to your notification channels. Please contact the PDS owner." .into(), )); } Err(e) => { error!("Failed to send 2FA code: {:?}", e); tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did) .await; return Err(ApiError::InternalError(Some( "Failed to send verification code. Please try again.".into(), ))); } } return Err(ApiError::AuthFactorTokenRequired); } Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::Verified) => {} Err(tranquil_pds::auth::legacy_2fa::Legacy2faFlowError::Challenge(e)) => { use tranquil_pds::auth::legacy_2fa::ChallengeError; return match e { ChallengeError::CacheUnavailable => { error!("Cache unavailable for 2FA, blocking legacy login"); Err(ApiError::ServiceUnavailable(Some( "2FA service temporarily unavailable. Please try again later or use an OAuth client.".into(), ))) } ChallengeError::RateLimited => Err(ApiError::RateLimitExceeded(Some( "Please wait before requesting a new verification code.".into(), ))), ChallengeError::CacheError => { error!("Cache error during 2FA challenge creation"); Err(ApiError::InternalError(None)) } }; } Err(tranquil_pds::auth::legacy_2fa::Legacy2faFlowError::Validation(e)) => { use tranquil_pds::auth::legacy_2fa::ValidationError; warn!("Invalid 2FA code for {}: {:?}", row.did, e); let msg = match e { ValidationError::TooManyAttempts => "Too many attempts. Please request a new code.", ValidationError::ChallengeExpired => "Code has expired. Please request a new code.", ValidationError::CacheUnavailable => { "2FA service temporarily unavailable. Please try again later." } ValidationError::ChallengeNotFound | ValidationError::InvalidCode | ValidationError::CacheError => "Invalid verification code", }; return Err(ApiError::InvalidCode(Some(msg.into()))); } } let access_meta = match tranquil_pds::auth::create_access_token_with_delegation( &row.did, &key_bytes, app_password_scopes.as_deref(), app_password_controller.as_ref(), None, ) { Ok(m) => m, Err(e) => { error!("Failed to create access token: {:?}", e); return Err(ApiError::InternalError(None)); } }; let refresh_meta = match tranquil_pds::auth::create_refresh_token_with_metadata(&row.did, &key_bytes) { Ok(m) => m, Err(e) => { error!("Failed to create refresh token: {:?}", e); return Err(ApiError::InternalError(None)); } }; let did_for_doc = row.did.clone(); let did_resolver = state.did_resolver.clone(); let session_data = tranquil_db_traits::SessionTokenCreate { did: row.did.clone(), access_jti: access_meta.jti.clone(), refresh_jti: refresh_meta.jti.clone(), access_expires_at: access_meta.expires_at, refresh_expires_at: refresh_meta.expires_at, login_type: tranquil_db_traits::LoginType::from_legacy_flag(is_legacy_login), mfa_verified: false, scope: app_password_scopes.clone(), controller_did: app_password_controller.clone(), app_password_name: app_password_name.clone(), }; let (insert_result, did_doc) = tokio::join!( state.repos.session.create_session(&session_data), did_resolver.fetch_did_document(&did_for_doc), ); if let Err(e) = insert_result { error!("Failed to insert session: {:?}", e); return Err(ApiError::InternalError(None)); } if is_legacy_login && !used_totp_factor { let alerts_enabled = match state.repos.infra.get_account_preferences(row.id).await { Ok(prefs) => prefs .iter() .find(|(name, _)| name == "legacy_login_alerts") .and_then(|(_, value)| value.as_bool()) .unwrap_or(true), Err(e) => { warn!("Failed to fetch legacy login alert preference: {:?}", e); true } };
if alerts_enabled { warn!( did = %row.did, ip = %client_ip, "Legacy login on TOTP-enabled account - sending notification" ); let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice( state.repos.user.as_ref(), state.repos.infra.as_ref(), row.id, tranquil_pds::comms::Notice::LegacyLoginAlert { channel: row.preferred_comms_channel, ip: client_ip, }, hostname, ) .await { error!("Failed to queue legacy login notification: {:?}", e); } } } let handle = row.handle.clone(); let is_active = account_state.is_active(); let status = account_state.status_for_session().map(String::from); let email_auth_factor_out = if email_2fa_enabled || has_totp { Some(true) } else { None }; Ok(( StatusCode::OK, Json(CreateSessionOutput { access_jwt: access_meta.token, refresh_jwt: refresh_meta.token, handle, did: row.did, did_doc: did_doc.ok(), email: row.email, email_confirmed: Some(row.channel_verification.email), email_auth_factor: email_auth_factor_out, active: Some(is_active), status, }), ) .into_response())}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct GetSessionOutput { pub handle: Handle, pub did: Did, pub active: bool, pub preferred_channel: String, pub preferred_channel_verified: bool, #[serde(skip_serializing_if = "Option::is_none")] pub preferred_locale: Option<String>, pub is_admin: bool, #[serde(skip_serializing_if = "Option::is_none")] pub email: Option<String>, #[serde(skip_serializing_if = "Option::is_none")] pub email_confirmed: Option<bool>, #[serde(skip_serializing_if = "Option::is_none")] pub email_auth_factor: Option<bool>, #[serde(skip_serializing_if = "Option::is_none")] pub status: Option<String>, #[serde(skip_serializing_if = "Option::is_none")] pub migrated_to_pds: Option<String>, #[serde(skip_serializing_if = "Option::is_none")] pub migrated_at: Option<chrono::DateTime<chrono::Utc>>, #[serde(skip_serializing_if = "Option::is_none")] pub did_doc: Option<serde_json::Value>,}
pub async fn get_session( State(state): State<AppState>, auth: Auth<Permissive>,) -> Result<Json<GetSessionOutput>, ApiError> { let permissions = auth.permissions(); let can_read_email = permissions.allows_email_read();
let did_for_doc = auth.did.clone(); let did_resolver = state.did_resolver.clone(); let (db_result, did_doc) = tokio::join!( state.repos.user.get_session_info_by_did(&auth.did), did_resolver.fetch_did_document(&did_for_doc) ); match db_result { Ok(Some(row)) => { let preferred_channel_verified = row .channel_verification .is_verified(row.preferred_comms_channel); let handle = row.handle.clone(); let account_state = AccountState::from_db_fields( row.deactivated_at, row.takedown_ref.clone(), row.migrated_to_pds.clone(), row.migrated_at, ); let email = match can_read_email { true => row.email.clone(), false => None, }; let email_confirmed = match can_read_email { true => Some(row.channel_verification.email), false => None, }; let email_auth_factor = match row.email_2fa_enabled || row.totp_enabled { true => Some(true), false => None, }; let (migrated_to_pds, migrated_at) = match &account_state { AccountState::Migrated { to_pds, at } => (Some(to_pds.clone()), Some(*at)), _ => (None, None), }; Ok(Json(GetSessionOutput { handle, did: auth.did.clone(), active: account_state.is_active(), preferred_channel: row.preferred_comms_channel.as_str().to_string(), preferred_channel_verified, preferred_locale: row.preferred_locale, is_admin: row.is_admin, email, email_confirmed, email_auth_factor, status: account_state.status_for_session().map(String::from), migrated_to_pds, migrated_at, did_doc: did_doc.ok(), })) } Ok(None) => Err(ApiError::AuthenticationFailed(None)), Err(e) => { error!("Database error in get_session: {:?}", e); Err(ApiError::InternalError(None)) } }}
pub async fn delete_session( State(state): State<AppState>, headers: axum::http::HeaderMap, auth: Auth<Active>,) -> Result<Json<EmptyResponse>, ApiError> { let jti = tranquil_pds::auth::extract_jti_from_headers(&headers) .ok_or(ApiError::AuthenticationRequired)?; match state .repos .session .delete_session_by_access_jti(&jti, &auth.did) .await { Ok(rows) if rows > 0 => { let session_cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &jti); let _ = state.cache.delete(&session_cache_key).await; Ok(Json(EmptyResponse {})) } Ok(_) => Err(ApiError::AuthenticationFailed(None)), Err(_) => Err(ApiError::AuthenticationFailed(None)), }}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct RefreshSessionOutput { pub access_jwt: String, pub refresh_jwt: String, pub handle: Handle, pub did: Did, #[serde(skip_serializing_if = "Option::is_none")] pub email: Option<String>, pub email_confirmed: bool, pub preferred_channel: String, pub preferred_channel_verified: bool, #[serde(skip_serializing_if = "Option::is_none")] pub preferred_locale: Option<String>, pub is_admin: bool, pub active: bool, #[serde(skip_serializing_if = "Option::is_none")] pub did_doc: Option<serde_json::Value>, #[serde(skip_serializing_if = "Option::is_none")] pub status: Option<String>,}
pub async fn refresh_session( State(state): State<AppState>, _rate_limit: RateLimited<RefreshSessionLimit>, headers: axum::http::HeaderMap,) -> Result<Json<RefreshSessionOutput>, ApiError> { let extracted = match tranquil_pds::auth::extract_auth_token_from_header( tranquil_pds::util::get_header_str(&headers, http::header::AUTHORIZATION), ) { Some(t) => t, None => return Err(ApiError::AuthenticationRequired), }; let refresh_token = extracted.token; let refresh_jti = match tranquil_pds::auth::get_jti_from_token(&refresh_token) { Ok(jti) => jti, Err(_) => { return Err(ApiError::AuthenticationFailed(Some( "Invalid token format".into(), ))); } }; if let Some(result) = dispatch_refresh_grace( &state, &refresh_token, &refresh_jti, state.repos.session.lookup_refresh_grace(&refresh_jti).await, ) .await { return result; } let session_row = match state .repos .session .get_session_for_refresh(&refresh_jti) .await { Ok(Some(row)) => row, Ok(None) => { return dispatch_refresh_grace( &state, &refresh_token, &refresh_jti, state.repos.session.lookup_refresh_grace(&refresh_jti).await, ) .await .unwrap_or_else(|| { Err(ApiError::AuthenticationFailed(Some( "Invalid refresh token".into(), ))) }); } Err(e) => { error!("Database error fetching session: {:?}", e); return Err(ApiError::InternalError(None)); } }; let key_bytes = match tranquil_pds::config::decrypt_key( &session_row.key_bytes, Some(session_row.encryption_version), ) { Ok(k) => k, Err(e) => { error!("Failed to decrypt user key: {:?}", e); return Err(ApiError::InternalError(None)); } }; if tranquil_pds::auth::verify_refresh_token(&refresh_token, &key_bytes).is_err() { return Err(ApiError::AuthenticationFailed(Some( "Invalid refresh token".into(), ))); } let new_access_meta = match tranquil_pds::auth::create_access_token_with_delegation( &session_row.did, &key_bytes, session_row.scope.as_deref(), session_row.controller_did.as_ref(), None, ) { Ok(m) => m, Err(e) => { error!("Failed to create access token: {:?}", e); return Err(ApiError::InternalError(None)); } }; let new_refresh_meta = match tranquil_pds::auth::create_refresh_token_with_metadata( &session_row.did, &key_bytes, ) { Ok(m) => m, Err(e) => { error!("Failed to create refresh token: {:?}", e); return Err(ApiError::InternalError(None)); } }; let refresh_data = tranquil_db_traits::SessionRefreshData { did: session_row.did.clone(), old_refresh_jti: refresh_jti.clone(), session_id: session_row.id, new_access_jti: new_access_meta.jti.clone(), new_refresh_jti: new_refresh_meta.jti.clone(), new_access_expires_at: new_access_meta.expires_at, new_refresh_expires_at: new_refresh_meta.expires_at, }; let (access_jwt, refresh_jwt) = match state .repos .session .refresh_session_atomic(&refresh_data) .await { Ok(tranquil_db_traits::RefreshSessionResult::Success) => { (new_access_meta.token, new_refresh_meta.token) } Ok(tranquil_db_traits::RefreshSessionResult::GraceReplay(replay)) => { // Lost a benign concurrent rotation; re-mint the winner's tokens // using this session's signing key (same user, unchanged by rotation). info!( "Concurrent refresh within grace window for session_id: {}; replaying tokens", session_row.id ); remint_grace_tokens(&replay, &key_bytes)? } Ok(tranquil_db_traits::RefreshSessionResult::Compromise) => { warn!( "Refresh token reuse outside grace window or unreplayable rotation conflict for session_id: {}", session_row.id ); return Err(ApiError::AuthenticationFailed(Some( "Refresh token has been revoked due to suspected compromise".into(), ))); } Err(e) => { error!("Database error during session refresh: {:?}", e); return Err(ApiError::InternalError(None)); } }; build_refresh_session_output(&state, session_row.did, access_jwt, refresh_jwt).await}
async fn dispatch_refresh_grace( state: &AppState, refresh_token: &str, presented_jti: &str, lookup: Result<tranquil_db_traits::RefreshGraceLookup, tranquil_db_traits::DbError>,) -> Option<Result<Json<RefreshSessionOutput>, ApiError>> { match lookup { Ok(tranquil_db_traits::RefreshGraceLookup::NotUsed) => None, Ok(tranquil_db_traits::RefreshGraceLookup::Replay(replay)) => { Some(serve_refresh_grace_replay(state, refresh_token, presented_jti, replay).await) } Ok(tranquil_db_traits::RefreshGraceLookup::Compromised { did, session_id, key_bytes, encryption_version, }) => Some(Err(revoke_compromised_session( state, refresh_token, presented_jti, did, session_id, key_bytes, encryption_version, ) .await)), Err(e) => { error!("Database error checking refresh token grace: {:?}", e); Some(Err(ApiError::InternalError(None))) } }}
async fn serve_refresh_grace_replay( state: &AppState, refresh_token: &str, presented_jti: &str, replay: tranquil_db_traits::RefreshGraceReplay,) -> Result<Json<RefreshSessionOutput>, ApiError> { let key = match tranquil_pds::config::decrypt_key(&replay.key_bytes, Some(replay.encryption_version)) { Ok(k) => k, Err(e) => { error!("Failed to decrypt user key for grace replay: {:?}", e); return Err(ApiError::InternalError(None)); } }; if tranquil_pds::auth::verify_refresh_token(refresh_token, &key).is_err() { return Err(ApiError::AuthenticationFailed(Some( "Invalid refresh token".into(), ))); } info!("Refresh token reuse within grace window for jti: {presented_jti}; replaying tokens"); let (access_jwt, refresh_jwt) = remint_grace_tokens(&replay, &key)?; build_refresh_session_output(state, replay.did, access_jwt, refresh_jwt).await}
async fn revoke_compromised_session( state: &AppState, refresh_token: &str, presented_jti: &str, did: Did, session_id: SessionId, key_bytes: Vec<u8>, encryption_version: i32,) -> ApiError { let key = match tranquil_pds::config::decrypt_key(&key_bytes, Some(encryption_version)) { Ok(k) => k, Err(e) => { error!("Failed to decrypt user key for grace check: {:?}", e); return ApiError::InternalError(None); } }; if tranquil_pds::auth::verify_refresh_token(refresh_token, &key).is_err() { return ApiError::AuthenticationFailed(Some("Invalid refresh token".into())); } warn!("Refresh token reuse outside grace window for jti: {presented_jti}; revoking session"); if let Err(e) = state .repos .session .delete_session_by_id(session_id, &did) .await { error!( "Failed to revoke session {} for refresh token reuse: {:?}", session_id.as_i32(), e ); return ApiError::InternalError(None); } ApiError::AuthenticationFailed(Some( "Refresh token has been revoked due to suspected compromise".into(), ))}
/// Re-mint the access/refresh JWTs for a grace-window replay from the session's/// current jtis and signing key. We never persist the signed JWTs; they are/// reconstructed on demand so a benignly-racing client converges on the same/// credentials the winning rotation produced. `key_bytes` is the owning user's/// already-decrypted signing key.fn remint_grace_tokens( replay: &tranquil_db_traits::RefreshGraceReplay, key_bytes: &[u8],) -> Result<(String, String), ApiError> { let access_jwt = tranquil_pds::auth::create_access_token_with_jti( &replay.did, key_bytes, replay.scope.as_deref(), replay.controller_did.as_ref(), None, &replay.access_jti, replay.access_expires_at, ) .map_err(|e| { error!("Failed to re-mint access token for grace replay: {:?}", e); ApiError::InternalError(None) })?; let refresh_jwt = tranquil_pds::auth::create_refresh_token_with_jti( &replay.did, key_bytes, &replay.refresh_jti, replay.refresh_expires_at, ) .map_err(|e| { error!("Failed to re-mint refresh token for grace replay: {:?}", e); ApiError::InternalError(None) })?; Ok((access_jwt, refresh_jwt))}
async fn build_refresh_session_output( state: &AppState, did: Did, access_jwt: String, refresh_jwt: String,) -> Result<Json<RefreshSessionOutput>, ApiError> { let did_resolver = state.did_resolver.clone(); let (db_result, did_doc) = tokio::join!( state.repos.user.get_session_info_by_did(&did), did_resolver.fetch_did_document(&did) ); match db_result { Ok(Some(u)) => { let preferred_channel_verified = u .channel_verification .is_verified(u.preferred_comms_channel); let handle = u.handle.clone(); let account_state = AccountState::from_db_fields(u.deactivated_at, u.takedown_ref.clone(), None, None); Ok(Json(RefreshSessionOutput { access_jwt, refresh_jwt, handle, did, email: u.email, email_confirmed: u.channel_verification.email, preferred_channel: u.preferred_comms_channel.as_str().to_string(), preferred_channel_verified, preferred_locale: u.preferred_locale, is_admin: u.is_admin, active: account_state.is_active(), did_doc: did_doc.ok(), status: account_state.status_for_session().map(String::from), })) } Ok(None) => { error!("User not found for existing session: {}", did); Err(ApiError::InternalError(None)) } Err(e) => { error!("Database error fetching user: {:?}", e); Err(ApiError::InternalError(None)) } }}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct ConfirmSignupInput { pub did: Did, pub verification_code: String,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct ConfirmSignupOutput { pub access_jwt: String, pub refresh_jwt: String, pub handle: Handle, pub did: Did, pub email: Option<String>, pub email_verified: bool, pub preferred_channel: tranquil_db_traits::CommsChannel, pub preferred_channel_verified: bool,}
pub async fn confirm_signup( State(state): State<AppState>, Json(input): Json<ConfirmSignupInput>,) -> Result<Json<ConfirmSignupOutput>, ApiError> { info!("confirm_signup called for DID: {}", input.did); let row = match state.repos.user.get_confirm_signup_by_did(&input.did).await { Ok(Some(row)) => row, Ok(None) => { warn!("User not found for confirm_signup: {}", input.did); return Err(ApiError::InvalidRequest( "Invalid DID or verification code".into(), )); } Err(e) => { error!("Database error in confirm_signup: {:?}", e); return Err(ApiError::InternalError(None)); } };
let Some(id) = row.channel_identifier() else { warn!( did = %input.did, channel = ?row.channel, "We can't confirm signup because the account doesn't have an identifier on file" ); return Err(ApiError::InvalidRequest("Invalid verification code".into())); };
let normalized_token = tranquil_pds::auth::verification_token::normalize_token_input(&input.verification_code); match tranquil_pds::auth::verification_token::verify_signup_token( &normalized_token, row.channel, id, ) { Ok(token_data) => { if token_data.did != input.did { warn!( "Token DID mismatch for confirm_signup: expected {}, got {}", input.did, token_data.did ); return Err(ApiError::InvalidRequest("Invalid verification code".into())); } } Err(tranquil_pds::auth::verification_token::VerifyError::Expired) => { warn!("Verification code expired for user: {}", input.did); return Err(ApiError::ExpiredToken(Some( "Verification code has expired".into(), ))); } Err(e) => { warn!("Invalid verification code for user {}: {:?}", input.did, e); return Err(ApiError::InvalidRequest("Invalid verification code".into())); } }
let key_bytes = match tranquil_pds::config::decrypt_key(&row.key_bytes, row.encryption_version) { Ok(k) => k, Err(e) => { error!("Failed to decrypt user key: {:?}", e); return Err(ApiError::InternalError(None)); } };
if let Err(e) = state .repos .user .set_channel_verified(&input.did, row.channel) .await { error!("Failed to update verification status: {:?}", e); return Err(ApiError::InternalError(None)); }
let session = match crate::identity::provision::create_and_store_session( &state, &row.did, &key_bytes, "transition:generic transition:chat.bsky", None, ) .await { Ok(s) => s, Err(_) => return Err(ApiError::InternalError(None)), };
let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice( state.repos.user.as_ref(), state.repos.infra.as_ref(), row.id, tranquil_pds::comms::Notice::Welcome, hostname, ) .await { warn!("Failed to enqueue welcome notification: {:?}", e); } Ok(Json(ConfirmSignupOutput { access_jwt: session.access_jwt, refresh_jwt: session.refresh_jwt, handle: row.handle, did: row.did, email: row.email, email_verified: matches!(row.channel, tranquil_db_traits::CommsChannel::Email), preferred_channel: row.channel, preferred_channel_verified: true, }))}
async fn resend_signup_verification( state: &AppState, row: &tranquil_db_traits::UserResendVerification, did: &Did, context: &'static str,) -> bool { let Some(id) = row.channel_identifier() else { warn!(did = %did, channel = ?row.channel, "We skipped {context} because the account doesn't have a recipient on file"); return false; }; match tranquil_pds::comms::VerificationTarget::resolve(row.channel, id, row.email.as_deref()) { Ok(target) => { crate::identity::provision::enqueue_signup_verification( state, row.id, did, row.channel, &target, ) .await; true } Err(_) => { warn!(did = %did, channel = ?row.channel, "We skipped {context} because the account doesn't have a valid recipient"); false } }}
const AUTO_VERIFY_DEBOUNCE: std::time::Duration = std::time::Duration::from_secs(120);
pub struct AutoResendResult { pub handle: tranquil_types::Handle, pub channel: tranquil_db_traits::CommsChannel,}
pub async fn auto_resend_verification(state: &AppState, did: &Did) -> Option<AutoResendResult> { let debounce_key = tranquil_pds::cache_keys::auto_verify_sent_key(did); let debounced = state.cache.get(&debounce_key).await.is_some(); let row = match state.repos.user.get_resend_verification_by_did(did).await { Ok(Some(row)) => row, Ok(None) => return None, Err(e) => { warn!( "Failed to fetch resend verification info for {}: {:?}", did, e ); return None; } }; if row.channel_verification.has_any_verified() { return None; } let result = AutoResendResult { handle: row.handle.clone(), channel: row.channel, }; if row.channel.verifies_via_bot() || debounced { return Some(result); } if resend_signup_verification(state, &row, did, "auto-resend verification").await { let _ = state .cache .set(&debounce_key, "1", AUTO_VERIFY_DEBOUNCE) .await; } Some(result)}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct ResendVerificationInput { pub did: Did,}
pub async fn resend_verification( State(state): State<AppState>, Json(input): Json<ResendVerificationInput>,) -> Result<Json<SuccessResponse>, ApiError> { info!("resend_verification called for DID: {}", input.did); let row = match state .repos .user .get_resend_verification_by_did(&input.did) .await { Ok(Some(row)) => row, Ok(None) => { return Err(ApiError::InvalidRequest("User not found".into())); } Err(e) => { error!("Database error in resend_verification: {:?}", e); return Err(ApiError::InternalError(None)); } }; if row.channel_verification.has_any_verified() { return Err(ApiError::InvalidRequest( "Account is already verified".into(), )); } resend_signup_verification(&state, &row, &input.did, "resend verification").await; Ok(Json(SuccessResponse { success: true }))}
#[derive(Serialize)]#[serde(rename_all = "lowercase")]pub enum SessionType { Legacy, OAuth,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct SessionInfo { pub id: String, pub session_type: SessionType, pub client_name: Option<String>, pub created_at: String, pub expires_at: String, pub is_current: bool,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct ListSessionsOutput { pub sessions: Vec<SessionInfo>,}
pub async fn list_sessions( State(state): State<AppState>, headers: HeaderMap, auth: Auth<Active>,) -> Result<Json<ListSessionsOutput>, ApiError> { let current_jti = tranquil_pds::auth::extract_jti_from_headers(&headers);
let jwt_rows = state .repos .session .list_sessions_by_did(&auth.did) .await .log_db_err("fetching JWT sessions")?;
let oauth_rows = state .repos .oauth .list_sessions_by_did(&auth.did) .await .log_db_err("fetching OAuth sessions")?;
let jwt_sessions = jwt_rows.into_iter().map(|row| SessionInfo { id: format!("jwt:{}", row.id), session_type: SessionType::Legacy, client_name: None, created_at: row.created_at.to_rfc3339(), expires_at: row.refresh_expires_at.to_rfc3339(), is_current: current_jti.as_ref() == Some(&row.access_jti), });
let is_oauth = auth.is_oauth(); let oauth_sessions = oauth_rows.into_iter().map(|row| { let client_name = extract_client_name(&row.client_id); let is_current_oauth = is_oauth && current_jti.as_deref() == Some(row.token_id.as_str()); SessionInfo { id: format!("oauth:{}", row.id), session_type: SessionType::OAuth, client_name: Some(client_name), created_at: row.created_at.to_rfc3339(), expires_at: row.expires_at.to_rfc3339(), is_current: is_current_oauth, } });
let mut sessions: Vec<SessionInfo> = jwt_sessions.chain(oauth_sessions).collect(); sessions.sort_by(|a, b| b.created_at.cmp(&a.created_at));
Ok(Json(ListSessionsOutput { sessions }))}
fn extract_client_name(client_id: &tranquil_types::ClientId) -> String { if client_id.starts_with("http://localhost") || client_id.starts_with("http://127.0.0.1") { "Localhost App".to_string() } else if let Ok(parsed) = reqwest::Url::parse(client_id.as_str()) { parsed.host_str().unwrap_or("Unknown App").to_string() } else { client_id.as_str().to_string() }}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct RevokeSessionInput { pub session_id: String,}
pub async fn revoke_session( State(state): State<AppState>, auth: Auth<Active>, Json(input): Json<RevokeSessionInput>,) -> Result<Json<EmptyResponse>, ApiError> { if let Some(jwt_id) = input.session_id.strip_prefix("jwt:") { let session_id = jwt_id .parse::<i32>() .map(SessionId::new) .map_err(|_| ApiError::InvalidRequest("Invalid session ID".into()))?; let access_jti = state .repos .session .get_session_access_jti_by_id(session_id, &auth.did) .await .log_db_err("in revoke_session")? .ok_or(ApiError::SessionNotFound)?; state .repos .session .delete_session_by_id(session_id, &auth.did) .await .log_db_err("deleting session")?; let cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &access_jti); if let Err(e) = state.cache.delete(&cache_key).await { warn!("Failed to invalidate session cache: {:?}", e); } info!(did = %&auth.did, session_id = %session_id, "JWT session revoked"); } else if let Some(oauth_id) = input.session_id.strip_prefix("oauth:") { let session_id = oauth_id .parse::<i32>() .map(TokenFamilyId::new) .map_err(|_| ApiError::InvalidRequest("Invalid session ID".into()))?; let deleted = state .repos .oauth .delete_session_by_id(session_id, &auth.did) .await .log_db_err("deleting OAuth session")?; if deleted == 0 { return Err(ApiError::SessionNotFound); } info!(did = %&auth.did, session_id = %session_id, "OAuth session revoked"); } else { return Err(ApiError::InvalidRequest("Invalid session ID format".into())); } Ok(Json(EmptyResponse {}))}
pub async fn revoke_all_sessions( State(state): State<AppState>, headers: HeaderMap, auth: Auth<Active>,) -> Result<Json<SuccessResponse>, ApiError> { let jti = tranquil_pds::auth::extract_jti_from_headers(&headers) .ok_or(ApiError::InvalidToken(None))?;
if auth.is_oauth() { state .repos .session .delete_sessions_by_did(&auth.did) .await .log_db_err("revoking JWT sessions")?; let token_id = TokenId::from(jti.clone().into_inner()); state .repos .oauth .delete_sessions_by_did_except(&auth.did, &token_id) .await .log_db_err("revoking OAuth sessions")?; } else { state .repos .session .delete_sessions_by_did_except_jti(&auth.did, &jti) .await .log_db_err("revoking JWT sessions")?; state .repos .oauth .delete_sessions_by_did(&auth.did) .await .log_db_err("revoking OAuth sessions")?; }
info!(did = %&auth.did, "All other sessions revoked"); Ok(Json(SuccessResponse { success: true }))}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct LegacyLoginPreferenceOutput { pub allow_legacy_login: bool, pub has_mfa: bool,}
pub async fn get_legacy_login_preference( State(state): State<AppState>, auth: Auth<Active>,) -> Result<Json<LegacyLoginPreferenceOutput>, ApiError> { let pref = state .repos .user .get_legacy_login_pref(&auth.did) .await .log_db_err("getting legacy login pref")? .ok_or(ApiError::AccountNotFound)?; Ok(Json(LegacyLoginPreferenceOutput { allow_legacy_login: pref.allow_legacy_login, has_mfa: pref.has_mfa, }))}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct UpdateLegacyLoginInput { pub allow_legacy_login: bool,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct UpdateLegacyLoginOutput { pub allow_legacy_login: bool,}
pub async fn update_legacy_login_preference( State(state): State<AppState>, auth: Auth<Active>, Json(input): Json<UpdateLegacyLoginInput>,) -> Result<Json<UpdateLegacyLoginOutput>, ApiError> { let session_mfa = require_legacy_session_mfa(&state, &auth).await?;
let reauth_mfa = require_reauth_window(&state, &auth).await?;
let updated = state .repos .user .update_legacy_login(reauth_mfa.did(), input.allow_legacy_login) .await .log_db_err("updating legacy login")?; if !updated { return Err(ApiError::AccountNotFound); } info!( did = %session_mfa.did(), allow_legacy_login = input.allow_legacy_login, "Legacy login preference updated" ); Ok(Json(UpdateLegacyLoginOutput { allow_legacy_login: input.allow_legacy_login, }))}
use tranquil_pds::comms::VALID_LOCALES;
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct UpdateLocaleInput { pub preferred_locale: String,}
pub async fn update_locale( State(state): State<AppState>, auth: Auth<Active>, Json(input): Json<UpdateLocaleInput>,) -> Result<Json<PreferredLocaleOutput>, ApiError> { if !VALID_LOCALES.contains(&input.preferred_locale.as_str()) { return Err(ApiError::InvalidRequest(format!( "Invalid locale. Valid options: {}", VALID_LOCALES.join(", ") ))); }
let updated = state .repos .user .update_locale(&auth.did, &input.preferred_locale) .await .log_db_err("updating locale")?; if !updated { return Err(ApiError::AccountNotFound); } info!( did = %&auth.did, locale = %input.preferred_locale, "User locale preference updated" ); Ok(Json(PreferredLocaleOutput { preferred_locale: Some(input.preferred_locale), }))}