Files
Genarrative/server-rs/crates/api-server/src/auth.rs
T
kdletters 458371a73d
Project CI / Repository checks (push) Successful in 4m11s
Project CI / Frontend tests (push) Successful in 8m38s
Project CI / Backend tests (push) Successful in 10m42s
Project CI / Native shell tests (push) Successful in 32m11s
将退款 outbox 主路径迁入 SpacetimeDB (#204)
## 变更内容

- 新增 `profile_wallet_refund_outbox` 表与 enqueue/process procedure,退款主路径进入 SpacetimeDB。
- 外部生成失败事务、inline 资产失败和跨节点 worker 统一使用库内 outbox,按 ledger 幂等并在事务内完成退款与删除。
- SpacetimeDB 完全不可达时才写本机 emergency spool,恢复时重新入库;兼容旧 spool 文件并保留 attempt 追踪。
- 更新 SpacetimeDB migration、生成 bindings、架构文档、运维恢复说明和项目决策记录。

## 验证

- `cargo check -p spacetime-module -p spacetime-client -p api-server --manifest-path server-rs/Cargo.toml`
- api-server / spacetime-client / spacetime-module / module-runtime 定向测试
- `npm run check:spacetime-schema`
- `npm run check:spacetime-runtime-access`
- `npm run check:server-rs-ddd`
- `npm run check:encoding`
- `git diff --check`

Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/204
Co-authored-by: kdletters <kdletters@qq.com>
Co-committed-by: kdletters <kdletters@qq.com>
2026-08-27 22:04:50 +08:00

476 lines
15 KiB
Rust

use axum::{
Json,
extract::{Extension, Request, State},
http::{
HeaderMap, StatusCode,
header::{AUTHORIZATION, COOKIE},
},
middleware::Next,
response::Response,
};
use platform_auth::{AccessTokenClaims, read_refresh_session_token, verify_access_token};
#[cfg(any())]
use platform_auth::{
RUNTIME_GUEST_SCOPE_PUBLIC_PLAY, RuntimeGuestTokenClaims, RuntimeGuestTokenClaimsInput,
sign_runtime_guest_token, verify_runtime_guest_token,
};
use serde_json::{Value, json};
#[cfg(any())]
use shared_contracts::auth::RuntimeGuestTokenResponse;
#[cfg(any())]
use shared_kernel::{format_rfc3339, new_uuid_simple_string};
#[cfg(test)]
use time::OffsetDateTime;
use tracing::warn;
use crate::{
api_response::json_success_body, http_error::AppError, request_context::RequestContext,
state::AppState,
};
// 统一把已校验的 claims 写入 request extensions,避免后续 handler 再次重复解析 Bearer token。
#[derive(Clone, Debug)]
pub struct AuthenticatedAccessToken {
claims: AccessTokenClaims,
}
#[derive(Clone, Debug)]
pub struct RefreshSessionToken {
token: String,
}
#[derive(Clone, Debug)]
#[cfg(any())]
pub enum RuntimePrincipal {
User(AuthenticatedAccessToken),
Guest(RuntimeGuestTokenClaims),
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg(any())]
pub enum RuntimePrincipalKind {
User,
Guest,
}
impl AuthenticatedAccessToken {
pub fn new(claims: AccessTokenClaims) -> Self {
Self { claims }
}
pub fn claims(&self) -> &AccessTokenClaims {
&self.claims
}
}
impl RefreshSessionToken {
pub fn new(token: String) -> Self {
Self { token }
}
pub fn token(&self) -> &str {
&self.token
}
}
#[cfg(any())]
impl RuntimePrincipal {
pub fn subject(&self) -> &str {
match self {
Self::User(authenticated) => authenticated.claims().user_id(),
Self::Guest(claims) => claims.subject(),
}
}
pub fn kind(&self) -> RuntimePrincipalKind {
match self {
Self::User(_) => RuntimePrincipalKind::User,
Self::Guest(_) => RuntimePrincipalKind::Guest,
}
}
}
#[cfg(any())]
impl RuntimePrincipalKind {
pub fn as_str(self) -> &'static str {
match self {
Self::User => "user",
Self::Guest => "guest",
}
}
}
#[cfg(any())]
pub async fn issue_runtime_guest_token(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
) -> Result<Json<Value>, AppError> {
let issued_at = OffsetDateTime::now_utc();
let claims = RuntimeGuestTokenClaims::from_input(
RuntimeGuestTokenClaimsInput {
subject: format!("guest-runtime-{}", new_uuid_simple_string()),
scope: RUNTIME_GUEST_SCOPE_PUBLIC_PLAY.to_string(),
},
state.auth_jwt_config(),
issued_at,
)
.map_err(|error| {
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_message(error.to_string())
})?;
let token = sign_runtime_guest_token(&claims, state.auth_jwt_config()).map_err(|error| {
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_message(error.to_string())
})?;
let expires_at = OffsetDateTime::from_unix_timestamp(claims.expires_at_unix() as i64)
.ok()
.and_then(|value| format_rfc3339(value).ok())
.unwrap_or_else(|| "1970-01-01T00:00:00Z".to_string());
Ok(json_success_body(
Some(&request_context),
RuntimeGuestTokenResponse {
token,
expires_at,
subject: claims.subject().to_string(),
scope: claims.scope().to_string(),
},
))
}
pub async fn require_bearer_auth(
State(state): State<AppState>,
mut request: Request,
next: Next,
) -> Result<Response, AppError> {
let headers = request.headers().clone();
let request_id = request_id_from_request(&request);
let Some(authenticated) = authenticate_request(&state, headers, request_id).await? else {
return Err(AppError::from_status(StatusCode::UNAUTHORIZED));
};
// JWT 会话校验走 SpacetimeDB;随后刷新用户/身份投影,保证所有受保护路由在
// 读取进程内工作集时都不会依赖粘性会话命中创建或更新它的 API 节点。
state
.refresh_auth_store_from_spacetime()
.await
.map_err(|error| {
warn!(error = %error, "受保护请求刷新认证投影失败");
AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
.with_message("认证状态服务暂不可用")
})?;
request.extensions_mut().insert(authenticated.clone());
let mut response = next.run(request).await;
response.extensions_mut().insert(authenticated);
Ok(response)
}
#[cfg(any())]
pub async fn require_runtime_principal_auth(
State(state): State<AppState>,
mut request: Request,
next: Next,
) -> Result<Response, AppError> {
let headers = request.headers().clone();
let request_id = request_id_from_request(&request);
let Some(principal) = authenticate_runtime_principal(&state, headers, request_id).await? else {
return Err(AppError::from_status(StatusCode::UNAUTHORIZED));
};
request.extensions_mut().insert(principal.clone());
let mut response = next.run(request).await;
response.extensions_mut().insert(principal);
Ok(response)
}
#[cfg(any())]
async fn authenticate_runtime_principal(
state: &AppState,
headers: HeaderMap,
request_id: String,
) -> Result<Option<RuntimePrincipal>, AppError> {
if !headers.contains_key(AUTHORIZATION) {
return Ok(None);
}
match authenticate_request(state, headers.clone(), request_id.clone()).await {
Ok(Some(authenticated)) => Ok(Some(RuntimePrincipal::User(authenticated))),
Ok(None) => Ok(None),
Err(_) => {
let bearer_token = extract_bearer_token(&headers)?;
let claims = verify_runtime_guest_token(&bearer_token, state.auth_jwt_config())
.map_err(|error| {
warn!(
%request_id,
error = %error,
"runtime guest JWT 校验失败"
);
AppError::from_status(StatusCode::UNAUTHORIZED)
})?;
if claims.scope() != RUNTIME_GUEST_SCOPE_PUBLIC_PLAY {
warn!(
%request_id,
scope = %claims.scope(),
"runtime guest JWT scope 非法"
);
return Err(AppError::from_status(StatusCode::UNAUTHORIZED));
}
Ok(Some(RuntimePrincipal::Guest(claims)))
}
}
}
pub async fn optional_access_token_from_headers(
state: &AppState,
_path: String,
headers: HeaderMap,
request_id: String,
) -> Result<Option<AuthenticatedAccessToken>, AppError> {
authenticate_request(state, headers, request_id).await
}
async fn authenticate_request(
state: &AppState,
headers: HeaderMap,
request_id: String,
) -> Result<Option<AuthenticatedAccessToken>, AppError> {
if !headers.contains_key(AUTHORIZATION) {
return Ok(None);
}
let bearer_token = extract_bearer_token(&headers)?;
let claims = verify_access_token(&bearer_token, state.auth_jwt_config()).map_err(|error| {
warn!(
%request_id,
error = %error,
"Bearer JWT 校验失败"
);
AppError::from_status(StatusCode::UNAUTHORIZED)
})?;
#[cfg(not(test))]
let session_is_active = state
.spacetime_client()
.validate_auth_session(spacetime_client::AuthSessionValidationRecordInput {
user_id: claims.user_id().to_string(),
session_id: claims.session_id().to_string(),
token_version: claims.token_version(),
})
.await
.map_err(|error| {
warn!(
%request_id,
user_id = %claims.user_id(),
session_id = %claims.session_id(),
error = %error,
"Bearer JWT SpacetimeDB 会话状态读取失败"
);
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
#[cfg(test)]
let session_is_active = {
let current_user = state
.auth_user_service()
.get_user_by_id(claims.user_id())
.map_err(|error| {
warn!(
%request_id,
error = %error,
"Bearer JWT 用户快照读取失败"
);
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
let Some(current_user) = current_user else {
warn!(
%request_id,
user_id = %claims.user_id(),
"Bearer JWT 对应用户不存在"
);
return Err(AppError::from_status(StatusCode::UNAUTHORIZED));
};
if current_user.token_version != claims.token_version() {
warn!(
%request_id,
user_id = %claims.user_id(),
token_version = claims.token_version(),
current_token_version = current_user.token_version,
"Bearer JWT 版本已失效"
);
return Err(AppError::from_status(StatusCode::UNAUTHORIZED)
.with_message("当前登录态已失效,请重新登录"));
}
state
.refresh_session_service()
.is_session_active_for_user(
claims.user_id(),
claims.session_id(),
OffsetDateTime::now_utc(),
)
.map_err(|error| {
warn!(
%request_id,
user_id = %claims.user_id(),
session_id = %claims.session_id(),
error = %error,
"Bearer JWT refresh session 状态读取失败"
);
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR)
})?
};
if !session_is_active {
warn!(
%request_id,
user_id = %claims.user_id(),
session_id = %claims.session_id(),
"Bearer JWT 对应 refresh session 已失效"
);
return Err(AppError::from_status(StatusCode::UNAUTHORIZED)
.with_message("当前登录态已失效,请重新登录"));
}
Ok(Some(AuthenticatedAccessToken::new(claims)))
}
fn request_id_from_request(request: &Request) -> String {
request
.extensions()
.get::<RequestContext>()
.map(|context| context.request_id().to_string())
.unwrap_or_else(|| "unknown".to_string())
}
pub async fn inspect_auth_claims(
Extension(request_context): Extension<RequestContext>,
Extension(authenticated): Extension<AuthenticatedAccessToken>,
) -> Json<Value> {
json_success_body(
Some(&request_context),
json!({
"claims": authenticated.claims(),
}),
)
}
pub async fn attach_refresh_session_token(
State(state): State<AppState>,
mut request: Request,
next: Next,
) -> Response {
if let Some(token) = request
.headers()
.get(COOKIE)
.and_then(|value| value.to_str().ok())
.and_then(|cookie_header| {
read_refresh_session_token(cookie_header, state.refresh_cookie_config())
})
{
request
.extensions_mut()
.insert(RefreshSessionToken::new(token));
}
next.run(request).await
}
pub async fn inspect_refresh_session_cookie(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
request: Request,
) -> Json<Value> {
let maybe_token = request.extensions().get::<RefreshSessionToken>();
json_success_body(
Some(&request_context),
json!({
"cookieName": state.refresh_cookie_config().cookie_name(),
"present": maybe_token.is_some(),
"tokenLength": maybe_token.map(|token| token.token().len()),
}),
)
}
fn extract_bearer_token(headers: &HeaderMap) -> Result<String, AppError> {
let authorization = headers
.get(AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.map(str::trim)
.ok_or_else(|| AppError::from_status(StatusCode::UNAUTHORIZED))?;
let token = authorization
.strip_prefix("Bearer ")
.or_else(|| authorization.strip_prefix("bearer "))
.map(str::trim)
.filter(|token| !token.is_empty())
.ok_or_else(|| AppError::from_status(StatusCode::UNAUTHORIZED))?;
Ok(token.to_string())
}
#[cfg(test)]
mod tests {
use super::{RefreshSessionToken, extract_bearer_token, optional_access_token_from_headers};
use crate::{config::AppConfig, state::AppState};
use axum::{
http::{HeaderMap, HeaderValue, StatusCode, header::AUTHORIZATION},
response::IntoResponse,
};
#[test]
fn extract_bearer_token_accepts_standard_header() {
let mut headers = HeaderMap::new();
headers.insert(
AUTHORIZATION,
HeaderValue::from_static("Bearer token-value"),
);
let token = extract_bearer_token(&headers).expect("bearer token should be extracted");
assert_eq!(token, "token-value");
}
#[test]
fn extract_bearer_token_rejects_missing_scheme() {
let mut headers = HeaderMap::new();
headers.insert(AUTHORIZATION, HeaderValue::from_static("Basic abc"));
let error = extract_bearer_token(&headers).expect_err("basic auth should be rejected");
assert_eq!(error.into_response().status(), StatusCode::UNAUTHORIZED);
}
#[test]
fn refresh_session_token_retains_original_value() {
let token = RefreshSessionToken::new("refresh-token-01".to_string());
assert_eq!(token.token(), "refresh-token-01");
}
#[tokio::test]
async fn internal_forwarding_headers_do_not_authenticate_without_bearer_token() {
let mut config = AppConfig::default();
config.internal_api_secret = Some("bridge-secret".to_string());
let state = AppState::new(config).expect("state should build");
let mut headers = HeaderMap::new();
headers.insert(
"x-genarrative-authenticated-user-id",
HeaderValue::from_static("user_forwarded_01"),
);
headers.insert(
"x-genarrative-internal-api-secret",
HeaderValue::from_static("bridge-secret"),
);
let authenticated = optional_access_token_from_headers(
&state,
"/api/runtime/puzzle/works".to_string(),
headers,
"req-retired-internal-forward".to_string(),
)
.await
.expect("forwarding headers should be ignored");
assert!(authenticated.is_none());
}
}