mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-08-19 07:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

bsky(auth): add grace period to legacy session refresh

Browse Source 
Concurrent or retried com.atproto.server.refreshSession calls presenting the
same refresh token hit the reuse-detection path, which deleted the session and
returned "Refresh token has been revoked due to suspected compromise" —
logging users out at random. The legacy flow had no grace period, unlike OAuth.

Mirror the reference atproto PDS: every rotated refresh token gets a 2h grace
window measured from its own rotation time (used_refresh_tokens.used_at in
postgres; a rotated_at_ms field appended to the metastore used-marker, with
old-format markers decoding as outside the window). A refresh presenting a
recently-rotated token is served the session's current tokens, re-minted on
the fly with the same jti/expiry — signed JWTs are never persisted. Reuse
outside the window still revokes the session.

The grace lookup returns the session's encrypted signing key so the handler
verifies the presented token's signature before minting replacement tokens or
revoking a session; a forged token bearing a known jti gets a generic
rejection with no side effects.

Integration tests asserting the old replay-gets-401 behavior are reworked to
the new contract and now also cover forged-signature replays and
out-of-window revocation.
This commit is contained in:
ave
2026-06-27 23:54:22 +03:00
committed by Tangled
parent ab4eba6dc4
commit 9dc184ee33
19 changed files with 897 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.sqlx/query-4dcee809896ead3de8ca0433856ed424211d79df201d08bbea0e4c576931a234.json
Generated
-18
View File
@@ -1,18 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE session_tokens SET access_jti = $1, refresh_jti = $2, access_expires_at = $3, refresh_expires_at = $4, updated_at = NOW() WHERE id = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Timestamptz",
"Timestamptz",
"Int4"
]
},
"nullable": []
},
"hash": "4dcee809896ead3de8ca0433856ed424211d79df201d08bbea0e4c576931a234"
}
.sqlx/query-b2e1736dbe2ab9114e373353bcc299176417f3c9220025f9521591ba62928bd7.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1 FOR UPDATE",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "session_id",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "b2e1736dbe2ab9114e373353bcc299176417f3c9220025f9521591ba62928bd7"
}
.sqlx/query-c08e3760be75071ac5e281cb3b05b3a02dccbf35b96c3d50c652878a38c9f9e6.json
Generated
+82
View File
@@ -0,0 +1,82 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT u.used_at, st.id AS session_id, st.did, st.scope, st.controller_did,\n st.access_jti, st.refresh_jti, st.access_expires_at, st.refresh_expires_at,\n k.key_bytes, k.encryption_version\n FROM used_refresh_tokens u\n JOIN session_tokens st ON st.id = u.session_id\n JOIN users us ON st.did = us.did\n JOIN user_keys k ON us.id = k.user_id\n WHERE u.refresh_jti = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "used_at",
"type_info": "Timestamptz"
},
{
"ordinal": 1,
"name": "session_id",
"type_info": "Int4"
},
{
"ordinal": 2,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "scope",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "controller_did",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "access_jti",
"type_info": "Text"
},
{
"ordinal": 6,
"name": "refresh_jti",
"type_info": "Text"
},
{
"ordinal": 7,
"name": "access_expires_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "refresh_expires_at",
"type_info": "Timestamptz"
},
{
"ordinal": 9,
"name": "key_bytes",
"type_info": "Bytea"
},
{
"ordinal": 10,
"name": "encryption_version",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false,
true,
true,
false,
false,
false,
false,
false,
true
]
},
"hash": "c08e3760be75071ac5e281cb3b05b3a02dccbf35b96c3d50c652878a38c9f9e6"
}
.sqlx/query-c0a7b73070e6c5b9baa9caddca3c5d24427b48ed02b5cac913a0c0af1e5c9379.json
Generated
-15
View File
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO used_refresh_tokens (refresh_jti, session_id)\n VALUES ($1, $2)\n ON CONFLICT (refresh_jti) DO NOTHING\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Int4"
]
},
"nullable": []
},
"hash": "c0a7b73070e6c5b9baa9caddca3c5d24427b48ed02b5cac913a0c0af1e5c9379"
}
.sqlx/query-fcd868a192d27fd4eccae92a884e881b8d6f09bf7ae08a9b431a44acbf2f91f3.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "session_id",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "fcd868a192d27fd4eccae92a884e881b8d6f09bf7ae08a9b431a44acbf2f91f3"
}
crates/tranquil-api/src/server/session.rs
+131 -27
View File
@@ -505,16 +505,72 @@ pub async fn refresh_session(
)));
}
};
if let Ok(Some(_)) = state
.repos
.session
.check_refresh_token_used(&refresh_jti)
.await
{
warn!("Refresh token reuse detected for jti: {}", refresh_jti);
return Err(ApiError::AuthenticationFailed(Some(
"Refresh token has been revoked due to suspected compromise".into(),
)));
match state.repos.session.lookup_refresh_grace(&refresh_jti).await {
Ok(tranquil_db_traits::RefreshGraceLookup::NotUsed) => {}
Ok(tranquil_db_traits::RefreshGraceLookup::Replay(replay)) => {
// Verify the presented token's signature before issuing anything: a
// rotated jti is a public value, so we must never mint for a forged
// unsigned token carrying it.
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: {refresh_jti}; replaying tokens"
);
let (access_jwt, refresh_jwt) = remint_grace_tokens(&replay, &key)?;
return build_refresh_session_output(&state, replay.did, access_jwt, refresh_jwt).await;
}
Ok(tranquil_db_traits::RefreshGraceLookup::Compromised {
session_id,
key_bytes,
encryption_version,
}) => {
// Never revoke a session for an unverified token: verify the
// signature first, so a forged unsigned token cannot force a logout.
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 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(),
)));
}
warn!(
"Refresh token reuse outside grace window for jti: {refresh_jti}; revoking session"
);
if let Err(e) = state.repos.session.delete_session_by_id(session_id).await {
error!(
"Failed to revoke session {} for refresh token reuse: {:?}",
session_id.as_i32(),
e
);
return Err(ApiError::InternalError(None));
}
return Err(ApiError::AuthenticationFailed(Some(
"Refresh token has been revoked due to suspected compromise".into(),
)));
}
Err(e) => {
error!("Database error checking refresh token grace: {:?}", e);
return Err(ApiError::InternalError(None));
}
}
let session_row = match state
.repos
@@ -579,22 +635,27 @@ pub async fn refresh_session(
new_access_expires_at: new_access_meta.expires_at,
new_refresh_expires_at: new_refresh_meta.expires_at,
};
match state
let (access_jwt, refresh_jwt) = match state
.repos
.session
.refresh_session_atomic(&refresh_data)
.await
{
Ok(tranquil_db_traits::RefreshSessionResult::Success) => {}
Ok(tranquil_db_traits::RefreshSessionResult::TokenAlreadyUsed) => {
warn!("Refresh token reuse detected during atomic operation");
return Err(ApiError::AuthenticationFailed(Some(
"Refresh token has been revoked due to suspected compromise".into(),
)));
Ok(tranquil_db_traits::RefreshSessionResult::Success) => {
(new_access_meta.token, new_refresh_meta.token)
}
Ok(tranquil_db_traits::RefreshSessionResult::ConcurrentRefresh) => {
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!(
"Concurrent refresh detected for session_id: {}",
"Refresh token reuse outside grace window or unreplayable rotation conflict for session_id: {}",
session_row.id
);
return Err(ApiError::AuthenticationFailed(Some(
@@ -605,12 +666,55 @@ pub async fn refresh_session(
error!("Database error during session refresh: {:?}", e);
return Err(ApiError::InternalError(None));
}
}
let did_for_doc = session_row.did.clone();
};
build_refresh_session_output(&state, session_row.did, access_jwt, refresh_jwt).await
}
/// 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_deref(),
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(&session_row.did),
did_resolver.fetch_did_document(&did_for_doc)
state.repos.user.get_session_info_by_did(&did),
did_resolver.fetch_did_document(&did)
);
match db_result {
Ok(Some(u)) => {
@@ -621,10 +725,10 @@ pub async fn refresh_session(
let account_state =
AccountState::from_db_fields(u.deactivated_at, u.takedown_ref.clone(), None, None);
Ok(Json(RefreshSessionOutput {
access_jwt: new_access_meta.token,
refresh_jwt: new_refresh_meta.token,
access_jwt,
refresh_jwt,
handle,
did: session_row.did,
did,
email: u.email,
email_confirmed: u.channel_verification.email,
preferred_channel: u.preferred_comms_channel.as_str().to_string(),
@@ -637,7 +741,7 @@ pub async fn refresh_session(
}))
}
Ok(None) => {
error!("User not found for existing session: {}", session_row.did);
error!("User not found for existing session: {}", did);
Err(ApiError::InternalError(None))
}
Err(e) => {
crates/tranquil-auth/src/lib.rs
+5 -4
View File
@@ -5,10 +5,11 @@ mod verify;
pub use token::{
create_access_token, create_access_token_hs256, create_access_token_hs256_with_metadata,
create_access_token_with_delegation, create_access_token_with_metadata,
create_access_token_with_scope_metadata, create_refresh_token, create_refresh_token_hs256,
create_refresh_token_hs256_with_metadata, create_refresh_token_with_metadata,
create_service_token, create_service_token_hs256,
create_access_token_with_delegation, create_access_token_with_jti,
create_access_token_with_metadata, create_access_token_with_scope_metadata,
create_refresh_token, create_refresh_token_hs256, create_refresh_token_hs256_with_metadata,
create_refresh_token_with_jti, create_refresh_token_with_metadata, create_service_token,
create_service_token_hs256,
};
pub use totp::{
crates/tranquil-auth/src/token.rs
+65 -4
View File
@@ -4,7 +4,7 @@ use super::types::{
use anyhow::Result;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use chrono::{Duration, Utc};
use chrono::{DateTime, Duration, Utc};
use hmac::{Hmac, Mac};
use k256::ecdsa::{Signature, SigningKey, signature::Signer};
use sha2::Sha256;
@@ -74,6 +74,54 @@ pub fn create_refresh_token_with_metadata(
)
}
/// Re-mint an access token carrying a specific `jti` and expiry. Used by the
/// refresh grace window to reproduce a session's current access token without
/// persisting the signed JWT itself.
pub fn create_access_token_with_jti(
did: &str,
key_bytes: &[u8],
scopes: Option<&str>,
controller_did: Option<&str>,
hostname: Option<&str>,
jti: &str,
expires_at: DateTime<Utc>,
) -> Result<String> {
let scope = scopes.unwrap_or(TokenScope::Access.as_str());
let act = controller_did.map(|c| ActClaim { sub: c.to_string() });
Ok(create_signed_token_pinned(
did,
scope,
TokenType::Access,
key_bytes,
expires_at,
jti.to_string(),
act,
hostname,
)?
.token)
}
/// Re-mint a refresh token carrying a specific `jti` and expiry. Counterpart to
/// [`create_access_token_with_jti`] for the refresh grace window.
pub fn create_refresh_token_with_jti(
did: &str,
key_bytes: &[u8],
jti: &str,
expires_at: DateTime<Utc>,
) -> Result<String> {
Ok(create_signed_token_pinned(
did,
TokenScope::Refresh.as_str(),
TokenType::Refresh,
key_bytes,
expires_at,
jti.to_string(),
None,
None,
)?
.token)
}
pub fn create_service_token(
did: &str,
aud: &str,
@@ -122,14 +170,27 @@ fn create_signed_token_with_act(
act: Option<ActClaim>,
hostname: Option<&str>,
) -> Result<TokenWithMetadata> {
let signing_key = SigningKey::from_slice(key_bytes)?;
let expires_at = Utc::now()
.checked_add_signed(duration)
.expect("valid timestamp");
let jti = uuid::Uuid::new_v4().to_string();
create_signed_token_pinned(did, scope, typ, key_bytes, expires_at, jti, act, hostname)
}
#[allow(clippy::too_many_arguments)]
fn create_signed_token_pinned(
did: &str,
scope: &str,
typ: TokenType,
key_bytes: &[u8],
expires_at: DateTime<Utc>,
jti: String,
act: Option<ActClaim>,
hostname: Option<&str>,
) -> Result<TokenWithMetadata> {
let signing_key = SigningKey::from_slice(key_bytes)?;
let expiration = expires_at.timestamp();
let jti = uuid::Uuid::new_v4().to_string();
let aud_hostname = hostname.map(|h| h.to_string()).unwrap_or_else(|| {
tranquil_config::try_get()
crates/tranquil-db-traits/src/lib.rs
+2 -1
View File
@@ -43,7 +43,8 @@ pub use repo::{
pub use scope::{DbScope, InvalidScopeError};
pub use sequence::{SequenceNumber, deserialize_optional_sequence};
pub use session::{
AppPasswordCreate, AppPasswordPrivilege, AppPasswordRecord, LoginType, RefreshSessionResult,
AppPasswordCreate, AppPasswordPrivilege, AppPasswordRecord, LoginType,
REFRESH_GRACE_PERIOD_SECS, RefreshGraceLookup, RefreshGraceReplay, RefreshSessionResult,
SessionForRefresh, SessionId, SessionListItem, SessionMfaStatus, SessionRefreshData,
SessionRepository, SessionToken, SessionTokenCreate,
};
crates/tranquil-db-traits/src/session.rs
+39 -12
View File
@@ -158,11 +158,47 @@ pub struct SessionMfaStatus {
pub last_reauth_at: Option<DateTime<Utc>>,
}
/// Window during which a just-rotated refresh token may still be presented by a
/// benignly-racing or retrying client without being treated as a compromise.
/// The window is measured per rotated token from its own rotation time
/// (`used_refresh_tokens.used_at` / the metastore marker's `rotated_at_ms`),
/// matching the reference atproto PDS `REFRESH_GRACE_MS`
/// (`account-manager.ts::rotateRefreshToken`), which shortens a rotated refresh
/// token's lifetime to a 2-hour revocation grace period.
pub const REFRESH_GRACE_PERIOD_SECS: i64 = 2 * 60 * 60;
/// The session's current token identity (plus its signing key) needed to re-mint
/// the access/refresh JWTs during a grace-window replay. We carry the key so the
/// caller can verify the presented token's signature before issuing anything.
#[derive(Debug, Clone)]
pub struct RefreshGraceReplay {
pub did: Did,
pub scope: Option<String>,
pub controller_did: Option<Did>,
pub access_jti: String,
pub refresh_jti: String,
pub access_expires_at: DateTime<Utc>,
pub refresh_expires_at: DateTime<Utc>,
pub key_bytes: Vec<u8>,
pub encryption_version: i32,
}
#[derive(Debug, Clone)]
pub enum RefreshSessionResult {
Success,
TokenAlreadyUsed,
ConcurrentRefresh,
GraceReplay(RefreshGraceReplay),
Compromise,
}
#[derive(Debug, Clone)]
pub enum RefreshGraceLookup {
NotUsed,
Replay(RefreshGraceReplay),
Compromised {
session_id: SessionId,
key_bytes: Vec<u8>,
encryption_version: i32,
},
}
#[derive(Debug, Clone)]
@@ -230,16 +266,7 @@ pub trait SessionRepository: Send + Sync {
app_password_name: &str,
) -> Result<Vec<String>, DbError>;
async fn check_refresh_token_used(
&self,
refresh_jti: &str,
) -> Result<Option<SessionId>, DbError>;
async fn mark_refresh_token_used(
&self,
refresh_jti: &str,
session_id: SessionId,
) -> Result<bool, DbError>;
async fn lookup_refresh_grace(&self, refresh_jti: &str) -> Result<RefreshGraceLookup, DbError>;
async fn list_app_passwords(&self, user_id: Uuid) -> Result<Vec<AppPasswordRecord>, DbError>;
crates/tranquil-db/src/postgres/session.rs
+79 -55
View File
@@ -1,10 +1,11 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use chrono::{DateTime, Duration, Utc};
use sqlx::PgPool;
use tranquil_db_traits::{
AppPasswordCreate, AppPasswordPrivilege, AppPasswordRecord, DbError, LoginType,
RefreshSessionResult, SessionForRefresh, SessionId, SessionListItem, SessionMfaStatus,
SessionRefreshData, SessionRepository, SessionToken, SessionTokenCreate,
REFRESH_GRACE_PERIOD_SECS, RefreshGraceLookup, RefreshGraceReplay, RefreshSessionResult,
SessionForRefresh, SessionId, SessionListItem, SessionMfaStatus, SessionRefreshData,
SessionRepository, SessionToken, SessionTokenCreate,
};
use tranquil_types::Did;
use uuid::Uuid;
@@ -267,40 +268,51 @@ impl SessionRepository for PostgresSessionRepository {
Ok(rows)
}
async fn check_refresh_token_used(
&self,
refresh_jti: &str,
) -> Result<Option<SessionId>, DbError> {
let row = sqlx::query_scalar!(
"SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1",
async fn lookup_refresh_grace(&self, refresh_jti: &str) -> Result<RefreshGraceLookup, DbError> {
let row = sqlx::query!(
r#"
SELECT u.used_at, st.id AS session_id, st.did, st.scope, st.controller_did,
st.access_jti, st.refresh_jti, st.access_expires_at, st.refresh_expires_at,
k.key_bytes, k.encryption_version
FROM used_refresh_tokens u
JOIN session_tokens st ON st.id = u.session_id
JOIN users us ON st.did = us.did
JOIN user_keys k ON us.id = k.user_id
WHERE u.refresh_jti = $1
"#,
refresh_jti
)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(row.map(SessionId::new))
}
// No marker (or a missing users/user_keys join row) degrades to NotUsed.
// That is safe: the normal refresh path then fails closed with "Invalid
// refresh token" without mutating any state.
let Some(r) = row else {
return Ok(RefreshGraceLookup::NotUsed);
};
async fn mark_refresh_token_used(
&self,
refresh_jti: &str,
session_id: SessionId,
) -> Result<bool, DbError> {
let result = sqlx::query!(
r#"
INSERT INTO used_refresh_tokens (refresh_jti, session_id)
VALUES ($1, $2)
ON CONFLICT (refresh_jti) DO NOTHING
"#,
refresh_jti,
session_id.as_i32()
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0)
let grace_cutoff = Utc::now() - Duration::seconds(REFRESH_GRACE_PERIOD_SECS);
if r.used_at > grace_cutoff {
Ok(RefreshGraceLookup::Replay(RefreshGraceReplay {
did: Did::from(r.did),
scope: r.scope,
controller_did: r.controller_did.map(Did::from),
access_jti: r.access_jti,
refresh_jti: r.refresh_jti,
access_expires_at: r.access_expires_at,
refresh_expires_at: r.refresh_expires_at,
key_bytes: r.key_bytes,
encryption_version: r.encryption_version.unwrap_or(0),
}))
} else {
Ok(RefreshGraceLookup::Compromised {
session_id: SessionId::new(r.session_id),
key_bytes: r.key_bytes,
encryption_version: r.encryption_version.unwrap_or(0),
})
}
}
async fn list_app_passwords(&self, user_id: Uuid) -> Result<Vec<AppPasswordRecord>, DbError> {
@@ -524,21 +536,10 @@ impl SessionRepository for PostgresSessionRepository {
) -> Result<RefreshSessionResult, DbError> {
let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
if let Ok(Some(session_id)) = sqlx::query_scalar!(
"SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1 FOR UPDATE",
data.old_refresh_jti
)
.fetch_optional(&mut *tx)
.await
{
let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_id)
.execute(&mut *tx)
.await;
tx.commit().await.map_err(map_sqlx_error)?;
return Ok(RefreshSessionResult::TokenAlreadyUsed);
}
let result = sqlx::query!(
// Atomically claim the old refresh jti. The INSERT serializes concurrent
// rotations of the same token: exactly one request inserts the row, the
// rest see `rows_affected == 0`.
let claimed = sqlx::query!(
"INSERT INTO used_refresh_tokens (refresh_jti, session_id) VALUES ($1, $2) ON CONFLICT (refresh_jti) DO NOTHING",
data.old_refresh_jti,
data.session_id.as_i32()
@@ -547,19 +548,42 @@ impl SessionRepository for PostgresSessionRepository {
.await
.map_err(map_sqlx_error)?;
if result.rows_affected() == 0 {
let _ = sqlx::query!(
"DELETE FROM session_tokens WHERE id = $1",
data.session_id.as_i32()
)
.execute(&mut *tx)
.await;
tx.commit().await.map_err(map_sqlx_error)?;
return Ok(RefreshSessionResult::ConcurrentRefresh);
if claimed.rows_affected() == 0 {
// Another request already rotated this token. Nothing to write, so
// end our transaction before reading the winner's committed row.
tx.rollback().await.map_err(map_sqlx_error)?;
// Within the grace window (measured from this token's own rotation
// time) we replay the session's current tokens so a benignly-racing
// client keeps a working session instead of being revoked.
match self.lookup_refresh_grace(&data.old_refresh_jti).await? {
RefreshGraceLookup::Replay(replay) => {
return Ok(RefreshSessionResult::GraceReplay(replay));
}
RefreshGraceLookup::Compromised { .. } | RefreshGraceLookup::NotUsed => {
// Outside the grace window, or the marker/session vanished
// concurrently: genuine reuse. Revoke the session (delete is
// idempotent).
sqlx::query!(
"DELETE FROM session_tokens WHERE id = $1",
data.session_id.as_i32()
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
return Ok(RefreshSessionResult::Compromise);
}
}
}
// We won the rotation.
sqlx::query!(
"UPDATE session_tokens SET access_jti = $1, refresh_jti = $2, access_expires_at = $3, refresh_expires_at = $4, updated_at = NOW() WHERE id = $5",
r#"
UPDATE session_tokens
SET access_jti = $1, refresh_jti = $2, access_expires_at = $3,
refresh_expires_at = $4, updated_at = NOW()
WHERE id = $5
"#,
data.new_access_jti,
data.new_refresh_jti,
data.new_access_expires_at,
crates/tranquil-pds/src/auth/mod.rs
+9 -8
View File
@@ -46,14 +46,15 @@ pub use tranquil_auth::{
ActClaim, Claims, Header, SigningAlgorithm, TokenData, TokenDecodeError, TokenScope, TokenType,
TokenVerifyError, TokenWithMetadata, TotpError, UnsafeClaims, create_access_token,
create_access_token_hs256, create_access_token_hs256_with_metadata,
create_access_token_with_delegation, create_access_token_with_metadata,
create_access_token_with_scope_metadata, create_refresh_token, create_refresh_token_hs256,
create_refresh_token_hs256_with_metadata, create_refresh_token_with_metadata,
create_service_token, create_service_token_hs256, generate_backup_codes,
generate_qr_png_base64, generate_totp_secret, generate_totp_uri, get_algorithm_from_token,
get_did_from_token, get_jti_from_token, hash_backup_code, is_backup_code_format,
verify_access_token, verify_access_token_hs256, verify_access_token_typed, verify_backup_code,
verify_refresh_token, verify_refresh_token_hs256, verify_token, verify_totp_code,
create_access_token_with_delegation, create_access_token_with_jti,
create_access_token_with_metadata, create_access_token_with_scope_metadata,
create_refresh_token, create_refresh_token_hs256, create_refresh_token_hs256_with_metadata,
create_refresh_token_with_jti, create_refresh_token_with_metadata, create_service_token,
create_service_token_hs256, generate_backup_codes, generate_qr_png_base64,
generate_totp_secret, generate_totp_uri, get_algorithm_from_token, get_did_from_token,
get_jti_from_token, hash_backup_code, is_backup_code_format, verify_access_token,
verify_access_token_hs256, verify_access_token_typed, verify_backup_code, verify_refresh_token,
verify_refresh_token_hs256, verify_token, verify_totp_code,
};
pub fn lxm_permits(lxm: &str, expected: &str) -> bool {
crates/tranquil-pds/tests/jwt_security.rs
+69 -2
View File
@@ -673,8 +673,12 @@ async fn test_deactivated_account_behavior() {
assert_eq!(post_body["error"], "AccountDeactivated");
}
// End-to-end regression for the legacy refresh grace window and the
// verify-before-mutate fixes: a signed replay within the 2h grace window converges
// on the winner's current tokens, while a forged replay is rejected without
// destroying the session.
#[tokio::test]
async fn test_refresh_token_replay_protection() {
async fn test_refresh_token_replay_grace_and_forgery() {
let url = base_url().await;
let http_client = client();
let suffix = &uuid::Uuid::new_v4().simple().to_string()[..8];
@@ -728,6 +732,7 @@ async fn test_refresh_token_replay_protection() {
let confirmed: Value = confirm.json().await.unwrap();
let refresh_jwt = confirmed["refreshJwt"].as_str().unwrap().to_string();
// 1. Rotate refresh_jwt once; capture the winner's tokens.
let first = http_client
.post(format!("{}/xrpc/com.atproto.server.refreshSession", url))
.header("Authorization", format!("Bearer {}", refresh_jwt))
@@ -735,12 +740,74 @@ async fn test_refresh_token_replay_protection() {
.await
.unwrap();
assert_eq!(first.status(), StatusCode::OK);
let first_body: Value = first.json().await.unwrap();
let winner_refresh = first_body["refreshJwt"].as_str().unwrap().to_string();
let winner_access = first_body["accessJwt"].as_str().unwrap().to_string();
// 2. Signed replay of the original (now-rotated) token within the grace
// window: 200, returning the session's current tokens. The re-minted JWTs are
// not byte-identical to the winner's because the pinned claims still carry a
// fresh `iat` (see create_signed_token_pinned), so assert they VERIFY and WORK
// instead: the returned refreshJwt drives a subsequent refresh.
let replay = http_client
.post(format!("{}/xrpc/com.atproto.server.refreshSession", url))
.header("Authorization", format!("Bearer {}", refresh_jwt))
.send()
.await
.unwrap();
assert_eq!(replay.status(), StatusCode::UNAUTHORIZED);
assert_eq!(replay.status(), StatusCode::OK);
let replay_body: Value = replay.json().await.unwrap();
let replayed_refresh = replay_body["refreshJwt"].as_str().unwrap().to_string();
let replayed_access = replay_body["accessJwt"].as_str().unwrap().to_string();
// The replayed access token authenticates.
let who = http_client
.get(format!("{}/xrpc/com.atproto.server.getSession", url))
.header("Authorization", format!("Bearer {}", replayed_access))
.send()
.await
.unwrap();
assert_eq!(who.status(), StatusCode::OK);
// The replayed refresh token rotates the session forward (200), proving the
// grace replay handed back the live session's current credentials.
let from_replay = http_client
.post(format!("{}/xrpc/com.atproto.server.refreshSession", url))
.header("Authorization", format!("Bearer {}", replayed_refresh))
.send()
.await
.unwrap();
assert_eq!(from_replay.status(), StatusCode::OK);
let from_replay_body: Value = from_replay.json().await.unwrap();
let current_refresh = from_replay_body["refreshJwt"].as_str().unwrap().to_string();
assert!(!winner_refresh.is_empty() && !winner_access.is_empty());
// 3. Forged replay: tamper the original token's signature segment with
// same-length valid base64url garbage. Must be rejected (401) WITHOUT
// destroying the session.
let parts: Vec<&str> = refresh_jwt.split('.').collect();
assert_eq!(parts.len(), 3);
let garbage_sig: String = parts[2]
.chars()
.map(|c| if c == 'A' { 'B' } else { 'A' })
.collect();
assert_eq!(garbage_sig.len(), parts[2].len());
let forged = format!("{}.{}.{}", parts[0], parts[1], garbage_sig);
let forged_res = http_client
.post(format!("{}/xrpc/com.atproto.server.refreshSession", url))
.header("Authorization", format!("Bearer {}", forged))
.send()
.await
.unwrap();
assert_eq!(forged_res.status(), StatusCode::UNAUTHORIZED);
// The session survived the forgery: the current refresh token still works.
let after_forge = http_client
.post(format!("{}/xrpc/com.atproto.server.refreshSession", url))
.header("Authorization", format!("Bearer {}", current_refresh))
.send()
.await
.unwrap();
assert_eq!(after_forge.status(), StatusCode::OK);
}
crates/tranquil-pds/tests/lifecycle_session.rs
+20 -4
View File
@@ -158,6 +158,9 @@ async fn test_session_lifecycle_refresh_invalidates_old() {
let refresh_body: Value = refresh_res.json().await.unwrap();
let new_refresh_jwt = refresh_body["refreshJwt"].as_str().unwrap();
assert_ne!(refresh_jwt, new_refresh_jwt, "Refresh tokens should differ");
// A signed reuse within the grace window is benign: it replays the session's
// current tokens rather than revoking. The deep assertions live in
// jwt_security.rs::test_refresh_token_replay_grace_and_forgery.
let reuse_res = client
.post(format!(
"{}/xrpc/com.atproto.server.refreshSession",
@@ -167,11 +170,24 @@ async fn test_session_lifecycle_refresh_invalidates_old() {
.send()
.await
.expect("Failed reuse attempt");
assert!(
reuse_res.status() == StatusCode::UNAUTHORIZED
|| reuse_res.status() == StatusCode::BAD_REQUEST,
"Old refresh token should be invalid after use"
assert_eq!(
reuse_res.status(),
StatusCode::OK,
"Signed reuse within grace replays the session"
);
let reuse_body: Value = reuse_res.json().await.unwrap();
let replayed_refresh = reuse_body["refreshJwt"].as_str().unwrap();
// The replayed refresh token works for a subsequent refresh.
let followup = client
.post(format!(
"{}/xrpc/com.atproto.server.refreshSession",
base_url().await
))
.bearer_auth(replayed_refresh)
.send()
.await
.expect("Failed followup refresh");
assert_eq!(followup.status(), StatusCode::OK);
}
#[tokio::test]
crates/tranquil-store/src/metastore/client.rs
+3 -19
View File
@@ -1596,13 +1596,13 @@ impl<S: StorageIO + 'static> tranquil_db_traits::SessionRepository for Metastore
recv(rx).await
}
async fn check_refresh_token_used(
async fn lookup_refresh_grace(
&self,
refresh_jti: &str,
) -> Result<Option<tranquil_db_traits::SessionId>, DbError> {
) -> Result<tranquil_db_traits::RefreshGraceLookup, DbError> {
let (tx, rx) = oneshot::channel();
self.pool.send(MetastoreRequest::Session(
SessionRequest::CheckRefreshTokenUsed {
SessionRequest::LookupRefreshGrace {
refresh_jti: refresh_jti.to_owned(),
tx,
},
@@ -1610,22 +1610,6 @@ impl<S: StorageIO + 'static> tranquil_db_traits::SessionRepository for Metastore
recv(rx).await
}
async fn mark_refresh_token_used(
&self,
refresh_jti: &str,
session_id: tranquil_db_traits::SessionId,
) -> Result<bool, DbError> {
let (tx, rx) = oneshot::channel();
self.pool.send(MetastoreRequest::Session(
SessionRequest::MarkRefreshTokenUsed {
refresh_jti: refresh_jti.to_owned(),
session_id,
tx,
},
))?;
recv(rx).await
}
async fn list_app_passwords(
&self,
user_id: Uuid,
crates/tranquil-store/src/metastore/handler.rs
+5 -23
View File
@@ -892,14 +892,9 @@ pub enum SessionRequest {
app_password_name: String,
tx: Tx<Vec<String>>,
},
CheckRefreshTokenUsed {
LookupRefreshGrace {
refresh_jti: String,
tx: Tx<Option<SessionId>>,
},
MarkRefreshTokenUsed {
refresh_jti: String,
session_id: SessionId,
tx: Tx<bool>,
tx: Tx<tranquil_db_traits::RefreshGraceLookup>,
},
ListAppPasswords {
user_id: Uuid,
@@ -960,8 +955,7 @@ impl SessionRequest {
Self::CreateSession { .. }
| Self::GetSessionByAccessJti { .. }
| Self::GetSessionForRefresh { .. }
| Self::CheckRefreshTokenUsed { .. }
| Self::MarkRefreshTokenUsed { .. }
| Self::LookupRefreshGrace { .. }
| Self::DeleteSessionByAccessJti { .. }
| Self::DeleteSessionById { .. } => Routing::Global,
Self::DeleteSessionsByDid { did, .. }
@@ -3744,23 +3738,11 @@ fn dispatch_session<S: StorageIO>(state: &HandlerState<S>, req: SessionRequest)
.map_err(metastore_to_db);
let _ = tx.send(result);
}
SessionRequest::CheckRefreshTokenUsed { refresh_jti, tx } => {
SessionRequest::LookupRefreshGrace { refresh_jti, tx } => {
let result = state
.metastore
.session_ops()
.check_refresh_token_used(&refresh_jti)
.map_err(metastore_to_db);
let _ = tx.send(result);
}
SessionRequest::MarkRefreshTokenUsed {
refresh_jti,
session_id,
tx,
} => {
let result = state
.metastore
.session_ops()
.mark_refresh_token_used(&refresh_jti, session_id)
.lookup_refresh_grace(&refresh_jti)
.map_err(metastore_to_db);
let _ = tx.send(result);
}
crates/tranquil-store/src/metastore/mod.rs
+239
View File
@@ -429,6 +429,245 @@ mod tests {
});
}
fn legacy_session_create(
did: &str,
access_jti: &str,
refresh_jti: &str,
) -> tranquil_db_traits::SessionTokenCreate {
let now = chrono::Utc::now();
tranquil_db_traits::SessionTokenCreate {
did: tranquil_types::Did::new(did.to_string()).unwrap(),
access_jti: access_jti.to_string(),
refresh_jti: refresh_jti.to_string(),
access_expires_at: now + chrono::Duration::minutes(120),
refresh_expires_at: now + chrono::Duration::days(90),
login_type: tranquil_db_traits::LoginType::Legacy,
mfa_verified: false,
scope: None,
controller_did: None,
app_password_name: None,
}
}
fn create_test_user(ms: &Metastore, did: &str, handle: &str) {
let input = tranquil_db_traits::CreatePasswordAccountInput {
handle: tranquil_types::Handle::new(handle.to_string()).unwrap(),
email: None,
did: tranquil_types::Did::new(did.to_string()).unwrap(),
password_hash: "test-hash".to_string(),
preferred_comms_channel: tranquil_db_traits::CommsChannel::Email,
discord_username: None,
telegram_username: None,
signal_username: None,
deactivated_at: None,
inbound_migration: false,
encrypted_key_bytes: vec![7u8; 32],
encryption_version: 0,
reserved_key_id: None,
commit_cid: "bafyreib2rxk3ryblouj3fxza5jvx6psmwewwessc4m6g6e7pqhhkwqomfi".to_string(),
repo_rev: "rev0".to_string(),
genesis_block_cids: vec![],
invite_code: None,
birthdate_pref: None,
};
ms.user_ops().create_password_account(&input).unwrap();
}
fn legacy_refresh_data(
session_id: tranquil_db_traits::SessionId,
old_refresh_jti: &str,
new_access_jti: &str,
new_refresh_jti: &str,
) -> tranquil_db_traits::SessionRefreshData {
let now = chrono::Utc::now();
tranquil_db_traits::SessionRefreshData {
old_refresh_jti: old_refresh_jti.to_string(),
session_id,
new_access_jti: new_access_jti.to_string(),
new_refresh_jti: new_refresh_jti.to_string(),
new_access_expires_at: now + chrono::Duration::minutes(120),
new_refresh_expires_at: now + chrono::Duration::days(90),
}
}
#[test]
fn legacy_refresh_grace_replays_within_window() {
use tranquil_db_traits::{RefreshGraceLookup, RefreshSessionResult};
let (_dir, ms) = open_fresh();
create_test_user(&ms, "did:plc:grace", "grace.test");
let ops = ms.session_ops();
let session_id = ops
.create_session(&legacy_session_create("did:plc:grace", "acc0", "ref0"))
.unwrap();
// The winning request rotates ref0 -> ref1.
let win = legacy_refresh_data(session_id, "ref0", "acc1", "ref1");
assert!(matches!(
ops.refresh_session_atomic(&win).unwrap(),
RefreshSessionResult::Success
));
// A racing client re-presents ref0; the up-front grace lookup points at
// the session's current tokens for re-minting (jtis, not signed JWTs),
// carrying the signing key so the caller can verify the presented token.
match ops.lookup_refresh_grace("ref0").unwrap() {
RefreshGraceLookup::Replay(replay) => {
assert_eq!(replay.did.as_str(), "did:plc:grace");
assert_eq!(replay.access_jti, "acc1");
assert_eq!(replay.refresh_jti, "ref1");
assert_eq!(replay.key_bytes, vec![7u8; 32]);
}
other => panic!("expected Replay, got {other:?}"),
}
// The atomic path (two requests both past the used-check) also yields
// the winner's current tokens rather than revoking.
let lose = legacy_refresh_data(session_id, "ref0", "accX", "refX");
match ops.refresh_session_atomic(&lose).unwrap() {
RefreshSessionResult::GraceReplay(replay) => {
assert_eq!(replay.access_jti, "acc1");
assert_eq!(replay.refresh_jti, "ref1");
}
other => panic!("expected GraceReplay, got {other:?}"),
}
// Crucially, the session is still alive on the winner's tokens — nobody
// got logged out.
let alive = ops.get_session_by_access_jti("acc1").unwrap();
assert!(
alive.is_some(),
"session must survive a benign concurrent refresh"
);
assert_eq!(alive.unwrap().refresh_jti, "ref1");
}
// Per-token grace: a token two rotations stale but rotated moments ago is
// still within its own grace window, so it must replay the session's CURRENT
// tokens (acc2/ref2) rather than revoking. Regression for the defect where
// only the immediate predecessor was replayable.
#[test]
fn legacy_refresh_superseded_token_within_grace_replays() {
use tranquil_db_traits::{RefreshGraceLookup, RefreshSessionResult};
let (_dir, ms) = open_fresh();
create_test_user(&ms, "did:plc:reuse", "reuse.test");
let ops = ms.session_ops();
let session_id = ops
.create_session(&legacy_session_create("did:plc:reuse", "acc0", "ref0"))
.unwrap();
ops.refresh_session_atomic(&legacy_refresh_data(session_id, "ref0", "acc1", "ref1"))
.unwrap();
ops.refresh_session_atomic(&legacy_refresh_data(session_id, "ref1", "acc2", "ref2"))
.unwrap();
// ref0 is two rotations back but was rotated just now: still in window.
match ops.lookup_refresh_grace("ref0").unwrap() {
RefreshGraceLookup::Replay(replay) => {
assert_eq!(replay.access_jti, "acc2");
assert_eq!(replay.refresh_jti, "ref2");
}
other => panic!("expected Replay, got {other:?}"),
}
match ops
.refresh_session_atomic(&legacy_refresh_data(session_id, "ref0", "z", "z"))
.unwrap()
{
RefreshSessionResult::GraceReplay(replay) => {
assert_eq!(replay.access_jti, "acc2");
assert_eq!(replay.refresh_jti, "ref2");
}
other => panic!("expected GraceReplay, got {other:?}"),
}
// The session is still alive on the current tokens — nobody logged out.
assert!(ops.get_session_by_access_jti("acc2").unwrap().is_some());
}
// An old-format (12-byte, no rotated_at) used marker has unknown rotation
// time and must classify as Compromised, never Replay.
#[test]
fn legacy_old_format_marker_is_compromise() {
use tranquil_db_traits::RefreshGraceLookup;
let (_dir, ms) = open_fresh();
create_test_user(&ms, "did:plc:oldfmt", "oldfmt.test");
let ops = ms.session_ops();
let session_id = ops
.create_session(&legacy_session_create("did:plc:oldfmt", "acc0", "ref0"))
.unwrap();
// Write a pre-upgrade 12-byte marker (TTL prefix + session_id) directly.
let mut marker = 0u64.to_be_bytes().to_vec();
marker.extend_from_slice(&session_id.as_i32().to_be_bytes());
assert_eq!(marker.len(), 12);
ms.partition(Partition::Auth)
.insert(
super::sessions::session_used_refresh_key("ref0").as_slice(),
marker,
)
.unwrap();
assert!(matches!(
ops.lookup_refresh_grace("ref0").unwrap(),
RefreshGraceLookup::Compromised { .. }
));
}
// A current-format marker whose rotation predates the grace window is genuine
// reuse: lookup classifies Compromised, the atomic path returns Compromise and
// revokes, and the session is gone afterwards.
#[test]
fn legacy_refresh_stale_rotation_outside_grace_is_compromise() {
use tranquil_db_traits::{
REFRESH_GRACE_PERIOD_SECS, RefreshGraceLookup, RefreshSessionResult,
};
let (_dir, ms) = open_fresh();
create_test_user(&ms, "did:plc:stale", "stale.test");
let ops = ms.session_ops();
let session_id = ops
.create_session(&legacy_session_create("did:plc:stale", "acc0", "ref0"))
.unwrap();
// Overwrite ref0's used marker with a current 20-byte marker rotated 3h ago,
// well outside the 2h grace window.
let stale_rotated_at_ms =
(chrono::Utc::now() - chrono::Duration::hours(3)).timestamp_millis();
let refresh_expires_at_ms =
(chrono::Utc::now() + chrono::Duration::days(90)).timestamp_millis();
assert!(
stale_rotated_at_ms
< (chrono::Utc::now() - chrono::Duration::seconds(REFRESH_GRACE_PERIOD_SECS))
.timestamp_millis()
);
ms.partition(Partition::Auth)
.insert(
super::sessions::session_used_refresh_key("ref0").as_slice(),
super::sessions::serialize_used_refresh_value(
refresh_expires_at_ms,
session_id.as_i32(),
stale_rotated_at_ms,
),
)
.unwrap();
match ops.lookup_refresh_grace("ref0").unwrap() {
RefreshGraceLookup::Compromised { key_bytes, .. } => {
assert_eq!(key_bytes, vec![7u8; 32]);
}
other => panic!("expected Compromised, got {other:?}"),
}
assert!(matches!(
ops.refresh_session_atomic(&legacy_refresh_data(session_id, "ref0", "z", "z"))
.unwrap(),
RefreshSessionResult::Compromise
));
// The session was actually revoked.
assert!(ops.get_session_by_access_jti("acc0").unwrap().is_none());
}
#[test]
fn reopen_preserves_partitions() {
let dir = tempfile::TempDir::new().unwrap();
crates/tranquil-store/src/metastore/session_ops.rs
+108 -65
View File
@@ -20,9 +20,9 @@ use super::user_hash::UserHashMap;
use super::users::UserValue;
use tranquil_db_traits::{
AppPasswordCreate, AppPasswordRecord, LoginType, RefreshSessionResult, SessionForRefresh,
SessionId, SessionListItem, SessionMfaStatus, SessionRefreshData, SessionToken,
SessionTokenCreate,
AppPasswordCreate, AppPasswordRecord, LoginType, REFRESH_GRACE_PERIOD_SECS, RefreshGraceLookup,
RefreshGraceReplay, RefreshSessionResult, SessionForRefresh, SessionId, SessionListItem,
SessionMfaStatus, SessionRefreshData, SessionToken, SessionTokenCreate,
};
use tranquil_types::Did;
@@ -171,6 +171,84 @@ impl SessionOps {
);
}
pub fn lookup_refresh_grace(
&self,
refresh_jti: &str,
) -> Result<RefreshGraceLookup, MetastoreError> {
let used_key = session_used_refresh_key(refresh_jti);
let used = self
.auth
.get(used_key.as_slice())
.map_err(MetastoreError::Fjall)?;
let Some(raw) = used else {
return Ok(RefreshGraceLookup::NotUsed);
};
let (session_id, rotated_at_ms) = deserialize_used_refresh_value(&raw)
.ok_or(MetastoreError::CorruptData("corrupt used refresh value"))?;
// Marker without a live session, or without a loadable user key, degrades
// to NotUsed: we cannot verify the presented token, so we never mutate.
let Some(session) = self.load_session_by_id(session_id)? else {
return Ok(RefreshGraceLookup::NotUsed);
};
let user_hash = self.resolve_user_hash_from_did(&session.did);
let Some(user) = self.load_user_value(user_hash)? else {
return Ok(RefreshGraceLookup::NotUsed);
};
// Per-token grace measured from this token's own rotation time. A legacy
// marker (no rotated_at) is treated as outside the window.
let grace_cutoff_ms =
(Utc::now() - chrono::Duration::seconds(REFRESH_GRACE_PERIOD_SECS)).timestamp_millis();
if let Some(rotated_at_ms) = rotated_at_ms
&& rotated_at_ms > grace_cutoff_ms
{
return Ok(RefreshGraceLookup::Replay(self.build_grace_replay(
&session,
user.key_bytes,
user.encryption_version,
)?));
}
Ok(RefreshGraceLookup::Compromised {
session_id: SessionId::new(session_id),
key_bytes: user.key_bytes,
encryption_version: user.encryption_version,
})
}
/// Assemble the session's current token identity plus its signing key for a
/// grace-window replay.
fn build_grace_replay(
&self,
session: &SessionTokenValue,
key_bytes: Vec<u8>,
encryption_version: i32,
) -> Result<RefreshGraceReplay, MetastoreError> {
let did = Did::new(session.did.clone())
.map_err(|_| MetastoreError::CorruptData("invalid session did"))?;
let access_expires_at =
DateTime::<Utc>::from_timestamp_millis(session.access_expires_at_ms)
.ok_or(MetastoreError::CorruptData("invalid access expiry"))?;
let refresh_expires_at =
DateTime::<Utc>::from_timestamp_millis(session.refresh_expires_at_ms)
.ok_or(MetastoreError::CorruptData("invalid refresh expiry"))?;
Ok(RefreshGraceReplay {
did,
scope: session.scope.clone(),
controller_did: session
.controller_did
.clone()
.and_then(|d| Did::new(d).ok()),
access_jti: session.access_jti.clone(),
refresh_jti: session.refresh_jti.clone(),
access_expires_at,
refresh_expires_at,
key_bytes,
encryption_version,
})
}
fn collect_sessions_for_did(
&self,
user_hash: UserHash,
@@ -540,51 +618,6 @@ impl SessionOps {
.collect())
}
pub fn check_refresh_token_used(
&self,
refresh_jti: &str,
) -> Result<Option<SessionId>, MetastoreError> {
let key = session_used_refresh_key(refresh_jti);
match self
.auth
.get(key.as_slice())
.map_err(MetastoreError::Fjall)?
{
Some(raw) => Ok(deserialize_used_refresh_value(&raw).map(SessionId::new)),
None => Ok(None),
}
}
pub fn mark_refresh_token_used(
&self,
refresh_jti: &str,
session_id: SessionId,
) -> Result<bool, MetastoreError> {
let key = session_used_refresh_key(refresh_jti);
let existing = self
.auth
.get(key.as_slice())
.map_err(MetastoreError::Fjall)?;
match existing {
Some(_) => Ok(false),
None => {
let session = self.load_session_by_id(session_id.as_i32())?;
let expires_at_ms = session
.map(|s| s.refresh_expires_at_ms)
.unwrap_or(Utc::now().timestamp_millis().saturating_add(86_400_000));
self.auth
.insert(
key.as_slice(),
serialize_used_refresh_value(expires_at_ms, session_id.as_i32()),
)
.map_err(MetastoreError::Fjall)?;
Ok(true)
}
}
}
pub fn list_app_passwords(
&self,
user_id: Uuid,
@@ -837,49 +870,59 @@ impl SessionOps {
.map_err(MetastoreError::Fjall)?;
if already_used.is_some() {
let mut batch = self.db.batch();
let session = self.load_session_by_id(data.session_id.as_i32())?;
if let Some(s) = session {
self.delete_session_indexes(&mut batch, &s);
// The old refresh token was already rotated. Within the per-token
// grace window, replay the session's current tokens so this
// benignly-racing client keeps a working session instead of being
// revoked.
match self.lookup_refresh_grace(&data.old_refresh_jti)? {
RefreshGraceLookup::Replay(replay) => {
return Ok(RefreshSessionResult::GraceReplay(replay));
}
RefreshGraceLookup::Compromised { .. } | RefreshGraceLookup::NotUsed => {
// Outside the grace window: genuine reuse. Revoke the session.
let mut batch = self.db.batch();
if let Some(s) = self.load_session_by_id(data.session_id.as_i32())? {
self.delete_session_indexes(&mut batch, &s);
}
batch.commit().map_err(MetastoreError::Fjall)?;
return Ok(RefreshSessionResult::Compromise);
}
}
batch.commit().map_err(MetastoreError::Fjall)?;
return Ok(RefreshSessionResult::TokenAlreadyUsed);
}
let mut session = match self.load_session_by_id(data.session_id.as_i32())? {
Some(s) => s,
None => return Ok(RefreshSessionResult::ConcurrentRefresh),
None => return Ok(RefreshSessionResult::Compromise),
};
if session.refresh_jti != data.old_refresh_jti {
return Ok(RefreshSessionResult::ConcurrentRefresh);
return Ok(RefreshSessionResult::Compromise);
}
let user_hash = self.resolve_user_hash_from_did(&session.did);
let old_access_jti = session.access_jti.clone();
let old_refresh_jti = session.refresh_jti.clone();
let rotated_at_ms = Utc::now().timestamp_millis();
session.access_jti = data.new_access_jti.clone();
session.refresh_jti = data.new_refresh_jti.clone();
session.access_expires_at_ms = data.new_access_expires_at.timestamp_millis();
session.refresh_expires_at_ms = data.new_refresh_expires_at.timestamp_millis();
session.updated_at_ms = Utc::now().timestamp_millis();
session.updated_at_ms = rotated_at_ms;
let new_access_index = SessionIndexValue {
user_hash: user_hash.raw(),
session_id: session.id,
};
let new_refresh_index = SessionIndexValue {
let index = SessionIndexValue {
user_hash: user_hash.raw(),
session_id: session.id,
};
let mut batch = self.db.batch();
// Record the rotation time so a benign replay of this token can be graced
// from its own rotation moment.
batch.insert(
&self.auth,
used_key.as_slice(),
serialize_used_refresh_value(session.refresh_expires_at_ms, session.id),
serialize_used_refresh_value(session.refresh_expires_at_ms, session.id, rotated_at_ms),
);
batch.remove(
@@ -899,12 +942,12 @@ impl SessionOps {
batch.insert(
&self.auth,
session_by_access_key(&data.new_access_jti).as_slice(),
new_access_index.serialize(session.refresh_expires_at_ms),
index.serialize(session.refresh_expires_at_ms),
);
batch.insert(
&self.auth,
session_by_refresh_key(&data.new_refresh_jti).as_slice(),
new_refresh_index.serialize(session.refresh_expires_at_ms),
index.serialize(session.refresh_expires_at_ms),
);
batch.insert(
&self.auth,
crates/tranquil-store/src/metastore/sessions.rs
+41 -13
View File
@@ -209,25 +209,34 @@ pub fn session_id_counter_key() -> SmallVec<[u8; 128]> {
KeyBuilder::new().tag(KeyTag::SESSION_ID_COUNTER).build()
}
fn serialize_ttl_i32(expires_at_ms: i64, session_id: i32) -> Vec<u8> {
let mut buf = Vec::with_capacity(12);
buf.extend_from_slice(&u64::try_from(expires_at_ms).unwrap_or(0).to_be_bytes());
buf.extend_from_slice(&session_id.to_be_bytes());
buf
}
fn deserialize_ttl_i32(bytes: &[u8]) -> Option<i32> {
let rest = bytes.get(8..)?;
let arr: [u8; 4] = rest.try_into().ok()?;
Some(i32::from_be_bytes(arr))
}
pub fn serialize_used_refresh_value(expires_at_ms: i64, session_id: i32) -> Vec<u8> {
serialize_ttl_i32(expires_at_ms, session_id)
pub fn serialize_used_refresh_value(
expires_at_ms: i64,
session_id: i32,
rotated_at_ms: i64,
) -> Vec<u8> {
let mut buf = Vec::with_capacity(20);
buf.extend_from_slice(&u64::try_from(expires_at_ms).unwrap_or(0).to_be_bytes());
buf.extend_from_slice(&session_id.to_be_bytes());
buf.extend_from_slice(&rotated_at_ms.to_be_bytes());
buf
}
pub fn deserialize_used_refresh_value(bytes: &[u8]) -> Option<i32> {
deserialize_ttl_i32(bytes)
/// Decode a used-refresh marker. The 20-byte format carries the rotation time;
/// the legacy 12-byte format predates it (rotation time unknown), so callers
/// must treat its `None` as outside the grace window.
pub fn deserialize_used_refresh_value(bytes: &[u8]) -> Option<(i32, Option<i64>)> {
let session_id = deserialize_ttl_i32(&bytes[..12.min(bytes.len())])?;
let rotated_at_ms = bytes
.get(12..20)
.and_then(|s| <[u8; 8]>::try_from(s).ok())
.map(i64::from_be_bytes);
Some((session_id, rotated_at_ms))
}
fn serialize_ttl_i64(timestamp_ms: i64) -> Vec<u8> {
@@ -336,10 +345,29 @@ mod tests {
fn used_refresh_value_roundtrip() {
let session_id = 7;
let expires_at_ms = 1700000600000i64;
let bytes = serialize_used_refresh_value(expires_at_ms, session_id);
let rotated_at_ms = 1700000123000i64;
let bytes = serialize_used_refresh_value(expires_at_ms, session_id, rotated_at_ms);
assert_eq!(bytes.len(), 20);
let ttl = u64::from_be_bytes(bytes[..8].try_into().unwrap());
assert_eq!(ttl, u64::try_from(expires_at_ms).unwrap_or(0));
assert_eq!(deserialize_used_refresh_value(&bytes), Some(session_id));
assert_eq!(
deserialize_used_refresh_value(&bytes),
Some((session_id, Some(rotated_at_ms)))
);
}
#[test]
fn used_refresh_value_legacy_12_byte_decodes_without_rotated_at() {
// Pre-upgrade markers carry no rotation time; rotated_at must be None.
let session_id = 9i32;
let expires_at_ms = 1700000600000i64;
let mut bytes = u64::try_from(expires_at_ms).unwrap().to_be_bytes().to_vec();
bytes.extend_from_slice(&session_id.to_be_bytes());
assert_eq!(bytes.len(), 12);
assert_eq!(
deserialize_used_refresh_value(&bytes),
Some((session_id, None))
);
}
#[test]