9362a10
ZhenYi
refactor: remove OAuth-related modules and update references for identity service integration
9362a10
use std::time::Duration;
use crate::models::users::{service_token_revocations, user_access_tokens};
use crate::protos::auth::common::v1::{Code, TokenAudience, TokenScope, credential};
use crate::protos::auth::v1::issue_service_token_request::Constraint;
use crate::protos::auth::v1::token_service_server::{TokenService, TokenServiceServer};
use crate::protos::auth::v1::*;
use prost::Message;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use serde::{Deserialize, Serialize};
use tonic::{Request, Response, Status};
use crate::server::AuthSvc;
use crate::server::errors::AuthError;
use crate::server::grpc::{context, convert, error, service_token, store};
use crate::server::http::access_token::types::{ALLOWED_APP_SCOPES, AccessTokenAuthRequest};
use crate::server::share::crypto;
const REVOKED_PREFIX: &str = "auth:svc-token:revoked:";
const META_PREFIX: &str = "auth:svc-token:meta:";
const ISSUE_PREFIX: &str = "auth:svc-token:issue:";
const REVOKE_PREFIX: &str = "auth:svc-token:revoke:";
#[derive(Clone, Debug, Serialize, Deserialize)]
struct TokenMeta {
user_id: i64,
exp: i64,
aud: i32,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
struct CachedIssue {
service_token: String,
token_id: String,
key_id: String,
subject_user_id: i64,
security_version: i32,
issued_at: i64,
expires_at: i64,
}
impl From<&IssueServiceTokenResponse> for CachedIssue {
fn from(response: &IssueServiceTokenResponse) -> Self {
Self {
service_token: response.service_token.clone(),
token_id: response.token_id.clone(),
key_id: response.key_id.clone(),
subject_user_id: response.subject_user_id,
security_version: response.security_version,
issued_at: response
.issued_at
.map(|stamp| stamp.seconds)
.unwrap_or_default(),
expires_at: response
.expires_at
.map(|stamp| stamp.seconds)
.unwrap_or_default(),
}
}
}
impl From<CachedIssue> for IssueServiceTokenResponse {
fn from(cached: CachedIssue) -> Self {
Self {
service_token: cached.service_token,
token_id: cached.token_id,
key_id: cached.key_id,
subject_user_id: cached.subject_user_id,
security_version: cached.security_version,
issued_at: Some(convert::timestamp(cached.issued_at, 0)),
expires_at: Some(convert::timestamp(cached.expires_at, 0)),
}
}
}
fn inactive(reason: Code) -> IntrospectTokenResponse {
IntrospectTokenResponse {
active: false,
reason: reason as i32,
..Default::default()
}
}
fn scope_label(code: i32) -> String {
TokenScope::try_from(code)
.ok()
.and_then(store::scope_name)
.map(str::to_owned)
.unwrap_or_else(|| code.to_string())
}
fn issue_cache_key(
event_id: &str,
request: &IssueServiceTokenRequest,
audience: TokenAudience,
ttl: u32,
) -> String {
let constraint = match request.constraint.as_ref() {
Some(Constraint::Chat(grant)) => grant.encode_to_vec(),
Some(Constraint::Sfu(grant)) => grant.encode_to_vec(),
None => Vec::new(),
};
let canonical = format!(
"sub={};aud={};ttl={};scopes={:?};constraint={}",
request.subject_user_id,
audience as i32,
ttl,
request.scopes,
store::sha256_digest(&String::from_utf8_lossy(&constraint)),
);
format!(
"{ISSUE_PREFIX}{event_id}:{}",
store::sha256_digest(&canonical)
)
}
#[derive(Clone)]
pub struct TokenApi {
svc: AuthSvc,
}
impl TokenApi {
pub fn new(svc: AuthSvc) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
TokenServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
TokenServiceServer::with_interceptor(self, gate)
}
}
#[tonic::async_trait]
impl TokenService for TokenApi {
async fn introspect_token(
&self,
request: Request<IntrospectTokenRequest>,
) -> Result<Response<IntrospectTokenResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.TokenService");
let result = self
.introspect_token_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("IntrospectToken", &result);
result
}
async fn issue_service_token(
&self,
request: Request<IssueServiceTokenRequest>,
) -> Result<Response<IssueServiceTokenResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.TokenService");
let result = self
.issue_service_token_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("IssueServiceToken", &result);
result
}
async fn revoke_service_token(
&self,
request: Request<RevokeServiceTokenRequest>,
) -> Result<Response<RevokeServiceTokenResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.TokenService");
let result = self
.revoke_service_token_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RevokeServiceToken", &result);
result
}
}
impl TokenApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn introspect_token_inner(
&self,
request_id: Option<&str>,
request: IntrospectTokenRequest,
) -> Result<Response<IntrospectTokenResponse>, Status> {
let required_scopes = request.required_scopes;
let Some(credential) = request.credential else {
return Err(error::invalid_argument(
"credential is required",
request_id,
));
};
let Some(kind) = credential.kind else {
return Err(error::invalid_argument(
"credential is required",
request_id,
));
};
match kind {
credential::Kind::AccessToken(token) => {
// 统一走访问令牌服务的校验链，避免两套判定逻辑漂移。
let response = match self
.svc
.authenticate_access_token(
AccessTokenAuthRequest {
token: token.to_owned(),
},
request_id.unwrap_or_default().to_owned(),
)
.await
{
Ok(response) => response,
Err(err) => {
let reason = match err {
AuthError::InvalidAccessToken => Code::InvalidAccessToken,
AuthError::SecurityVersionStale { .. } => Code::SecurityVersionStale,
AuthError::UserNotFound
| AuthError::AccountUnavailable { .. }
| AuthError::LoginForbidden { .. }
| AuthError::TokenRevoked
| AuthError::TokenExpired => Code::Unauthenticated,
_ => Code::Unauthenticated,
};
return Ok(Response::new(inactive(reason)));
}
};
let scopes = store::scope_codes(&store::scopes_from_names(&response.scope));
if !store::scopes_cover(&scopes, &required_scopes) {
return Ok(Response::new(inactive(Code::ScopeInsufficient)));
}
let record = match user_access_tokens::Entity::find_by_id(response.token_id)
.one(self.svc.db().read())
.await
{
Ok(record) => record,
Err(err) => {
return Err(error::status(&AuthError::from(err), request_id));
}
};
let Some(record) = record else {
return Ok(Response::new(inactive(Code::InvalidAccessToken)));
};
let user = match store::load_user(self.svc.db(), record.user_id).await {
Ok(user) => user,
Err(_) => return Ok(Response::new(inactive(Code::Unauthenticated))),
};
let mut principal = store::principal(
&user,
crate::protos::auth::common::v1::AuthMethod::AccessToken,
);
principal.access_key_id = record.id.to_string();
principal.scopes = scopes.clone();
principal.scopes_bounded = true;
Ok(Response::new(IntrospectTokenResponse {
active: true,
reason: 0,
token_id: record.id.to_string(),
audience: TokenAudience::Unspecified as i32,
principal: Some(principal),
scopes,
security_version: record.security_version,
issued_at: convert::datetime_opt(Some(record.created_at)),
expires_at: convert::datetime_opt(record.expires_at),
}))
}
credential::Kind::ServiceToken(token) => {
let claims = match service_token::verify(
&self.svc.jwt().jwks(),
token.trim(),
&service_token::ALL_AUDIENCES,
) {
Ok(claims) => claims,
Err(AuthError::TokenExpired) => {
return Ok(Response::new(inactive(Code::TokenExpired)));
}
Err(_) => {
return Ok(Response::new(inactive(Code::InvalidAccessToken)));
}
};
if !crate::jwt::scope::valid(&required_scopes)
|| !store::scopes_cover(&claims.scopes, &required_scopes)
{
return Ok(Response::new(inactive(Code::ScopeInsufficient)));
}
let revoked_key = format!("{REVOKED_PREFIX}{}", claims.token_id);
let revoked_in_cache = self
.svc
.cache()
.get::<String>(&revoked_key)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
.is_some();
if !revoked_in_cache {
// Redis 撤销记录过期/丢失时回查签发时落库的撤销表，防止撤销被绕过。
let persisted = service_token_revocations::Entity::find_by_id(&claims.token_id)
.one(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if persisted.is_some_and(|row| row.revoked_at.is_some()) {
return Ok(Response::new(inactive(Code::TokenRevoked)));
}
} else {
return Ok(Response::new(inactive(Code::TokenRevoked)));
}
let user = match store::load_active_user(self.svc.db(), claims.user_id).await {
Ok(user) => user,
Err(AuthError::UserNotFound) => {
return Ok(Response::new(inactive(Code::Unauthenticated)));
}
Err(err) => return Err(error::status(&err, request_id)),
};
if user.security_version != claims.security_version {
return Ok(Response::new(inactive(Code::SecurityVersionStale)));
}
Ok(Response::new(IntrospectTokenResponse {
active: true,
reason: 0,
token_id: claims.token_id,
audience: service_token::audience_from_name(&claims.audience)
.map(|audience| audience as i32)
.unwrap_or(0),
principal: None,
scopes: claims.scopes,
security_version: user.security_version,
issued_at: Some(convert::timestamp(claims.issued_at, 0)),
expires_at: Some(convert::timestamp(claims.expires_at, 0)),
}))
}
_ => Err(error::invalid_argument(
"credential kind must be access_token or service_token",
request_id,
)),
}
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn issue_service_token_inner(
&self,
request_id: Option<&str>,
request: IssueServiceTokenRequest,
) -> Result<Response<IssueServiceTokenResponse>, Status> {
let audience =
TokenAudience::try_from(request.audience).unwrap_or(TokenAudience::Unspecified);
if audience == TokenAudience::Unspecified {
return Err(error::invalid_argument(
"audience must be specified",
request_id,
));
}
let ttl = if request.ttl_seconds == 0 {
service_token::DEFAULT_TTL_SECS
} else {
request.ttl_seconds
};
if ttl > service_token::MAX_TTL_SECS {
return Err(error::status(&AuthError::InvalidExpiry, request_id));
}
let user = store::load_active_user(self.svc.db(), request.subject_user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
if !crate::jwt::scope::valid(&request.scopes) {
return Err(error::status(&AuthError::InvalidScope, request_id));
}
let allowed_names: Vec<String> = ALLOWED_APP_SCOPES
.iter()
.map(|name| (*name).to_owned())
.collect();
let allowed_codes = store::scope_codes(&store::scopes_from_names(&allowed_names));
if let Some(missing) = request
.scopes
.iter()
.find(|code| **code != 0 && !allowed_codes.contains(code))
{
return Err(error::status(
&AuthError::ScopeInsufficient {
required: scope_label(*missing),
},
request_id,
));
}
let event_id = request.event_id.trim().to_owned();
if !event_id.is_empty() {
let key = issue_cache_key(&event_id, &request, audience, ttl);
if let Some(cached) = self
.svc
.cache()
.get_json::<CachedIssue>(&key)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
{
return Ok(Response::new(cached.into()));
}
}
let (chat, sfu) = match request.constraint.as_ref() {
Some(Constraint::Chat(grant)) => (Some(service_token::ChatGrant::from(grant)), None),
Some(Constraint::Sfu(grant)) => (None, Some(service_token::SfuGrant::from(grant))),
None => (None, None),
};
let issued_at = crypto::now_secs() as i64;
let expires_at = issued_at + ttl as i64;
if expires_at <= issued_at {
return Err(error::status(&AuthError::TokenExpired, request_id));
}
let token_id = store::new_id().to_string();
let claims = service_token::ServiceTokenClaims {
issuer: self.svc.signing().issuer().to_owned(),
token_type: service_token::TOKEN_TYPE.to_owned(),
user_id: user.id,
audience: service_token::audience_name(audience),
token_id: token_id.clone(),
issued_at,
expires_at,
security_version: user.security_version,
scopes: request.scopes.clone(),
service_id: String::new(),
chat,
sfu,
};
let token = service_token::issue(self.svc.signing(), &claims)
.map_err(|err| error::status(&err, request_id))?;
let ttl_duration = Duration::from_secs((expires_at - issued_at) as u64);
let meta_key = format!("{META_PREFIX}{token_id}");
self.svc
.cache()
.set_json_ex(
&meta_key,
&TokenMeta {
user_id: user.id,
exp: expires_at,
aud: audience as i32,
},
ttl_duration,
)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
// 元数据同时落库：Redis 记录丢失后撤销与内省仍可回查（见 revoke/introspect）。
let persisted = service_token_revocations::ActiveModel {
token_id: Set(token_id.clone()),
user_id: Set(user.id),
audience: Set(audience as i32),
expires_at: Set(chrono::DateTime::from_timestamp(expires_at, 0).unwrap_or_default()),
..Default::default()
}
.insert(self.svc.db().write())
.await;
if let Err(error) = persisted {
tracing::warn!(token_id, %error, "service token metadata persistence failed");
}
let response = IssueServiceTokenResponse {
service_token: token,
token_id,
key_id: self.svc.signing().kid().to_owned(),
subject_user_id: user.id,
security_version: user.security_version,
issued_at: Some(convert::timestamp(issued_at, 0)),
expires_at: Some(convert::timestamp(expires_at, 0)),
};
if !event_id.is_empty() {
let cache_key = issue_cache_key(&event_id, &request, audience, ttl);
self.svc
.cache()
.set_json_ex(&cache_key, &CachedIssue::from(&response), ttl_duration)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
}
Ok(Response::new(response))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn revoke_service_token_inner(
&self,
request_id: Option<&str>,
request: RevokeServiceTokenRequest,
) -> Result<Response<RevokeServiceTokenResponse>, Status> {
let token_id = request.token_id.trim().to_owned();
if token_id.is_empty() {
return Err(error::invalid_argument("token_id is required", request_id));
}
let cache = self.svc.cache();
let now = crypto::now_secs() as i64;
let meta_key = format!("{META_PREFIX}{token_id}");
let cached_meta = cache
.get_json::<TokenMeta>(&meta_key)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let expires_at = if let Some(token_meta) = cached_meta {
token_meta.exp
} else {
// Redis 元数据丢失时回查签发时落库的记录，撤销不再依赖缓存存活。
let persisted = service_token_revocations::Entity::find_by_id(&token_id)
.one(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let Some(persisted) = persisted else {
return Err(Status::not_found("service token metadata is unavailable"));
};
if persisted.revoked_at.is_some() {
return Ok(Response::new(RevokeServiceTokenResponse {
accepted: true,
already_revoked: true,
}));
}
persisted.expires_at.timestamp()
};
let ttl = Duration::from_secs((expires_at - now).max(60) as u64);
let event_id = request.event_id.trim().to_owned();
if !event_id.is_empty() {
let claimed = cache
.set_nx_ex(&format!("{REVOKE_PREFIX}{event_id}"), "1", ttl)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if !claimed {
return Ok(Response::new(RevokeServiceTokenResponse {
accepted: true,
already_revoked: true,
}));
}
}
let revoked_key = format!("{REVOKED_PREFIX}{token_id}");
let already_revoked = !cache
.set_nx_ex(&revoked_key, request.reason.clone(), ttl)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
// 同步持久层撤销状态：Redis 撤销记录丢失后，内省仍可通过表记录识别已撤销令牌。
let marked = service_token_revocations::Entity::update_many()
.col_expr(
service_token_revocations::Column::RevokedAt,
sea_orm::sea_query::Expr::value(chrono::Utc::now()),
)
.col_expr(
service_token_revocations::Column::RevokeReason,
sea_orm::sea_query::Expr::value(request.reason.clone()),
)
.filter(service_token_revocations::Column::TokenId.eq(&token_id))
.filter(service_token_revocations::Column::RevokedAt.is_null())
.exec(self.svc.db().write())
.await;
if let Err(error) = marked {
tracing::warn!(token_id, %error, "service token revocation persistence failed");
}
Ok(Response::new(RevokeServiceTokenResponse {
accepted: true,
already_revoked,
}))
}
}