b6f7f73
ZhenYi
refactor: reorganize configuration files and update module paths for improved structure
b6f7f73
use crate::models::users::user_access_tokens;
use crate::models::users::user_events::UserEventType;
use crate::protos::auth::common::v1::{AuthMethod, Code, PageResponse, TokenScope};
use crate::protos::auth::v1::access_key_service_server::{
AccessKeyService, AccessKeyServiceServer,
};
use crate::protos::auth::v1::*;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use tonic::{Request, Response, Status};
use uuid::Uuid;
use crate::server::AuthSvc;
use crate::server::errors::AuthError;
use crate::server::grpc::{context, convert, error, store};
use crate::server::http::access_token::types::TOKEN_HASH_KEY_ID;
#[derive(Clone)]
pub struct AccessKeyApi {
svc: AuthSvc,
}
impl AccessKeyApi {
pub fn new(svc: AuthSvc) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
AccessKeyServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().oauth.rpc_key.clone());
AccessKeyServiceServer::with_interceptor(self, gate)
}
}
fn invalid(reason: Code) -> VerifyAccessKeyResponse {
VerifyAccessKeyResponse {
valid: false,
reason: reason as i32,
..Default::default()
}
}
#[tonic::async_trait]
impl AccessKeyService for AccessKeyApi {
async fn verify_access_key(
&self,
request: Request<VerifyAccessKeyRequest>,
) -> Result<Response<VerifyAccessKeyResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AccessKeyService");
let result = self
.verify_access_key_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("VerifyAccessKey", &result);
result
}
async fn create_access_key(
&self,
request: Request<CreateAccessKeyRequest>,
) -> Result<Response<CreateAccessKeyResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AccessKeyService");
let result = self
.create_access_key_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CreateAccessKey", &result);
result
}
async fn rotate_access_key(
&self,
request: Request<RotateAccessKeyRequest>,
) -> Result<Response<RotateAccessKeyResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AccessKeyService");
let result = self
.rotate_access_key_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RotateAccessKey", &result);
result
}
async fn list_access_keys(
&self,
request: Request<ListAccessKeysRequest>,
) -> Result<Response<ListAccessKeysResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AccessKeyService");
let result = self
.list_access_keys_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListAccessKeys", &result);
result
}
async fn revoke_access_key(
&self,
request: Request<RevokeAccessKeyRequest>,
) -> Result<Response<RevokeAccessKeyResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AccessKeyService");
let result = self
.revoke_access_key_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RevokeAccessKey", &result);
result
}
}
impl AccessKeyApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn verify_access_key_inner(
&self,
request_id: Option<&str>,
request: VerifyAccessKeyRequest,
) -> Result<Response<VerifyAccessKeyResponse>, Status> {
let access_key = request.access_key.trim();
if access_key.is_empty() {
return Err(error::invalid_argument(
"access_key is required",
request_id,
));
}
let hash = store::sha256_digest(access_key);
// 主库读：副本延迟会让刚撤销的 key 在窗口内继续通过验验。
let record = user_access_tokens::Entity::find()
.filter(user_access_tokens::Column::TokenHash.eq(hash))
.one(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let Some(record) = record else {
return Ok(Response::new(invalid(Code::Unauthenticated)));
};
if record.hash_key_id != TOKEN_HASH_KEY_ID {
return Ok(Response::new(invalid(Code::Unauthenticated)));
}
if record.revoked_at.is_some() {
return Ok(Response::new(invalid(Code::TokenRevoked)));
}
if record
.expires_at
.is_some_and(|expires| expires <= store::now())
{
return Ok(Response::new(invalid(Code::TokenExpired)));
}
let user = match store::load_active_user(self.svc.db(), record.user_id).await {
Ok(user) => user,
Err(
AuthError::AccountUnavailable { .. }
| AuthError::LoginForbidden { .. }
| AuthError::UserNotFound,
) => return Ok(Response::new(invalid(Code::UserDisabled))),
Err(err) => return Err(error::status(&err, request_id)),
};
if record.security_version != user.security_version {
return Ok(Response::new(invalid(Code::SecurityVersionStale)));
}
let scopes = store::scope_codes(&store::scopes_from_names(&record.scope));
if !store::scopes_cover(&scopes, &request.required_scopes) {
return Ok(Response::new(invalid(Code::ScopeInsufficient)));
}
user_access_tokens::Entity::update_many()
.col_expr(
user_access_tokens::Column::LastUsedAt,
Expr::value(store::now()),
)
.filter(user_access_tokens::Column::Id.eq(record.id))
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let client_ip = if request.client_ip.trim().is_empty() {
None
} else {
Some(request.client_ip.clone())
};
store::audit(
self.svc.db(),
record.user_id,
None,
UserEventType::Login,
Some(serde_json::json!({
"method": "access_key",
"access_key_id": record.id.to_string(),
})),
request_id,
client_ip,
None,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let mut principal = store::principal(&user, AuthMethod::AccessKey);
principal.access_key_id = record.id.to_string();
principal.scopes = scopes;
principal.scopes_bounded = true;
Ok(Response::new(VerifyAccessKeyResponse {
valid: true,
reason: Code::Unspecified as i32,
principal: Some(principal),
key: Some(convert::access_key_info(&record)),
current_security_version: user.security_version,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn create_access_key_inner(
&self,
request_id: Option<&str>,
request: CreateAccessKeyRequest,
) -> Result<Response<CreateAccessKeyResponse>, Status> {
let scope: Vec<String> = request
.scopes
.iter()
.filter_map(|code| {
TokenScope::try_from(*code)
.ok()
.and_then(store::scope_name)
.map(str::to_owned)
})
.collect();
let created = self
.svc
.create_access_token_for_user(
request.user_id,
crate::server::http::access_token::types::AccessTokenCreateRequest {
name: request.name,
description: Some(request.description).filter(|text| !text.is_empty()),
scope,
expires_in_secs: (request.expires_in_secs > 0)
.then_some(request.expires_in_secs),
},
None,
Some(request.client_ip).filter(|ip| !ip.is_empty()),
request_id.unwrap_or_default(),
)
.await
.map_err(|err| error::status(&err, request_id))?;
let row = user_access_tokens::Entity::find_by_id(created.id)
.one(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
.ok_or_else(|| error::unavailable("access key vanished after insert", request_id))?;
Ok(Response::new(CreateAccessKeyResponse {
key: Some(convert::access_key_info(&row)),
token: created.token,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn rotate_access_key_inner(
&self,
request_id: Option<&str>,
request: RotateAccessKeyRequest,
) -> Result<Response<RotateAccessKeyResponse>, Status> {
let access_key_id = Uuid::parse_str(request.access_key_id.trim()).map_err(|_| {
error::invalid_argument("access_key_id is not a valid uuid", request_id)
})?;
let created = self
.svc
.rotate_access_token_for_user(
request.user_id,
access_key_id,
true,
non_empty(&request.client_ip),
None,
request_id.unwrap_or_default().to_owned(),
)
.await
.map_err(|failure| error::status(&failure, request_id))?;
let row = user_access_tokens::Entity::find_by_id(created.id)
.one(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
.ok_or_else(|| error::unavailable("access key vanished after rotate", request_id))?;
Ok(Response::new(RotateAccessKeyResponse {
key: Some(convert::access_key_info(&row)),
token: created.token,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_access_keys_inner(
&self,
request_id: Option<&str>,
request: ListAccessKeysRequest,
) -> Result<Response<ListAccessKeysResponse>, Status> {
let page = request.page.unwrap_or_default();
let offset = match store::parse_page_token(&page.page_token) {
Some(offset) => offset,
None => {
return Err(error::invalid_argument(
"page_token is malformed",
request_id,
));
}
};
let page_size = convert::page_size(page.page_size);
let mut query = user_access_tokens::Entity::find()
.filter(user_access_tokens::Column::UserId.eq(request.user_id));
if !request.include_revoked {
query = query.filter(user_access_tokens::Column::RevokedAt.is_null());
}
let rows = query
.order_by_desc(user_access_tokens::Column::CreatedAt)
.limit(page_size as u64)
.offset(offset)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), 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()
};
let keys = rows.iter().map(convert::access_key_info).collect();
Ok(Response::new(ListAccessKeysResponse {
keys,
page: Some(PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn revoke_access_key_inner(
&self,
request_id: Option<&str>,
request: RevokeAccessKeyRequest,
) -> Result<Response<RevokeAccessKeyResponse>, Status> {
let access_key_id = Uuid::parse_str(request.access_key_id.trim()).map_err(|_| {
error::invalid_argument("access_key_id is not a valid uuid", request_id)
})?;
let row = user_access_tokens::Entity::find_by_id(access_key_id)
.filter(user_access_tokens::Column::UserId.eq(request.user_id))
.one(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let Some(row) = row else {
return Ok(Response::new(RevokeAccessKeyResponse {
revoked: false,
already_revoked: false,
}));
};
if row.revoked_at.is_some() {
return Ok(Response::new(RevokeAccessKeyResponse {
revoked: false,
already_revoked: true,
}));
}
let revoked_at = store::now();
// 条件更新 + 行数校验：并发双撤时只有一方真正生效，避免审计重复与版本双增。
let updated = user_access_tokens::Entity::update_many()
.col_expr(
user_access_tokens::Column::RevokedAt,
Expr::value(revoked_at),
)
.col_expr(
user_access_tokens::Column::Version,
Expr::col(user_access_tokens::Column::Version).add(1),
)
.filter(user_access_tokens::Column::Id.eq(access_key_id))
.filter(user_access_tokens::Column::RevokedAt.is_null())
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if updated.rows_affected == 0 {
return Ok(Response::new(RevokeAccessKeyResponse {
revoked: false,
already_revoked: true,
}));
}
if let Err(err) = store::audit(
self.svc.db(),
request.user_id,
None,
UserEventType::AccessKeyRemoved,
Some(serde_json::json!({
"access_key_id": access_key_id.to_string(),
"reason": request.reason,
})),
request_id,
None,
None,
)
.await
{
// 撤销已提交，审计失败不能把结果退回给客户端（重试只会拿到 already_revoked）。
tracing::warn!(%err, access_key_id = %access_key_id, "failed to record access key revocation");
}
Ok(Response::new(RevokeAccessKeyResponse {
revoked: true,
already_revoked: false,
}))
}
}
fn non_empty(value: &str) -> Option<String> {
let trimmed = value.trim();
if trimmed.is_empty() {
None
} else {
Some(trimmed.to_owned())
}
}