e870254
use crate::jwt::authority::{AccessRecord, Authority, UserRecord};
use crate::jwt::{AppClaims, JwtError};
use crate::models::users::{user_sessions, users};
use crate::session::Session;
use async_trait::async_trait;
use chrono::{DateTime, Duration, Utc};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, Set};
use uuid::Uuid;
use crate::server::share::{crypto, user as store};
use crate::server::{AuthError, AuthResult, AuthSvc};
const MAX_BATCH_PROFILES: usize = store::MAX_BATCH_USER_IDS;
/// 会话凭据摘要的算法标识，落在 `user_sessions.hash_key_id`。
pub const SESSION_HASH_KEY_ID: &str = "logical-session-v1";
/// 会话 ID 的摘要。会话 ID 本身即凭据，库里只存摘要。
pub fn session_token_hash(id: Uuid) -> String {
crypto::sha256_hex(id.to_string().as_bytes())
}
pub(crate) fn record(user: users::Model) -> UserRecord {
UserRecord {
id: user.id,
username: user.username,
lowername: user.lowername,
displayname: user.displayname,
birth_date: user.birth_date,
email: user.email,
public_show_email: user.public_show_email,
email_verified_at: user.email_verified_at,
allow_use: user.allow_use,
forbid_login: user.forbid_login,
forbid_login_reason: user.forbid_login_reason,
endtime_forbid_login_at: user.endtime_forbid_login_at,
is_admin: user.is_admin,
bot: user.bot,
system: user.system,
created_at: user.created_at,
updated_at: user.updated_at,
security_version: user.security_version,
version: user.version,
deleted_at: user.deleted_at,
}
}
fn invalid(error: impl std::fmt::Display) -> JwtError {
tracing::debug!(%error, "authority check failed");
JwtError::Revoked
}
impl AuthSvc {
#[tracing::instrument(skip_all, fields(user_id = user.id, mfa = mfa, request_id = %request_id.unwrap_or_default()))]
pub async fn bind_auth_session(
&self,
session: &Session,
user: &users::Model,
mfa: bool,
request_id: Option<&str>,
) -> AuthResult<()> {
let now = Utc::now();
let id = Uuid::now_v7();
let ttl = i64::try_from(self.session_max_age_secs())
.unwrap_or(86400)
.max(1);
user_sessions::ActiveModel {
id: Set(id),
user_id: Set(user.id),
token_hash: Set(session_token_hash(id)),
hash_key_id: Set(SESSION_HASH_KEY_ID.into()),
security_version: Set(user.security_version),
expires_at: Set(now + Duration::seconds(ttl)),
mfa_verified_at: Set(mfa.then_some(now)),
ip_address: Set(session.client_ip()?),
user_agent: Set(session.user_agent()?),
..Default::default()
}
.insert(self.db().write())
.await?;
session.set("auth_session_id", id.to_string())?;
session.set("mfa_verified", mfa)?;
session.set_user_id(user.id)?;
session.set_security_version(user.security_version)?;
session.renew();
Ok(())
}
#[tracing::instrument(skip_all, fields(user_id = user_id, session_id = %session_id, request_id = %request_id.unwrap_or_default()))]
pub async fn check_session(
&self,
user_id: i64,
version: i32,
session_id: &str,
request_id: Option<&str>,
) -> Result<AccessRecord, JwtError> {
// security_version 判定一律读主库：副本延迟会让刚吊销的旧版本短暂放行。
let user = users::Entity::find_by_id(user_id)
.one(self.db().write())
.await
.map_err(invalid)?
.ok_or(JwtError::Revoked)?;
let user = record(user);
user.require_active()?;
if version <= 0 || version != user.security_version {
return Err(JwtError::Revoked);
}
let id = Uuid::parse_str(session_id).map_err(invalid)?;
let row = user_sessions::Entity::find_by_id(id)
.one(self.db().write())
.await
.map_err(invalid)?
.ok_or(JwtError::Revoked)?;
// `token_hash` 是会话凭据的摘要：只有拿着明文会话 ID 的一方能算出来。
// 不校验它，单有数据库读权限的人就能直接拿行里的 ID 复会话。
if !crate::server::share::crypto::constant_time_eq(
row.token_hash.as_bytes(),
session_token_hash(id).as_bytes(),
) {
return Err(JwtError::Revoked);
}
if row.user_id != user_id
|| row.security_version != version
|| row.revoked_at.is_some()
|| row.expires_at <= Utc::now()
{
return Err(JwtError::Revoked);
}
let mut record = AccessRecord::from(&user);
record.mfa_verified = row.mfa_verified_at.is_some();
Ok(record)
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
pub async fn require_session(
&self,
session: &Session,
request_id: Option<&str>,
) -> AuthResult<i64> {
let user_id = session.user_id()?.ok_or(AuthError::Unauthenticated)?;
let version = session
.security_version()?
.ok_or(AuthError::Unauthenticated)?;
let id = session
.get::<String>("auth_session_id")?
.ok_or(AuthError::Unauthenticated)?;
self.check_session(user_id, version, &id, request_id)
.await
.map_err(|_| AuthError::Unauthenticated)?;
Ok(user_id)
}
/// 会话建立时间（`user_sessions.created_at`），供 `max_age` 与 id_token 的 `auth_time` 使用。
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
pub async fn session_created_at(
&self,
session: &Session,
request_id: Option<&str>,
) -> AuthResult<Option<DateTime<Utc>>> {
let Some(id) = session.get::<String>("auth_session_id")? else {
return Ok(None);
};
let Ok(id) = Uuid::parse_str(&id) else {
return Ok(None);
};
let row = user_sessions::Entity::find_by_id(id)
.one(self.db().read())
.await?;
Ok(row.map(|row| row.created_at))
}
}
#[async_trait]
impl Authority for AuthSvc {
async fn validate(&self, _: &str, claims: &AppClaims) -> Result<AccessRecord, JwtError> {
self.check_session(
claims.user_id,
claims.security_version,
claims.session_id.as_deref().ok_or(JwtError::Revoked)?,
crate::envelope::context::request_id().as_deref(),
)
.await
}
async fn user(&self, user_id: i64) -> Result<UserRecord, JwtError> {
let user = users::Entity::find_by_id(user_id)
.one(self.db().read())
.await
.map_err(invalid)?
.ok_or(JwtError::Revoked)?;
Ok(record(user))
}
async fn users(&self, user_ids: &[i64]) -> Result<Vec<UserRecord>, JwtError> {
if user_ids.len() > MAX_BATCH_PROFILES {
return Err(JwtError::Malformed);
}
let mut unique: Vec<i64> = Vec::with_capacity(user_ids.len());
for user_id in user_ids.iter().copied().filter(|user_id| *user_id > 0) {
if !unique.contains(&user_id) {
unique.push(user_id);
}
}
if unique.is_empty() {
return Ok(Vec::new());
}
let rows = users::Entity::find()
.filter(users::Column::Id.is_in(unique))
.all(self.db().read())
.await
.map_err(invalid)?;
Ok(rows.into_iter().map(record).collect())
}
async fn session(
&self,
user_id: i64,
version: i32,
session_id: &str,
) -> Result<AccessRecord, JwtError> {
self.check_session(
user_id,
version,
session_id,
crate::envelope::context::request_id().as_deref(),
)
.await
}
async fn update(&self, user_id: i64, patch: serde_json::Value) -> Result<UserRecord, JwtError> {
let user = store::load_active_user(self.db(), user_id)
.await
.map_err(invalid)?;
let version = patch
.get("version")
.and_then(|v| v.as_i64())
.ok_or(JwtError::Malformed)?;
if i64::from(user.version) != version {
return Err(JwtError::Malformed);
}
let mut active = user.clone().into_active_model();
if let Some(name) = patch.get("username").and_then(|v| v.as_str()) {
let name = crate::server::share::validate::validate_username(name).map_err(invalid)?;
active.lowername = Set(name.to_lowercase());
active.username = Set(name);
}
if let Some(value) = patch.get("displayname") {
active.displayname = Set(match value.as_str() {
Some(value) => Some(
crate::server::share::validate::validate_display_name(value)
.map_err(invalid)?,
),
None if value.is_null() => None,
_ => return Err(JwtError::Malformed),
});
}
if let Some(value) = patch.get("public_show_email") {
active.public_show_email = Set(value.as_bool().ok_or(JwtError::Malformed)?);
}
active.version = Set(user.version.checked_add(1).ok_or(JwtError::Malformed)?);
active.updated_at = Set(Utc::now());
let updated = users::Entity::update_many()
.set(active)
.filter(users::Column::Id.eq(user_id))
.filter(users::Column::Version.eq(user.version))
.exec_with_returning(self.db().write())
.await
.map_err(invalid)?
.pop()
.ok_or(JwtError::Malformed)?;
crate::server::me::service::invalidate_me_cache(self.cache(), user_id).await;
Ok(record(updated))
}
}