e870254
use std::time::Duration;
use crate::models::users::users;
use crate::protos::auth::common::v1::Code;
use crate::protos::auth::v1::authorization_service_server::{
AuthorizationService, AuthorizationServiceServer,
};
use crate::protos::auth::v1::*;
use chrono::Utc;
use sea_orm::EntityTrait;
use tonic::{Request, Response, Status};
use crate::server::AuthSvc;
use crate::server::access_token::types::ALLOWED_APP_SCOPES;
use crate::server::errors::{AuthError, AuthResult};
use crate::server::grpc::{agreement, context, error, store};
const MAX_BATCH_CAPABILITIES: usize = 32;
const ADMIN_CACHE_TTL_SECS: u64 = 300;
fn requires_admin(capability: PlatformCapability) -> bool {
matches!(
capability,
PlatformCapability::AdminPanel
| PlatformCapability::ManageUsers
| PlatformCapability::ManagePlatform
)
}
#[derive(Clone)]
pub struct AuthorizationApi {
svc: AuthSvc,
}
impl AuthorizationApi {
pub fn new(svc: AuthSvc) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
AuthorizationServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().oauth.rpc_key.clone());
AuthorizationServiceServer::with_interceptor(self, gate)
}
async fn decide(
&self,
user_id: i64,
capability: PlatformCapability,
require_mfa: bool,
) -> AuthResult<(bool, Code)> {
let user = users::Entity::find_by_id(user_id)
.one(self.svc.db().read())
.await?;
let Some(user) = user else {
return Ok((false, Code::UserDisabled));
};
if user.deleted_at.is_some() || !user.allow_use {
return Ok((false, Code::UserDisabled));
}
if user.forbid_login
&& user
.endtime_forbid_login_at
.is_none_or(|until| until > Utc::now())
{
return Ok((false, Code::LoginForbidden));
}
if require_mfa {
return Ok((false, Code::MfaRequired));
}
if requires_admin(capability) && !user.is_admin {
return Ok((false, Code::Forbidden));
}
Ok((true, Code::Unspecified))
}
async fn agreements_pending(&self, user_id: i64) -> AuthResult<bool> {
Ok(!agreement::pending_keys(&self.svc, user_id)
.await?
.is_empty())
}
}
#[tonic::async_trait]
impl AuthorizationService for AuthorizationApi {
async fn authorize(
&self,
request: Request<AuthorizeRequest>,
) -> Result<Response<AuthorizeResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuthorizationService");
let result = self
.authorize_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("Authorize", &result);
result
}
async fn batch_authorize(
&self,
request: Request<BatchAuthorizeRequest>,
) -> Result<Response<BatchAuthorizeResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuthorizationService");
let result = self
.batch_authorize_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("BatchAuthorize", &result);
result
}
async fn is_platform_admin(
&self,
request: Request<IsPlatformAdminRequest>,
) -> Result<Response<IsPlatformAdminResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuthorizationService");
let result = self
.is_platform_admin_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("IsPlatformAdmin", &result);
result
}
async fn get_capabilities(
&self,
request: Request<GetCapabilitiesRequest>,
) -> Result<Response<GetCapabilitiesResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.AuthorizationService");
let result = self
.get_capabilities_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetCapabilities", &result);
result
}
}
impl AuthorizationApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn authorize_inner(
&self,
request_id: Option<&str>,
request: AuthorizeRequest,
) -> Result<Response<AuthorizeResponse>, Status> {
let inner = &request;
let capability = PlatformCapability::try_from(inner.capability)
.unwrap_or(PlatformCapability::Unspecified);
if capability == PlatformCapability::Unspecified {
return Err(error::invalid_argument(
"capability must be specified",
request_id,
));
}
let (mut allowed, mut reason) = self
.decide(inner.user_id, capability, inner.require_mfa)
.await
.map_err(|error_value| error::status(&error_value, request_id))?;
if allowed
&& self
.agreements_pending(inner.user_id)
.await
.map_err(|error_value| error::status(&error_value, request_id))?
{
allowed = false;
reason = Code::AgreementNotAccepted;
}
Ok(Response::new(AuthorizeResponse {
allowed,
reason: reason as i32,
cached: false,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn batch_authorize_inner(
&self,
request_id: Option<&str>,
request: BatchAuthorizeRequest,
) -> Result<Response<BatchAuthorizeResponse>, Status> {
let inner = &request;
if inner.capabilities.len() > MAX_BATCH_CAPABILITIES {
return Err(error::invalid_argument(
format!("at most {MAX_BATCH_CAPABILITIES} capabilities can be checked at once"),
request_id,
));
}
let agreements_pending = self
.agreements_pending(inner.user_id)
.await
.map_err(|error_value| error::status(&error_value, request_id))?;
let mut decisions = Vec::with_capacity(inner.capabilities.len());
for raw in &inner.capabilities {
let capability =
PlatformCapability::try_from(*raw).unwrap_or(PlatformCapability::Unspecified);
let (mut allowed, mut reason) = if capability == PlatformCapability::Unspecified {
(false, Code::InvalidArgument)
} else {
self.decide(inner.user_id, capability, inner.require_mfa)
.await
.map_err(|error_value| error::status(&error_value, request_id))?
};
if allowed && agreements_pending {
allowed = false;
reason = Code::AgreementNotAccepted;
}
decisions.push(batch_authorize_response::Decision {
capability: *raw,
allowed,
reason: reason as i32,
});
}
Ok(Response::new(BatchAuthorizeResponse { decisions }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn is_platform_admin_inner(
&self,
request_id: Option<&str>,
request: IsPlatformAdminRequest,
) -> Result<Response<IsPlatformAdminResponse>, Status> {
let user_id = request.user_id;
let cache_key = format!("auth:admin:{user_id}");
if let Some((is_admin, security_version)) = self
.svc
.cache()
.get_json::<(bool, i32)>(&cache_key)
.await
.map_err(|error_value| error::status(&AuthError::from(error_value), request_id))?
{
return Ok(Response::new(IsPlatformAdminResponse {
is_admin,
security_version,
}));
}
let user = users::Entity::find_by_id(user_id)
.one(self.svc.db().read())
.await
.map_err(|error_value| error::status(&AuthError::from(error_value), request_id))?;
let Some(user) = user else {
return Ok(Response::new(IsPlatformAdminResponse {
is_admin: false,
security_version: 0,
}));
};
let _ = self
.svc
.cache()
.set_json_ex(
&cache_key,
&(user.is_admin, user.security_version),
Duration::from_secs(ADMIN_CACHE_TTL_SECS),
)
.await;
Ok(Response::new(IsPlatformAdminResponse {
is_admin: user.is_admin,
security_version: user.security_version,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_capabilities_inner(
&self,
request_id: Option<&str>,
request: GetCapabilitiesRequest,
) -> Result<Response<GetCapabilitiesResponse>, Status> {
let user_id = request.user_id;
let user = users::Entity::find_by_id(user_id)
.one(self.svc.db().read())
.await
.map_err(|error_value| error::status(&AuthError::from(error_value), request_id))?
.ok_or_else(|| error::status(&AuthError::UserNotFound, request_id))?;
let mfa_enabled = store::has_enabled_mfa(self.svc.db(), user_id)
.await
.map_err(|error_value| error::status(&error_value, request_id))?;
// RPC 面已由共享密钥信任，调用方看到的是账号的全部可用 scope。
let scopes: Vec<i32> = (1..=ALLOWED_APP_SCOPES.len() as i32).collect();
let mut capabilities = Vec::new();
for raw in 1..=9i32 {
let Ok(capability) = PlatformCapability::try_from(raw) else {
continue;
};
let (allowed, _) = self
.decide(user_id, capability, false)
.await
.map_err(|error_value| error::status(&error_value, request_id))?;
if allowed {
capabilities.push(raw);
}
}
Ok(Response::new(GetCapabilitiesResponse {
allow_use: user.allow_use,
forbid_login: user.forbid_login,
email_verified: user.email_verified_at.is_some(),
mfa_enabled,
is_admin: user.is_admin,
security_version: user.security_version,
scopes,
capabilities,
pending_agreements: agreement::pending_keys(&self.svc, user_id)
.await
.map_err(|error_value| error::status(&error_value, request_id))?,
}))
}
}