use super::*;
use crate::runtime::native_catalog::{NativeModel, native_model};
use sqlx::Row;
pub(super) enum RotateOutcome {
Rotated {
new_refresh_token: String,
family: TokenFamilyRow,
},
NotFound,
Reuse,
}
pub(super) struct TokenFamilyRow {
pub family_id: String,
pub user_id: String,
pub principal_id: String,
pub tenant_id: String,
pub project_id: String,
pub device_id: String,
pub revoked: bool,
}
fn token_family_model() -> NativeModel {
native_model(
"udb.core.authn.entity.v1.TokenFamily",
&[
"family_id",
"session_id",
"user_id",
"principal_id",
"tenant_id",
"project_id",
"device_id",
"current_refresh_jti_hash",
"previous_refresh_jti_hash",
"reuse_detected_at",
"revoked_at",
"revocation_reason",
"updated_at",
],
)
}
impl AuthnServiceImpl {
fn require_family_pool(&self) -> Result<&PgPool, Status> {
self.pg_pool.as_ref().ok_or_else(|| {
Status::failed_precondition(
"refresh-token rotation requires the native Postgres auth store",
)
})
}
fn refresh_jti_hash(&self, jti: &str) -> String {
authn::hash_secret(&format!("refresh-jti:{jti}"), &self.hash_key())
}
pub(super) async fn mint_refresh_family(
&self,
user_id: &str,
principal_id: &str,
tenant_id: &str,
project_id: &str,
device_id: &str,
session_handle: &str,
now: u64,
) -> Result<String, Status> {
let _ = now;
let Some(pool) = self.pg_pool.as_ref() else {
return Ok(String::new());
};
let m = token_family_model();
let family_id = Uuid::new_v4().to_string();
let jti = Uuid::new_v4().simple().to_string();
let jti_hash = self.refresh_jti_hash(&jti);
let session_hash = if session_handle.trim().is_empty() {
String::new()
} else {
authn::hash_secret(session_handle, &self.hash_key())
};
if let Ok(runtime) = self.authn_runtime() {
let context = self.authn_context(tenant_id, project_id);
let record = authn_record([
("family_id", LogicalValue::String(family_id.clone())),
("session_id", LogicalValue::String(session_hash.clone())),
("user_id", LogicalValue::String(user_id.to_string())),
(
"principal_id",
LogicalValue::String(principal_id.to_string()),
),
("tenant_id", LogicalValue::String(tenant_id.to_string())),
("project_id", LogicalValue::String(project_id.to_string())),
("device_id", LogicalValue::String(device_id.to_string())),
(
"current_refresh_jti_hash",
LogicalValue::String(jti_hash.clone()),
),
]);
runtime
.native_entity_write_for_service(
"authn",
&context,
"udb.core.authn.entity.v1.TokenFamily",
record,
crate::ir::ConflictStrategy::Error,
)
.await
.map_err(|err| Status::internal(format!("mint refresh family failed: {err}")))?;
return Ok(authn::token_family::format_refresh_token(&family_id, &jti));
}
let sql = format!(
"INSERT INTO {rel} ({id}, {session}, {user}, {principal}, {tenant}, {project}, {device}, {cur}) \
VALUES ($1::UUID, $2, $3, $4, $5, $6, $7, $8)",
rel = m.relation,
id = m.q("family_id"),
session = m.q("session_id"),
user = m.q("user_id"),
principal = m.q("principal_id"),
tenant = m.q("tenant_id"),
project = m.q("project_id"),
device = m.q("device_id"),
cur = m.q("current_refresh_jti_hash"),
);
sqlx::query(&sql)
.bind(&family_id)
.bind(&session_hash)
.bind(user_id)
.bind(principal_id)
.bind(tenant_id)
.bind(project_id)
.bind(device_id)
.bind(&jti_hash)
.execute(pool)
.await
.map_err(|err| Status::internal(format!("mint refresh family failed: {err}")))?;
Ok(authn::token_family::format_refresh_token(&family_id, &jti))
}
pub(super) async fn revoke_families_for_session(
&self,
session_handle: &str,
) -> Result<u64, Status> {
let Some(pool) = self.pg_pool.as_ref() else {
return Ok(0);
};
if session_handle.trim().is_empty() {
return Ok(0);
}
let session_hash = authn::hash_secret(session_handle, &self.hash_key());
if let Ok(runtime) = self.authn_runtime() {
let context = self.authn_context("", "");
let mut assignments = std::collections::BTreeMap::new();
assignments.insert("revoked_at".to_string(), LogicalAssignment::ServerNow);
assignments.insert(
"revocation_reason".to_string(),
authn_set(LogicalValue::String("logout".to_string())),
);
assignments.insert("updated_at".to_string(), LogicalAssignment::ServerNow);
let op = LogicalUpdate {
message_type: "udb.core.authn.entity.v1.TokenFamily".to_string(),
filter: authn_and(vec![
authn_eq("session_id", LogicalValue::String(session_hash.clone())),
LogicalFilter::IsNull("revoked_at".to_string()),
]),
assignments,
return_fields: Vec::new(),
require_affected: false,
};
return runtime
.native_entity_update_for_service("authn", &context, op)
.await
.map(|(affected, _)| affected)
.map_err(|err| {
Status::internal(format!("revoke families for session failed: {err}"))
});
}
let m = token_family_model();
let sql = format!(
"UPDATE {rel} SET {revoked} = COALESCE({revoked}, NOW()), \
{reason} = COALESCE(NULLIF({reason}, ''), 'logout'), {updated} = NOW() \
WHERE {session} = $1 AND {revoked} IS NULL",
rel = m.relation,
revoked = m.q("revoked_at"),
reason = m.q("revocation_reason"),
updated = m.q("updated_at"),
session = m.q("session_id"),
);
Ok(sqlx::query(&sql)
.bind(&session_hash)
.execute(pool)
.await
.map_err(|err| Status::internal(format!("revoke families for session failed: {err}")))?
.rows_affected())
}
pub(super) async fn revoke_families_for_principal(
&self,
principal_id: &str,
) -> Result<u64, Status> {
let Some(pool) = self.pg_pool.as_ref() else {
return Ok(0);
};
if principal_id.trim().is_empty() {
return Ok(0);
}
if let Ok(runtime) = self.authn_runtime() {
let context = self.authn_context("", "");
let mut total = 0u64;
for field in ["user_id", "principal_id"] {
let mut assignments = std::collections::BTreeMap::new();
assignments.insert("revoked_at".to_string(), LogicalAssignment::ServerNow);
assignments.insert(
"revocation_reason".to_string(),
authn_set(LogicalValue::String("logout_all".to_string())),
);
assignments.insert("updated_at".to_string(), LogicalAssignment::ServerNow);
let op = LogicalUpdate {
message_type: "udb.core.authn.entity.v1.TokenFamily".to_string(),
filter: authn_and(vec![
authn_eq(field, LogicalValue::String(principal_id.to_string())),
LogicalFilter::IsNull("revoked_at".to_string()),
]),
assignments,
return_fields: Vec::new(),
require_affected: false,
};
let (affected, _) = runtime
.native_entity_update_for_service("authn", &context, op)
.await
.map_err(|err| {
Status::internal(format!("revoke families for principal failed: {err}"))
})?;
total += affected;
}
return Ok(total);
}
let m = token_family_model();
let sql = format!(
"UPDATE {rel} SET {revoked} = COALESCE({revoked}, NOW()), \
{reason} = COALESCE(NULLIF({reason}, ''), 'logout_all'), {updated} = NOW() \
WHERE ({user} = $1 OR {principal} = $1) AND {revoked} IS NULL",
rel = m.relation,
revoked = m.q("revoked_at"),
reason = m.q("revocation_reason"),
updated = m.q("updated_at"),
user = m.q("user_id"),
principal = m.q("principal_id"),
);
Ok(sqlx::query(&sql)
.bind(principal_id)
.execute(pool)
.await
.map_err(|err| {
Status::internal(format!("revoke families for principal failed: {err}"))
})?
.rows_affected())
}
pub(super) async fn rotate_refresh_family(
&self,
family_id: &str,
presented_jti: &str,
now: u64,
) -> Result<RotateOutcome, Status> {
let _ = now;
let pool = self.require_family_pool()?;
let m = token_family_model();
let presented_hash = self.refresh_jti_hash(presented_jti);
let new_jti = Uuid::new_v4().simple().to_string();
let new_hash = self.refresh_jti_hash(&new_jti);
let rotate_sql = format!(
"UPDATE {rel} SET {prev} = {cur}, {cur} = $3, {updated} = NOW() \
WHERE {id} = $1::UUID AND {cur} = $2 AND {revoked} IS NULL \
RETURNING {user}::TEXT AS user_id, {principal}::TEXT AS principal_id, \
{tenant}::TEXT AS tenant_id, COALESCE({project}::TEXT,'') AS project_id, \
COALESCE({device}::TEXT,'') AS device_id",
rel = m.relation,
prev = m.q("previous_refresh_jti_hash"),
cur = m.q("current_refresh_jti_hash"),
updated = m.q("updated_at"),
id = m.q("family_id"),
revoked = m.q("revoked_at"),
user = m.q("user_id"),
principal = m.q("principal_id"),
tenant = m.q("tenant_id"),
project = m.q("project_id"),
device = m.q("device_id"),
);
if let Some(row) = sqlx::query(&rotate_sql)
.bind(family_id)
.bind(&presented_hash)
.bind(&new_hash)
.fetch_optional(pool)
.await
.map_err(|err| Status::internal(format!("rotate refresh family failed: {err}")))?
{
let family = TokenFamilyRow {
family_id: family_id.to_string(),
user_id: row.try_get("user_id").unwrap_or_default(),
principal_id: row.try_get("principal_id").unwrap_or_default(),
tenant_id: row.try_get("tenant_id").unwrap_or_default(),
project_id: row.try_get("project_id").unwrap_or_default(),
device_id: row.try_get("device_id").unwrap_or_default(),
revoked: false,
};
return Ok(RotateOutcome::Rotated {
new_refresh_token: authn::token_family::format_refresh_token(family_id, &new_jti),
family,
});
}
let inspect_sql = format!(
"SELECT {user}::TEXT AS user_id, {principal}::TEXT AS principal_id, \
{tenant}::TEXT AS tenant_id, COALESCE({project}::TEXT,'') AS project_id, \
COALESCE({device}::TEXT,'') AS device_id, \
({prev} = $2) AS is_previous, ({revoked} IS NOT NULL) AS is_revoked \
FROM {rel} WHERE {id} = $1::UUID",
user = m.q("user_id"),
principal = m.q("principal_id"),
tenant = m.q("tenant_id"),
project = m.q("project_id"),
device = m.q("device_id"),
prev = m.q("previous_refresh_jti_hash"),
revoked = m.q("revoked_at"),
rel = m.relation,
id = m.q("family_id"),
);
let Some(row) = sqlx::query(&inspect_sql)
.bind(family_id)
.bind(&presented_hash)
.fetch_optional(pool)
.await
.map_err(|err| Status::internal(format!("inspect refresh family failed: {err}")))?
else {
return Ok(RotateOutcome::NotFound);
};
let is_previous: bool = row.try_get("is_previous").unwrap_or(false);
let is_revoked: bool = row.try_get("is_revoked").unwrap_or(false);
let family = TokenFamilyRow {
family_id: family_id.to_string(),
user_id: row.try_get("user_id").unwrap_or_default(),
principal_id: row.try_get("principal_id").unwrap_or_default(),
tenant_id: row.try_get("tenant_id").unwrap_or_default(),
project_id: row.try_get("project_id").unwrap_or_default(),
device_id: row.try_get("device_id").unwrap_or_default(),
revoked: is_revoked,
};
if is_previous || is_revoked {
self.revoke_family_on_reuse(&family).await?;
return Ok(RotateOutcome::Reuse);
}
Ok(RotateOutcome::NotFound)
}
async fn revoke_family_on_reuse(&self, family: &TokenFamilyRow) -> Result<(), Status> {
let pool = self.require_family_pool()?;
let m = token_family_model();
let sql = format!(
"UPDATE {rel} SET {revoked} = COALESCE({revoked}, NOW()), \
{reuse} = COALESCE({reuse}, NOW()), \
{reason} = COALESCE({reason}, 'refresh_reuse_detected') \
WHERE {id} = $1::UUID",
rel = m.relation,
revoked = m.q("revoked_at"),
reuse = m.q("reuse_detected_at"),
reason = m.q("revocation_reason"),
id = m.q("family_id"),
);
sqlx::query(&sql)
.bind(&family.family_id)
.execute(pool)
.await
.map_err(|err| Status::internal(format!("revoke reused family failed: {err}")))?;
let principal = if family.principal_id.is_empty() {
family.user_id.clone()
} else {
family.principal_id.clone()
};
if !principal.is_empty() {
let _ = self
.sessions
.revoke_all_for_principal(&principal, now_unix())
.await;
}
if family.revoked {
return Ok(());
}
self.metrics.record_refresh_reuse_detected();
self.emit_event(
AuthEvent::new(
topics::REFRESH_REUSE_DETECTED,
family.family_id.clone(),
family.tenant_id.clone(),
serde_json::json!({
"family_id": family.family_id,
"principal_id": family.principal_id,
"user_id": family.user_id,
"tenant_id": family.tenant_id,
"project_id": family.project_id,
"device_id": family.device_id,
"severity": "high",
"reason": "refresh_token_reuse",
}),
)
.with_correlation(format!("refresh-reuse:{}", family.family_id)),
)
.await;
Ok(())
}
}