538d47b
use crate::models::users::user_events::UserEventType;
use crate::models::users::{user_sessions, users};
use crate::protos::auth::v1::session_service_server::{SessionService, SessionServiceServer};
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::AuthSvc;
use crate::errors::AuthError;
use crate::internal::{context, convert, error, store};
pub const MAX_LIST_PAGE: u64 = 100;
pub const SESSION_LIST_MAX: usize = 200;
#[derive(Clone)]
pub struct SessionApi {
svc: AuthSvc,
}
impl SessionApi {
pub fn new(svc: AuthSvc) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
SessionServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
SessionServiceServer::with_interceptor(self, gate)
}
}
#[tonic::async_trait]
impl SessionService for SessionApi {
async fn list_user_sessions(
&self,
request: Request<ListUserSessionsRequest>,
) -> Result<Response<ListUserSessionsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
let result = self
.list_user_sessions_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListUserSessions", &result);
result
}
async fn revoke_session(
&self,
request: Request<RevokeSessionRequest>,
) -> Result<Response<RevokeSessionResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
let result = self
.revoke_session_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RevokeSession", &result);
result
}
async fn revoke_user_sessions(
&self,
request: Request<RevokeUserSessionsRequest>,
) -> Result<Response<RevokeUserSessionsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
let result = self
.revoke_user_sessions_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RevokeUserSessions", &result);
result
}
async fn bump_security_version(
&self,
request: Request<BumpSecurityVersionRequest>,
) -> Result<Response<BumpSecurityVersionResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.SessionService");
let result = self
.bump_security_version_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("BumpSecurityVersion", &result);
result
}
}
impl SessionApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_user_sessions_inner(
&self,
request_id: Option<&str>,
request: ListUserSessionsRequest,
) -> Result<Response<ListUserSessionsResponse>, 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) as u64;
let mut limit = if page_size < MAX_LIST_PAGE {
page_size
} else {
MAX_LIST_PAGE
};
let remaining = (SESSION_LIST_MAX as u64).saturating_sub(offset);
if limit > remaining {
limit = remaining;
}
let mut query =
user_sessions::Entity::find().filter(user_sessions::Column::UserId.eq(request.user_id));
if !request.include_revoked {
query = query.filter(user_sessions::Column::RevokedAt.is_null());
}
if !request.include_expired {
query = query.filter(user_sessions::Column::ExpiresAt.gt(store::now()));
}
let rows = query
.order_by_desc(user_sessions::Column::LastSeenAt)
.limit(limit)
.offset(offset)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let next_page_token = if limit > 0 && rows.len() as u64 == limit {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
let sessions = rows.iter().map(convert::session_info).collect();
Ok(Response::new(ListUserSessionsResponse {
sessions,
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn revoke_session_inner(
&self,
request_id: Option<&str>,
request: RevokeSessionRequest,
) -> Result<Response<RevokeSessionResponse>, Status> {
let session_id = Uuid::parse_str(request.session_id.trim())
.map_err(|_| error::invalid_argument("session_id is not a valid uuid", request_id))?;
let user = store::load_user(self.svc.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let security_version = user.security_version;
let row = user_sessions::Entity::find_by_id(session_id)
.filter(user_sessions::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(RevokeSessionResponse {
revoked: false,
security_version,
}));
};
if row.revoked_at.is_some() {
return Ok(Response::new(RevokeSessionResponse {
revoked: true,
security_version,
}));
}
let revoked_at = store::now();
// 条件更新 + 行数校验：并发双撤时只有一方生效。
let updated = user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::RevokedAt, Expr::value(revoked_at))
.col_expr(
user_sessions::Column::Version,
Expr::col(user_sessions::Column::Version).add(1),
)
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(revoked_at))
.filter(user_sessions::Column::Id.eq(session_id))
.filter(user_sessions::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(RevokeSessionResponse {
revoked: true,
security_version,
}));
}
if let Err(err) =
store::drop_session_by_id(self.svc.cache(), request.user_id, session_id).await
{
tracing::warn!(%err, %session_id, "failed to drop the revoked session state");
}
if let Err(err) = store::audit(
self.svc.db(),
request.user_id,
None,
UserEventType::Logout,
Some(serde_json::json!({
"reason": request.reason,
"session_id": session_id,
})),
request_id,
None,
None,
)
.await
{
// 会话已失效，审计失败不回退结果。
tracing::warn!(%err, %session_id, "failed to record session revocation");
}
Ok(Response::new(RevokeSessionResponse {
revoked: true,
security_version,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn revoke_user_sessions_inner(
&self,
request_id: Option<&str>,
request: RevokeUserSessionsRequest,
) -> Result<Response<RevokeUserSessionsResponse>, Status> {
let except = match request.except_session_id.trim() {
"" => None,
value => Some(Uuid::parse_str(value).map_err(|_| {
error::invalid_argument("except_session_id is not a valid uuid", request_id)
})?),
};
let now = store::now();
let mut update = user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::RevokedAt, Expr::value(now))
.col_expr(
user_sessions::Column::Version,
Expr::col(user_sessions::Column::Version).add(1),
)
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(now))
.filter(user_sessions::Column::UserId.eq(request.user_id))
.filter(user_sessions::Column::RevokedAt.is_null());
if let Some(except) = except {
update = update.filter(user_sessions::Column::Id.ne(except));
}
let affected = update
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if let Err(err) =
store::drop_user_session_states(self.svc.cache(), request.user_id, except).await
{
tracing::warn!(%err, user_id = request.user_id, "failed to drop revoked sessions");
}
let user = store::load_user(self.svc.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let mut security_version = user.security_version;
if request.bump_security_version {
users::Entity::update_many()
.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(request.user_id))
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
crate::me::service::invalidate_me_cache(self.svc.cache(), request.user_id).await;
security_version = store::load_user(self.svc.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?
.security_version;
}
Ok(Response::new(RevokeUserSessionsResponse {
revoked_count: affected.rows_affected as u32,
security_version,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn bump_security_version_inner(
&self,
request_id: Option<&str>,
request: BumpSecurityVersionRequest,
) -> Result<Response<BumpSecurityVersionResponse>, Status> {
store::load_user(self.svc.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let now = store::now();
users::Entity::update_many()
.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(request.user_id))
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let security_version = store::load_user(self.svc.db(), request.user_id)
.await
.map_err(|err| error::status(&err, request_id))?
.security_version;
let mut revoked_sessions = 0u32;
if request.revoke_sessions {
let revoked = user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::RevokedAt, Expr::value(now))
.col_expr(
user_sessions::Column::Version,
Expr::col(user_sessions::Column::Version).add(1),
)
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(now))
.filter(user_sessions::Column::UserId.eq(request.user_id))
.filter(user_sessions::Column::RevokedAt.is_null())
.exec(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
revoked_sessions = revoked.rows_affected as u32;
}
crate::me::service::invalidate_me_cache(self.svc.cache(), request.user_id).await;
store::audit(
self.svc.db(),
request.user_id,
None,
UserEventType::Logout,
Some(serde_json::json!({
"reason": request.reason,
"revoke_sessions": request.revoke_sessions,
})),
request_id,
None,
None,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(BumpSecurityVersionResponse {
security_version,
revoked_sessions,
bumped_at: Some(convert::now_timestamp()),
}))
}
}