ef0b83a
use crate::models::users::users;
use crate::protos::auth::common::v1::{Code, PageResponse};
use crate::protos::auth::internal::v1::authority_service_server::{
AuthorityService, AuthorityServiceServer,
};
use crate::protos::auth::internal::v1::{
AuthorityAccess, AuthorityUser, AuthorityUserPatch, BatchGetUserRequest, BatchGetUserResponse,
CancelAccountDeletionRequest, CancelAccountDeletionResponse, DeleteAccountRequest,
DeleteAccountResponse, GetSessionRequest, GetSessionResponse, GetUserRequest, GetUserResponse,
ListUsersRequest, ListUsersResponse, PurgeAccountRequest, PurgeAccountResponse,
UpdateUserRequest, UpdateUserResponse, ValidateRequest, ValidateResponse,
};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use tonic::{Request, Response, Status};
use crate::AuthService;
use crate::api::{context, convert, error, store};
use crate::errors::AuthError;
pub struct AuthorityApi {
svc: AuthService,
}
impl AuthorityApi {
pub fn new(svc: AuthService) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
AuthorityServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
AuthorityServiceServer::with_interceptor(self, gate)
}
}
fn status(error_value: &AuthError, request_id: Option<&str>) -> Status {
let code = match error_value {
AuthError::Unauthenticated | AuthError::SecurityVersionStale { .. } => {
Code::Unauthenticated
}
AuthError::Forbidden => Code::Forbidden,
AuthError::UserNotFound => Code::UserNotFound,
AuthError::SessionNotFound => Code::SessionNotFound,
AuthError::MalformedRequest => Code::MalformedRequest,
AuthError::VersionMismatch { .. } => Code::Conflict,
_ => Code::Internal,
};
error::status_of(code, error_value.message(), request_id)
}
fn user_status(error_value: &AuthError, request_id: Option<&str>) -> Status {
match error_value {
AuthError::UserNotFound => {
error::status_of(Code::UserNotFound, "user not found", request_id)
}
other => status(other, request_id),
}
}
fn access(user: &users::Model, mfa_verified: bool) -> AuthorityAccess {
AuthorityAccess {
id: user.id,
username: user.username.clone(),
allow_use: user.allow_use,
forbid_login: user.forbid_login,
endtime_forbid_login_at: convert::datetime_opt(user.endtime_forbid_login_at),
is_admin: user.is_admin,
security_version: user.security_version,
deleted_at: convert::datetime_opt(user.deleted_at),
mfa_verified,
}
}
fn user(record: &users::Model) -> AuthorityUser {
AuthorityUser {
id: record.id,
username: record.username.clone(),
lowername: record.lowername.clone(),
displayname: record.displayname.clone(),
birth_date: record.birth_date.and_then(convert::date_timestamp),
email: record.email.clone(),
public_show_email: record.public_show_email,
email_verified_at: convert::datetime_opt(record.email_verified_at),
allow_use: record.allow_use,
forbid_login: record.forbid_login,
forbid_login_reason: record.forbid_login_reason.clone(),
endtime_forbid_login_at: convert::datetime_opt(record.endtime_forbid_login_at),
is_admin: record.is_admin,
bot: record.bot,
system: record.system,
created_at: Some(convert::timestamp_from_datetime(record.created_at)),
updated_at: Some(convert::timestamp_from_datetime(record.updated_at)),
security_version: record.security_version,
version: record.version,
deleted_at: convert::datetime_opt(record.deleted_at),
}
}
fn patch_value(patch: &AuthorityUserPatch) -> Result<serde_json::Value, String> {
if patch.displayname.is_some() && patch.clear_displayname {
return Err("patch carries both displayname and clear_displayname".into());
}
if let Some(username) = patch.username.as_deref()
&& username.trim().is_empty()
{
return Err("patch carries an empty username".into());
}
if patch.username.is_none()
&& patch.displayname.is_none()
&& !patch.clear_displayname
&& patch.public_show_email.is_none()
{
return Err("patch carries no updatable field".into());
}
let mut value = serde_json::json!({ "version": i64::from(patch.version) });
if let Some(username) = patch.username.as_deref() {
value["username"] = serde_json::Value::String(username.to_owned());
}
if patch.clear_displayname {
value["displayname"] = serde_json::Value::Null;
} else if let Some(displayname) = patch.displayname.as_deref() {
value["displayname"] = serde_json::Value::String(displayname.to_owned());
}
if let Some(show) = patch.public_show_email {
value["public_show_email"] = serde_json::Value::Bool(show);
}
Ok(value)
}
#[tonic::async_trait]
impl AuthorityService for AuthorityApi {
async fn redeem_socket_ticket(
&self,
request: Request<crate::protos::auth::internal::v1::RedeemSocketTicketRequest>,
) -> Result<Response<crate::protos::auth::internal::v1::RedeemSocketTicketResponse>, Status>
{
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.redeem_socket_ticket_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RedeemSocketTicket", &result);
result
}
async fn validate(
&self,
request: Request<ValidateRequest>,
) -> Result<Response<ValidateResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.validate_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("Validate", &result);
result
}
async fn get_user(
&self,
request: Request<GetUserRequest>,
) -> Result<Response<GetUserResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.get_user_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetUser", &result);
result
}
async fn batch_get_user(
&self,
request: Request<BatchGetUserRequest>,
) -> Result<Response<BatchGetUserResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.batch_get_user_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("BatchGetUser", &result);
result
}
async fn get_session(
&self,
request: Request<GetSessionRequest>,
) -> Result<Response<GetSessionResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.get_session_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetSession", &result);
result
}
async fn update_user(
&self,
request: Request<UpdateUserRequest>,
) -> Result<Response<UpdateUserResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.update_user_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateUser", &result);
result
}
async fn list_users(
&self,
request: Request<ListUsersRequest>,
) -> Result<Response<ListUsersResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.list_users_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListUsers", &result);
result
}
async fn delete_account(
&self,
request: Request<DeleteAccountRequest>,
) -> Result<Response<DeleteAccountResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.delete_account_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("DeleteAccount", &result);
result
}
async fn cancel_account_deletion(
&self,
request: Request<CancelAccountDeletionRequest>,
) -> Result<Response<CancelAccountDeletionResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.cancel_account_deletion_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CancelAccountDeletion", &result);
result
}
async fn purge_account(
&self,
request: Request<PurgeAccountRequest>,
) -> Result<Response<PurgeAccountResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.AuthorityService");
let result = self
.purge_account_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("PurgeAccount", &result);
result
}
}
impl AuthorityApi {
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
async fn redeem_socket_ticket_inner(
&self,
_request_id: Option<&str>,
_payload: crate::protos::auth::internal::v1::RedeemSocketTicketRequest,
) -> Result<Response<crate::protos::auth::internal::v1::RedeemSocketTicketResponse>, Status>
{
Err(Status::unimplemented("socket tickets are not available"))
}
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
async fn validate_inner(
&self,
_request_id: Option<&str>,
_payload: ValidateRequest,
) -> Result<Response<ValidateResponse>, Status> {
Err(Status::unimplemented(
"access token validation is not available",
))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_user_inner(
&self,
request_id: Option<&str>,
payload: GetUserRequest,
) -> Result<Response<GetUserResponse>, Status> {
let user_id = payload.user_id;
if user_id <= 0 {
return Err(error::invalid_argument(
"user_id must be positive",
request_id,
));
}
let record = self
.svc
.user(user_id)
.await
.map_err(|error_value| user_status(&error_value, request_id))?;
Ok(Response::new(GetUserResponse {
user: Some(user(&record)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn batch_get_user_inner(
&self,
request_id: Option<&str>,
payload: BatchGetUserRequest,
) -> Result<Response<BatchGetUserResponse>, Status> {
let user_ids = payload.user_ids;
if user_ids.is_empty() {
return Ok(Response::new(BatchGetUserResponse { users: Vec::new() }));
}
if user_ids.len() > crate::api::store::MAX_BATCH_USER_IDS {
return Err(error::invalid_argument(
"too many user ids in one batch",
request_id,
));
}
let records = self
.svc
.users(&user_ids)
.await
.map_err(|error_value| user_status(&error_value, request_id))?;
Ok(Response::new(BatchGetUserResponse {
users: records.iter().map(user).collect(),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_session_inner(
&self,
request_id: Option<&str>,
payload: GetSessionRequest,
) -> Result<Response<GetSessionResponse>, Status> {
if payload.user_id <= 0 {
return Err(error::invalid_argument(
"user_id must be positive",
request_id,
));
}
if payload.version <= 0 {
return Err(error::invalid_argument(
"version must be positive",
request_id,
));
}
if payload.session_id.is_empty() {
return Err(error::invalid_argument(
"session_id is required",
request_id,
));
}
let (user, mfa_verified) = self
.svc
.check_session(
payload.user_id,
payload.version,
&payload.session_id,
request_id,
)
.await
.map_err(|error_value| status(&error_value, request_id))?;
tracing::debug!(
request_id,
user_id = payload.user_id,
"authority revalidated a session"
);
Ok(Response::new(GetSessionResponse {
access: Some(access(&user, mfa_verified)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_user_inner(
&self,
request_id: Option<&str>,
payload: UpdateUserRequest,
) -> Result<Response<UpdateUserResponse>, Status> {
if payload.user_id <= 0 {
return Err(error::invalid_argument(
"user_id must be positive",
request_id,
));
}
let patch = payload
.patch
.ok_or_else(|| error::invalid_argument("patch is required", request_id))?;
if patch.version <= 0 {
return Err(error::invalid_argument(
"version must be positive",
request_id,
));
}
let value = patch_value(&patch).map_err(|message| {
error::invalid_argument(format!("invalid patch: {message}"), request_id)
})?;
let record =
self.svc.update(payload.user_id, value).await.map_err(
|error_value| match error_value {
AuthError::VersionMismatch { .. } => {
error::status_of(Code::Conflict, "account version conflict", request_id)
}
other => status(&other, request_id),
},
)?;
tracing::info!(
request_id,
user_id = record.id,
version = record.version,
"authority updated an account"
);
Ok(Response::new(UpdateUserResponse {
user: Some(user(&record)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_users_inner(
&self,
request_id: Option<&str>,
payload: ListUsersRequest,
) -> Result<Response<ListUsersResponse>, Status> {
let filter = payload.filter.unwrap_or_default();
if filter.user_ids.len() > crate::api::store::MAX_BATCH_USER_IDS {
return Err(error::invalid_argument(
"too many user ids in one filter",
request_id,
));
}
let page = payload.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
let query = order_by(
users::Entity::find().filter(filter_condition(&filter)),
payload.order,
);
let rows = query
.limit(page_size as u64)
.offset(offset)
.all(self.svc.db().read())
.await
.map_err(|value| error::status(&crate::errors::AuthError::from(value), request_id))?;
let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
Ok(Response::new(ListUsersResponse {
users: rows.iter().map(user).collect(),
page: Some(PageResponse { next_page_token }),
}))
}
async fn delete_account_inner(
&self,
request_id: Option<&str>,
payload: DeleteAccountRequest,
) -> Result<Response<DeleteAccountResponse>, Status> {
if payload.user_id <= 0 {
return Err(error::invalid_argument(
"user_id must be positive",
request_id,
));
}
self.svc
.delete_account(
payload.user_id,
&payload.reason,
request_id.unwrap_or_default(),
)
.await
.map_err(|failure| error::status(&failure, request_id))?;
Ok(Response::new(DeleteAccountResponse {}))
}
async fn cancel_account_deletion_inner(
&self,
request_id: Option<&str>,
payload: CancelAccountDeletionRequest,
) -> Result<Response<CancelAccountDeletionResponse>, Status> {
if payload.user_id <= 0 {
return Err(error::invalid_argument(
"user_id must be positive",
request_id,
));
}
self.svc
.cancel_account_deletion(payload.user_id, request_id.unwrap_or_default())
.await
.map_err(|failure| error::status(&failure, request_id))?;
Ok(Response::new(CancelAccountDeletionResponse {}))
}
async fn purge_account_inner(
&self,
request_id: Option<&str>,
payload: PurgeAccountRequest,
) -> Result<Response<PurgeAccountResponse>, Status> {
if payload.user_id <= 0 {
return Err(error::invalid_argument(
"user_id must be positive",
request_id,
));
}
self.svc
.purge_account(payload.user_id, request_id.unwrap_or_default())
.await
.map_err(|failure| error::status(&failure, request_id))?;
Ok(Response::new(PurgeAccountResponse {}))
}
}
/// 把 proto 里的筛选条件翻成 SeaORM 条件树。未设置的字段一律跳过，
/// 因此空 filter 等价于「全量列表」。
fn filter_condition(
filter: &crate::protos::auth::internal::v1::ListUsersFilter,
) -> sea_orm::Condition {
use crate::protos::auth::internal::v1::{UserKind, UserState};
use sea_orm::Condition;
let mut any = Condition::all();
// 子串匹配统一走 LIKE：username / lowername / displayname / email 四列任一命中即可。
// 大小写不敏感靠 lowername 与 email 落库即小写来保证，displayname 则补一趟小写比较。
let needle = filter.query.trim();
if !needle.is_empty() {
let lowered = needle.to_lowercase();
any = any.add(
Condition::any()
.add(users::Column::Username.contains(needle))
.add(users::Column::Lowername.contains(&lowered))
.add(users::Column::Displayname.contains(needle))
.add(users::Column::Email.contains(&lowered)),
);
}
if !filter.user_ids.is_empty() {
any = any.add(users::Column::Id.is_in(filter.user_ids.iter().copied()));
}
// 形态筛选：system 优先于 bot，与 UserKind 的互斥语义一致。
match UserKind::try_from(filter.kind).ok() {
Some(UserKind::Human) => {
any = any
.add(users::Column::Bot.eq(false))
.add(users::Column::System.eq(false));
}
Some(UserKind::Bot) => {
any = any
.add(users::Column::Bot.eq(true))
.add(users::Column::System.eq(false));
}
Some(UserKind::System) => {
any = any.add(users::Column::System.eq(true));
}
Some(UserKind::Unspecified) | None => {}
}
// 状态筛选：优先级 DELETED > LOGIN_FORBIDDEN > UNAVAILABLE > ACTIVE，
// 与 UserState 的注释一致；DB 侧用同一顺序表达，避免一条记录落两档。
match UserState::try_from(filter.state).ok() {
Some(UserState::Active) => {
any = any
.add(users::Column::DeletedAt.is_null())
.add(users::Column::AllowUse.eq(true))
.add(users::Column::ForbidLogin.eq(false));
}
Some(UserState::Unavailable) => {
any = any
.add(users::Column::DeletedAt.is_null())
.add(users::Column::AllowUse.eq(false));
}
Some(UserState::LoginForbidden) => {
any = any
.add(users::Column::DeletedAt.is_null())
.add(users::Column::ForbidLogin.eq(true));
}
Some(UserState::Deleted) => {
any = any.add(users::Column::DeletedAt.is_not_null());
}
Some(UserState::Unspecified) | None => {}
}
if let Some(is_admin) = filter.is_admin {
any = any.add(users::Column::IsAdmin.eq(is_admin));
}
if let Some(verified) = filter.email_verified {
any = if verified {
any.add(users::Column::EmailVerifiedAt.is_not_null())
} else {
any.add(users::Column::EmailVerifiedAt.is_null())
};
}
any
}
fn order_by(
mut query: sea_orm::Select<users::Entity>,
order: i32,
) -> sea_orm::Select<users::Entity> {
use crate::protos::auth::internal::v1::UserOrder;
match UserOrder::try_from(order).ok() {
Some(UserOrder::CreatedAtAsc) => query = query.order_by_asc(users::Column::CreatedAt),
Some(UserOrder::LastLoginDesc) => {
query = query
.order_by_desc(users::Column::LastLoginAt)
.order_by_desc(users::Column::Id);
}
Some(UserOrder::IdAsc) => query = query.order_by_asc(users::Column::Id),
// 默认与 UNSPECIFIED 都走创建时间倒序；补 id 做稳定排序，翻页不跳行。
Some(UserOrder::CreatedAtDesc) | Some(UserOrder::Unspecified) | None => {
query = query
.order_by_desc(users::Column::CreatedAt)
.order_by_desc(users::Column::Id);
}
}
query
}