b7cd2b1
use std::sync::{Arc, OnceLock};
use async_trait::async_trait;
use chrono::{DateTime, NaiveDate, Utc};
use serde::{Deserialize, Serialize};
use crate::{AppClaims, JwtError};
pub const MAX_BATCH_USER_IDS: usize = 500;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct UserRecord {
pub id: i64,
pub username: String,
pub lowername: String,
pub displayname: Option<String>,
pub birth_date: Option<NaiveDate>,
pub email: String,
pub public_show_email: bool,
pub email_verified_at: Option<DateTime<Utc>>,
pub allow_use: bool,
pub forbid_login: bool,
pub forbid_login_reason: Option<String>,
pub endtime_forbid_login_at: Option<DateTime<Utc>>,
pub is_admin: bool,
pub bot: bool,
pub system: bool,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub security_version: i32,
pub version: i32,
pub deleted_at: Option<DateTime<Utc>>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AccessRecord {
pub id: i64,
pub username: String,
pub allow_use: bool,
pub forbid_login: bool,
pub endtime_forbid_login_at: Option<DateTime<Utc>>,
pub is_admin: bool,
pub security_version: i32,
pub deleted_at: Option<DateTime<Utc>>,
/// 会话是否已通过第二因子；来源是 `user_sessions.mfa_verified_at`，令牌自身不再声明。
pub mfa_verified: bool,
}
impl From<&UserRecord> for AccessRecord {
fn from(user: &UserRecord) -> Self {
Self {
id: user.id,
username: user.username.clone(),
allow_use: user.allow_use,
forbid_login: user.forbid_login,
endtime_forbid_login_at: user.endtime_forbid_login_at,
is_admin: user.is_admin,
security_version: user.security_version,
deleted_at: user.deleted_at,
mfa_verified: false,
}
}
}
impl From<UserRecord> for AccessRecord {
fn from(user: UserRecord) -> Self {
Self::from(&user)
}
}
fn check_active(
id: i64,
allow_use: bool,
forbid_login: bool,
endtime_forbid_login_at: Option<DateTime<Utc>>,
deleted_at: Option<DateTime<Utc>>,
) -> Result<(), JwtError> {
if id <= 0
|| deleted_at.is_some()
|| !allow_use
|| (forbid_login && endtime_forbid_login_at.is_none_or(|until| until > Utc::now()))
{
return Err(JwtError::Revoked);
}
Ok(())
}
impl UserRecord {
pub fn require_active(&self) -> Result<(), JwtError> {
check_active(
self.id,
self.allow_use,
self.forbid_login,
self.endtime_forbid_login_at,
self.deleted_at,
)
}
}
impl AccessRecord {
pub fn require_active(&self) -> Result<(), JwtError> {
check_active(
self.id,
self.allow_use,
self.forbid_login,
self.endtime_forbid_login_at,
self.deleted_at,
)
}
}
/// 账号与会话状态的唯一来源。实现方只有两个：oauth 进程内的 `AuthSvc`（本地直读），
/// 以及 api / socket 进程内的 `GrpcAuthority`（走 auth.internal.v1.AuthorityService）。
#[async_trait]
pub trait Authority: Send + Sync {
async fn validate(&self, token: &str, claims: &AppClaims) -> Result<AccessRecord, JwtError>;
async fn user(&self, user_id: i64) -> Result<UserRecord, JwtError>;
async fn users(&self, user_ids: &[i64]) -> Result<Vec<UserRecord>, JwtError> {
let mut users = Vec::with_capacity(user_ids.len());
for user_id in user_ids {
users.push(self.user(*user_id).await?);
}
Ok(users)
}
async fn session(
&self,
user_id: i64,
version: i32,
session_id: &str,
) -> Result<AccessRecord, JwtError>;
async fn update(&self, user_id: i64, patch: serde_json::Value) -> Result<UserRecord, JwtError>;
}
static LOCAL: OnceLock<Arc<dyn Authority>> = OnceLock::new();
pub fn install(authority: Arc<dyn Authority>) -> Result<(), JwtError> {
LOCAL
.set(authority)
.map_err(|_| JwtError::Config("authority is already installed".into()))
}
fn installed() -> Result<&'static Arc<dyn Authority>, JwtError> {
LOCAL
.get()
.ok_or_else(|| JwtError::Config("jwt authority is not installed".into()))
}
pub async fn authenticate(token: &str) -> Result<AppClaims, JwtError> {
authenticate_for_audience(token, crate::require_installed()?.audience()).await
}
pub async fn authenticate_for_audience(token: &str, audience: &str) -> Result<AppClaims, JwtError> {
let mut claims = crate::verify_for_audience(token, audience)?;
if claims.token_type != "access" || claims.user_id <= 0 {
return Err(JwtError::Malformed);
}
let user = installed()?.validate(token, &claims).await?;
user.require_active()?;
if user.id != claims.user_id || user.security_version != claims.security_version {
return Err(JwtError::Revoked);
}
claims.is_admin = user.is_admin;
claims.mfa_verified = user.mfa_verified;
claims.username = Some(user.username);
Ok(claims)
}
pub async fn user(user_id: i64) -> Result<UserRecord, JwtError> {
installed()?.user(user_id).await
}
pub async fn users(user_ids: &[i64]) -> Result<Vec<UserRecord>, JwtError> {
if user_ids.is_empty() {
return Ok(Vec::new());
}
if user_ids.len() > MAX_BATCH_USER_IDS {
return Err(JwtError::Malformed);
}
installed()?.users(user_ids).await
}
pub async fn session(
user_id: i64,
version: i32,
session_id: &str,
) -> Result<AccessRecord, JwtError> {
installed()?.session(user_id, version, session_id).await
}
pub async fn update(user_id: i64, patch: serde_json::Value) -> Result<UserRecord, JwtError> {
installed()?.update(user_id, patch).await
}
#[cfg(test)]
mod tests {
use super::*;
use base64::Engine;
use std::sync::atomic::{AtomicBool, AtomicI32, Ordering};
struct Fixture {
version: AtomicI32,
disabled: AtomicBool,
revoked: AtomicBool,
}
#[async_trait]
impl Authority for Fixture {
async fn validate(&self, _: &str, claims: &AppClaims) -> Result<AccessRecord, JwtError> {
if self.revoked.load(Ordering::SeqCst)
|| claims.session_id.as_deref() != Some("fixture-session")
{
return Err(JwtError::Revoked);
}
Ok(self.record(claims.user_id).await?.into())
}
async fn user(&self, id: i64) -> Result<UserRecord, JwtError> {
self.record(id).await
}
async fn session(
&self,
id: i64,
version: i32,
session_id: &str,
) -> Result<AccessRecord, JwtError> {
if self.revoked.load(Ordering::SeqCst) || session_id != "fixture-session" {
return Err(JwtError::Revoked);
}
let record = self.record(id).await?;
if record.security_version != version {
return Err(JwtError::Revoked);
}
Ok(record.into())
}
async fn update(&self, id: i64, _: serde_json::Value) -> Result<UserRecord, JwtError> {
self.record(id).await
}
}
impl Fixture {
async fn record(&self, id: i64) -> Result<UserRecord, JwtError> {
Ok(UserRecord {
id,
username: "member".into(),
lowername: "member".into(),
displayname: None,
birth_date: None,
email: "member@example.test".into(),
public_show_email: false,
email_verified_at: None,
allow_use: !self.disabled.load(Ordering::SeqCst),
forbid_login: false,
forbid_login_reason: None,
endtime_forbid_login_at: None,
is_admin: false,
bot: false,
system: false,
created_at: Utc::now(),
updated_at: Utc::now(),
security_version: self.version.load(Ordering::SeqCst),
version: 1,
deleted_at: None,
})
}
}
#[tokio::test]
async fn signed_tokens_obey_current_account_and_session_state() {
let key = base64::engine::general_purpose::STANDARD.encode([7_u8; 32]);
let config = config::identity::jwt::JwtConfig {
issuer: "fixture".into(),
audience: Some("gitdata.ai".into()),
leeway_secs: 0,
key_id: "fixture".into(),
private_key: Some(key.clone()),
jwks_url: None,
jwks_sha256: None,
retired: vec![],
authority_grpc_url: None,
profile_url: None,
};
crate::JwtState::load(&config).await.unwrap();
let authority = Arc::new(Fixture {
version: AtomicI32::new(1),
disabled: AtomicBool::new(false),
revoked: AtomicBool::new(false),
});
install(authority.clone()).unwrap();
let now = crate::now_secs();
let claims = AppClaims {
is_admin: true,
issuer: "fixture".into(),
app_id: "api".into(),
user_id: 1,
audience: "gitdata.ai".into(),
token_id: "fixture-token".into(),
issued_at: now,
expires_at: now + 600,
security_version: 1,
session_id: Some("fixture-session".into()),
mfa_verified: false,
scope: None,
username: None,
token_type: "access".into(),
};
let token = crate::sign(&crate::signing_key(&key).unwrap(), "fixture", &claims).unwrap();
assert!(!authenticate(&token).await.unwrap().is_admin);
assert!(session(1, 1, "fixture-session").await.is_ok());
assert!(session(1, 2, "fixture-session").await.is_err());
authority.version.store(2, Ordering::SeqCst);
assert!(authenticate(&token).await.is_err());
authority.version.store(1, Ordering::SeqCst);
authority.disabled.store(true, Ordering::SeqCst);
assert!(authenticate(&token).await.is_err());
authority.disabled.store(false, Ordering::SeqCst);
authority.revoked.store(true, Ordering::SeqCst);
assert!(authenticate(&token).await.is_err());
assert!(session(1, 1, "fixture-session").await.is_err());
assert!(crate::service::from_metadata(&tonic::metadata::MetadataMap::new()).is_err());
}
}