538d47b
use crate::models::users::users;
use crate::protos::auth::v1::identity_service_server::{IdentityService, IdentityServiceServer};
use crate::protos::auth::v1::*;
use chrono::Utc;
use rand::RngExt as _;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use tonic::{Request, Response, Status};
use crate::AuthSvc;
use crate::errors::{AuthError, AuthResult};
use crate::internal::{context, error, store};
#[derive(Clone)]
pub struct IdentityApi {
svc: AuthSvc,
}
impl IdentityApi {
pub fn new(svc: AuthSvc) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
IdentityServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
IdentityServiceServer::with_interceptor(self, gate)
}
}
fn random_bot_id() -> i64 {
let mut rng = rand::rng();
let digits = rng.random_range(5u32..=10);
let upper = 10i64.pow(digits);
rng.random_range((upper / 10)..upper)
}
fn bot_public(user: &users::Model) -> crate::protos::auth::common::v1::UserPublic {
crate::internal::store::user_public(user, String::new(), None)
}
impl IdentityApi {
async fn change_password_impl(
&self,
request: ChangePasswordRequest,
request_id: Option<&str>,
) -> AuthResult<ChangePasswordResponse> {
let db = self.svc.db();
let user = store::load_active_user(db, request.user_id).await?;
let stored = user.password_hash.clone();
if stored.is_empty() {
// 没有存量密码就不能「凭旧密码」修改：否则持 rpc_key 的一方能直接设置密码。
return Err(AuthError::InvalidPassword);
}
if !stored.is_empty() {
let check = crate::share::password::verify(
request.old_password_hash.clone(),
stored,
request_id,
)
.await?;
if check == crate::share::password::PasswordCheck::Mismatch {
return Err(AuthError::InvalidPassword);
}
}
crate::share::password::validate_new_password(
&request.new_password_hash,
&self.svc.auth().password_policy,
)?;
let new_hash =
crate::share::password::hash(request.new_password_hash.clone(), request_id).await?;
let now = store::now();
users::Entity::update_many()
.col_expr(users::Column::PasswordHash, Expr::value(new_hash))
.col_expr(users::Column::PasswordUpdatedAt, Expr::value(Some(now)))
.col_expr(users::Column::PasswordMustChange, Expr::value(false))
.col_expr(
users::Column::SecurityVersion,
Expr::col(users::Column::SecurityVersion).add(1),
)
.col_expr(users::Column::UpdatedAt, Expr::value(now))
.filter(users::Column::Id.eq(user.id))
.exec(db.write())
.await?;
let keep = match request.keep_session_id.trim() {
"" => None,
value => uuid::Uuid::parse_str(value).ok(),
};
let mut update = crate::models::users::user_sessions::Entity::update_many()
.col_expr(
crate::models::users::user_sessions::Column::RevokedAt,
Expr::value(Some(now)),
)
.col_expr(
crate::models::users::user_sessions::Column::Version,
Expr::col(crate::models::users::user_sessions::Column::Version).add(1),
)
.col_expr(
crate::models::users::user_sessions::Column::UpdatedAt,
Expr::value(now),
)
.filter(crate::models::users::user_sessions::Column::UserId.eq(user.id))
.filter(crate::models::users::user_sessions::Column::RevokedAt.is_null());
if let Some(keep) = keep {
update = update.filter(crate::models::users::user_sessions::Column::Id.ne(keep));
}
let revoked = update.exec(db.write()).await?;
if let Err(err) = store::drop_user_session_states(self.svc.cache(), user.id, keep).await {
tracing::warn!(%err, user_id = user.id, "failed to drop revoked sessions");
}
crate::me::service::invalidate_me_cache(self.svc.cache(), user.id).await;
store::audit(
db,
user.id,
Some(user.id),
crate::models::users::user_events::UserEventType::PasswordChange,
Some(serde_json::json!({ "action": "password_changed" })),
request_id,
match request.client_ip.trim() {
"" => None,
value => Some(value.to_owned()),
},
None,
)
.await?;
let security_version = store::load_user(db, user.id).await?.security_version;
if let Some(keep) = keep {
if let Err(err) = store::refresh_session_version(
self.svc.cache(),
user.id,
keep,
security_version,
self.svc.session_max_age_secs(),
)
.await
{
tracing::warn!(%err, user_id = user.id, "failed to refresh the kept session version");
}
}
Ok(ChangePasswordResponse {
changed: true,
security_version,
revoked_sessions: revoked.rows_affected as u32,
})
}
}
#[tonic::async_trait]
impl IdentityService for IdentityApi {
async fn authenticate(
&self,
_request: Request<AuthenticateRequest>,
) -> Result<Response<AuthenticateResponse>, Status> {
Err(Status::unimplemented("authenticate is not available"))
}
async fn verify_credential(
&self,
_request: Request<VerifyCredentialRequest>,
) -> Result<Response<VerifyCredentialResponse>, Status> {
Err(Status::unimplemented("verify_credential is not available"))
}
async fn get_user_public(
&self,
request: Request<GetUserPublicRequest>,
) -> Result<Response<GetUserPublicResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
let result = self
.get_user_public_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetUserPublic", &result);
result
}
async fn batch_get_user_public(
&self,
request: Request<BatchGetUserPublicRequest>,
) -> Result<Response<BatchGetUserPublicResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
let result = self
.batch_get_user_public_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("BatchGetUserPublic", &result);
result
}
async fn get_user_status(
&self,
request: Request<GetUserStatusRequest>,
) -> Result<Response<GetUserStatusResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
let result = self
.get_user_status_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetUserStatus", &result);
result
}
async fn ensure_bot_user(
&self,
request: Request<EnsureBotUserRequest>,
) -> Result<Response<EnsureBotUserResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
let result = self
.ensure_bot_user_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("EnsureBotUser", &result);
result
}
async fn change_password(
&self,
request: Request<ChangePasswordRequest>,
) -> Result<Response<ChangePasswordResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.IdentityService");
let result = self
.change_password_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ChangePassword", &result);
result
}
}
impl IdentityApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_user_public_inner(
&self,
request_id: Option<&str>,
payload: GetUserPublicRequest,
) -> Result<Response<GetUserPublicResponse>, Status> {
let reference = payload
.user
.as_ref()
.ok_or_else(|| error::status(&AuthError::MalformedRequest, request_id))?;
let user = store::find_user(self.svc.db(), reference)
.await
.map_err(|error_value| error::status(&error_value, request_id))?
.ok_or_else(|| error::status(&AuthError::UserNotFound, request_id))?;
let email = user.email.clone();
Ok(Response::new(GetUserPublicResponse {
user: Some(store::user_public(&user, email, None)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn batch_get_user_public_inner(
&self,
request_id: Option<&str>,
payload: BatchGetUserPublicRequest,
) -> Result<Response<BatchGetUserPublicResponse>, Status> {
let ids = payload.user_ids;
if ids.is_empty() {
return Ok(Response::new(BatchGetUserPublicResponse {
users: Vec::new(),
}));
}
if ids.len() > store::MAX_BATCH_USER_IDS {
return Err(error::invalid_argument(
format!(
"at most {} user ids can be fetched at once",
store::MAX_BATCH_USER_IDS
),
request_id,
));
}
let mut unique: Vec<i64> = Vec::with_capacity(ids.len());
for id in ids {
if !unique.contains(&id) {
unique.push(id);
}
}
let rows = users::Entity::find()
.filter(users::Column::Id.is_in(unique.iter().copied()))
.all(self.svc.db().read())
.await
.map_err(|error_value| error::status(&AuthError::from(error_value), request_id))?;
let mut users = Vec::with_capacity(unique.len());
for user in &rows {
users.push(store::user_public(user, user.email.clone(), None));
}
Ok(Response::new(BatchGetUserPublicResponse { users }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_user_status_inner(
&self,
request_id: Option<&str>,
payload: GetUserStatusRequest,
) -> Result<Response<GetUserStatusResponse>, Status> {
let reference = payload
.user
.as_ref()
.ok_or_else(|| error::status(&AuthError::MalformedRequest, request_id))?;
let user = store::find_user(self.svc.db(), reference)
.await
.map_err(|error_value| error::status(&error_value, request_id))?;
let Some(user) = user else {
return Ok(Response::new(GetUserStatusResponse {
status: Some(store::empty_status()),
}));
};
let mfa = store::active_mfa_count(self.svc.db(), user.id)
.await
.map_err(|error_value| error::status(&error_value, request_id))?;
Ok(Response::new(GetUserStatusResponse {
status: Some(store::user_status_of(&user, mfa)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn ensure_bot_user_inner(
&self,
request_id: Option<&str>,
payload: EnsureBotUserRequest,
) -> Result<Response<EnsureBotUserResponse>, Status> {
let username = payload.username.trim();
let lower = username.to_lowercase();
if lower.is_empty()
|| lower.len() > 32
|| !lower
.chars()
.all(|value| value.is_ascii_alphanumeric() || value == '-' || value == '_')
{
return Err(error::status(&AuthError::InvalidUsername, request_id));
}
let db = self.svc.db().write();
if let Some(existing) = users::Entity::find()
.filter(users::Column::Lowername.eq(&lower))
.one(db)
.await
.map_err(|value| error::status(&AuthError::from(value), request_id))?
{
if !existing.bot {
return Err(error::status(&AuthError::UsernameTaken, request_id));
}
return Ok(Response::new(EnsureBotUserResponse {
user: Some(bot_public(&existing)),
created: false,
}));
}
let displayname = if payload.displayname.trim().is_empty() {
username.to_string()
} else {
payload.displayname.trim().to_string()
};
let now = Utc::now();
let password_hash = crate::share::password::hash(format!("bot-{lower}-{now}"), request_id)
.await
.map_err(|value| error::status(&value, request_id))?;
let inserted = users::ActiveModel {
id: Set(random_bot_id()),
username: Set(username.to_string()),
lowername: Set(lower.clone()),
displayname: Set(Some(displayname)),
email: Set(format!("{lower}@bots.invalid")),
email_verified_at: Set(Some(now)),
password_hash: Set(password_hash),
password_updated_at: Set(Some(now)),
allow_use: Set(true),
forbid_login: Set(true),
forbid_login_reason: Set(Some("bot account".to_string())),
bot: Set(true),
system: Set(payload.system),
created_at: Set(now),
..Default::default()
}
.insert(db)
.await;
match inserted {
Ok(user) => Ok(Response::new(EnsureBotUserResponse {
user: Some(bot_public(&user)),
created: true,
})),
Err(failure) => {
// 只有唯一约束冲突才能走「已存在」分支；其余数据库错误必须原样上抛。
if !matches!(
failure.sql_err(),
Some(sea_orm::SqlErr::UniqueConstraintViolation(_))
) {
return Err(error::status(&AuthError::from(failure), request_id));
}
let existing = users::Entity::find()
.filter(users::Column::Lowername.eq(&lower))
.one(db)
.await
.map_err(|value| error::status(&AuthError::from(value), request_id))?
.ok_or_else(|| error::status(&AuthError::UsernameTaken, request_id))?;
if !existing.bot {
return Err(error::status(&AuthError::UsernameTaken, request_id));
}
Ok(Response::new(EnsureBotUserResponse {
user: Some(bot_public(&existing)),
created: false,
}))
}
}
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn change_password_inner(
&self,
request_id: Option<&str>,
payload: ChangePasswordRequest,
) -> Result<Response<ChangePasswordResponse>, Status> {
self.change_password_impl(payload, request_id)
.await
.map(Response::new)
.map_err(|failure| error::status(&failure, request_id))
}
}