diff --git a/apps/aether-gateway/src/handlers/admin/users/batch.rs b/apps/aether-gateway/src/handlers/admin/users/batch.rs index 7dd7f5485..22584fea8 100644 --- a/apps/aether-gateway/src/handlers/admin/users/batch.rs +++ b/apps/aether-gateway/src/handlers/admin/users/batch.rs @@ -1,6 +1,7 @@ use super::{ build_admin_users_bad_request_response, build_admin_users_permission_denied_response, - build_admin_users_read_only_response, disabled_user_policy_detail, disabled_user_policy_field, + build_admin_users_read_only_response, build_admin_users_wallet_permission_denied_response, + disabled_user_policy_detail, disabled_user_policy_field, management_token_may_adjust_admin_wallet_balance, management_token_may_administer_user_accounts, normalize_admin_user_role, }; @@ -111,6 +112,11 @@ enum AdminUserWalletBalanceOperation { Deduct, } +enum AdminBatchWalletBalanceAdjustmentError { + WalletLookup, + BalanceAdjustment, +} + pub(in super::super) async fn build_admin_resolve_user_selection_response( state: &AdminAppState<'_>, _request_context: &AdminRequestContext<'_>, @@ -177,7 +183,7 @@ pub(in super::super) async fn build_admin_user_batch_action_response( if mutation.wallet_balance_adjustment.is_some() && !management_token_may_adjust_admin_wallet_balance(request_context) { - return Ok(build_admin_users_permission_denied_response( + return Ok(build_admin_users_wallet_permission_denied_response( request_context, )); } @@ -290,7 +296,20 @@ pub(in super::super) async fn build_admin_user_batch_action_response( })); continue; } - Err(_) => { + Err(AdminBatchWalletBalanceAdjustmentError::WalletLookup) => { + record_batch_action_interruption( + &resolved.items, + item_index, + false, + "读取用户钱包失败，批次已中止，该用户未执行", + &mut failures, + &mut uncertain_user_ids, + &mut unprocessed_user_ids, + ); + interrupted = true; + break; + } + Err(AdminBatchWalletBalanceAdjustmentError::BalanceAdjustment) => { record_batch_action_interruption( &resolved.items, item_index, @@ -903,13 +922,14 @@ async fn apply_batch_user_wallet_balance_adjustment( user_id: &str, adjustment: AdminUserWalletBalanceAdjustment, operator_id: Option<&str>, -) -> Result { +) -> Result { // Resolve only the wallet ID; the repository clamps the deduction under its row lock. let Some(wallet) = state .find_wallet(aether_data::repository::wallet::WalletLookupKey::UserId( user_id, )) - .await? + .await + .map_err(|_| AdminBatchWalletBalanceAdjustmentError::WalletLookup)? else { return Ok(false); }; @@ -918,7 +938,7 @@ async fn apply_batch_user_wallet_balance_adjustment( AdminUserWalletBalanceOperation::Deduct => -adjustment.amount, }; - Ok(state + state .admin_adjust_wallet_balance( &wallet.id, amount, @@ -927,8 +947,9 @@ async fn apply_batch_user_wallet_balance_adjustment( Some("管理员批量调整用户余额"), true, ) - .await? - .is_some()) + .await + .map_err(|_| AdminBatchWalletBalanceAdjustmentError::BalanceAdjustment) + .map(|result| result.is_some()) } fn build_admin_user_batch_bad_request_response(detail: String) -> Response { diff --git a/apps/aether-gateway/src/handlers/admin/users/mod.rs b/apps/aether-gateway/src/handlers/admin/users/mod.rs index 1ee823993..5a5e9d16a 100644 --- a/apps/aether-gateway/src/handlers/admin/users/mod.rs +++ b/apps/aether-gateway/src/handlers/admin/users/mod.rs @@ -49,9 +49,10 @@ use self::shared::AdminUpdateUserPatch; use self::shared::{ admin_default_user_initial_gift, build_admin_users_bad_request_response, build_admin_users_data_unavailable_response, build_admin_users_permission_denied_response, - build_admin_users_read_only_response, disabled_user_policy_detail, disabled_user_policy_field, - format_optional_datetime_iso8601, legacy_admin_list_policy_mode, - legacy_admin_rate_limit_policy_mode, management_token_may_adjust_admin_wallet_balance, + build_admin_users_read_only_response, build_admin_users_wallet_permission_denied_response, + disabled_user_policy_detail, disabled_user_policy_field, format_optional_datetime_iso8601, + legacy_admin_list_policy_mode, legacy_admin_rate_limit_policy_mode, + management_token_may_adjust_admin_wallet_balance, management_token_may_administer_user_accounts, normalize_admin_optional_user_email, normalize_admin_user_group_ids, normalize_admin_user_role, normalize_admin_username, validate_admin_user_password, AdminCreateUserApiKeyRequest, AdminCreateUserRequest, diff --git a/apps/aether-gateway/src/handlers/admin/users/shared.rs b/apps/aether-gateway/src/handlers/admin/users/shared.rs index df322a49f..b74bd0c9a 100644 --- a/apps/aether-gateway/src/handlers/admin/users/shared.rs +++ b/apps/aether-gateway/src/handlers/admin/users/shared.rs @@ -204,6 +204,34 @@ pub(super) fn build_admin_users_permission_denied_response( ) } +pub(super) fn build_admin_users_wallet_permission_denied_response( + request_context: &crate::handlers::admin::request::AdminRequestContext<'_>, +) -> Response { + let actor_id = request_context + .decision() + .and_then(|decision| decision.admin_principal.as_ref()) + .and_then(|principal| principal.management_token_id.as_deref()) + .unwrap_or("unknown"); + crate::handlers::admin::shared::attach_admin_audit_response( + ( + http::StatusCode::FORBIDDEN, + Json(json!({ + "detail": "management token permission denied", + "required_permissions": ["admin:wallets:write", "admin:wallets:admin"], + "permission_mode": "any_of", + "route_family": request_context.route_family(), + "route_kind": request_context.route_kind(), + "request_path": request_context.path(), + })), + ) + .into_response(), + "admin_user_wallet_balance_permission_denied", + "permission_denied", + "admin_user_wallet_balance", + actor_id, + ) +} + pub(super) fn normalize_admin_optional_user_email( value: Option<&str>, ) -> Result, String> { @@ -409,9 +437,52 @@ pub(super) fn format_optional_datetime_iso8601( #[cfg(test)] mod tests { - use super::{normalize_admin_user_api_formats, AdminUpdateUserApiKeyRequest}; + use super::{ + build_admin_users_wallet_permission_denied_response, normalize_admin_user_api_formats, + AdminUpdateUserApiKeyRequest, + }; + use crate::control::{GatewayControlDecision, GatewayPublicRequestContext}; + use crate::handlers::admin::request::AdminRequestContext; + use axum::http::{HeaderMap, Method, Uri}; use serde_json::json; + #[test] + fn wallet_permission_denial_uses_wallet_audit_category() { + let uri: Uri = "/api/admin/users/batch-action" + .parse() + .expect("uri should parse"); + let method = Method::POST; + let headers = HeaderMap::new(); + let decision = GatewayControlDecision::synthetic( + uri.path(), + Some("admin_proxy".to_string()), + Some("users_manage".to_string()), + Some("batch_user_action".to_string()), + Some("admin:users".to_string()), + ); + let context = GatewayPublicRequestContext::from_request_parts( + "trace-wallet-permission-denied", + &method, + &uri, + &headers, + Some(decision), + ); + let request_context = AdminRequestContext::new(&context); + + let response = build_admin_users_wallet_permission_denied_response(&request_context); + let event = response + .extensions() + .get::() + .expect("wallet denial should attach an audit event"); + + assert_eq!( + event.event_name, + "admin_user_wallet_balance_permission_denied" + ); + assert_eq!(event.action, "permission_denied"); + assert_eq!(event.target_type, "admin_user_wallet_balance"); + } + #[test] fn admin_user_api_formats_accept_current_canonical_signatures() { assert_eq!( diff --git a/apps/aether-gateway/src/state/app.rs b/apps/aether-gateway/src/state/app.rs index 3f2c7657d..a3996a1ab 100644 --- a/apps/aether-gateway/src/state/app.rs +++ b/apps/aether-gateway/src/state/app.rs @@ -494,6 +494,8 @@ pub struct AppState { #[cfg(test)] pub(crate) auth_wallet_adjustment_error_for_tests: Option, #[cfg(test)] + pub(crate) auth_wallet_lookup_error_for_tests: Option, + #[cfg(test)] pub(crate) admin_wallet_payment_order_store: Option>>>, #[cfg(test)] diff --git a/apps/aether-gateway/src/state/core.rs b/apps/aether-gateway/src/state/core.rs index 7e3e060ae..8acfaf40c 100644 --- a/apps/aether-gateway/src/state/core.rs +++ b/apps/aether-gateway/src/state/core.rs @@ -468,6 +468,8 @@ impl AppState { #[cfg(test)] auth_wallet_adjustment_error_for_tests: None, #[cfg(test)] + auth_wallet_lookup_error_for_tests: None, + #[cfg(test)] admin_wallet_payment_order_store: Some(Arc::new(StdMutex::new(HashMap::new()))), #[cfg(test)] admin_payment_callback_store: Some(Arc::new(StdMutex::new(HashMap::new()))), diff --git a/apps/aether-gateway/src/state/runtime/wallet/reads.rs b/apps/aether-gateway/src/state/runtime/wallet/reads.rs index 90896a419..a0bf2d187 100644 --- a/apps/aether-gateway/src/state/runtime/wallet/reads.rs +++ b/apps/aether-gateway/src/state/runtime/wallet/reads.rs @@ -5,6 +5,19 @@ impl AppState { &self, lookup: aether_data::repository::wallet::WalletLookupKey<'_>, ) -> Result, GatewayError> { + #[cfg(test)] + if let Some(failed_user_id) = self.auth_wallet_lookup_error_for_tests.as_deref() { + let lookup_user_id = match &lookup { + aether_data::repository::wallet::WalletLookupKey::UserId(user_id) => Some(*user_id), + _ => None, + }; + if lookup_user_id == Some(failed_user_id) { + return Err(GatewayError::Internal( + "injected test wallet lookup failure".to_string(), + )); + } + } + #[cfg(test)] if let Some(store) = self.auth_wallet_store.as_ref() { let wallet = { diff --git a/apps/aether-gateway/src/state/testing.rs b/apps/aether-gateway/src/state/testing.rs index cfa990a32..30cb1c079 100644 --- a/apps/aether-gateway/src/state/testing.rs +++ b/apps/aether-gateway/src/state/testing.rs @@ -480,6 +480,11 @@ impl AppState { self } + pub(crate) fn fail_auth_wallet_lookup_for_tests(mut self, user_id: impl Into) -> Self { + self.auth_wallet_lookup_error_for_tests = Some(user_id.into()); + self + } + pub(crate) fn with_admin_wallet_payment_orders_for_tests(mut self, orders: I) -> Self where I: IntoIterator, diff --git a/apps/aether-gateway/src/tests/control/admin/users_batch.rs b/apps/aether-gateway/src/tests/control/admin/users_batch.rs index 2a1cb6466..bb6fa0308 100644 --- a/apps/aether-gateway/src/tests/control/admin/users_batch.rs +++ b/apps/aether-gateway/src/tests/control/admin/users_batch.rs @@ -176,6 +176,7 @@ async fn gateway_batches_wallet_addition_deduction_and_clamped_deduction_per_use async fn gateway_requires_wallet_write_permission_for_batch_balance_adjustments() { let users_write_token = "ae-batch-users-write-only"; let wallet_write_token = "ae-batch-users-wallet-write"; + let wallet_admin_token = "ae-batch-users-wallet-admin"; let token_owner = sample_user_with_role("token-owner", "admin"); let target_user = sample_user("user-1"); let mut users_only = sample_management_token( @@ -194,8 +195,16 @@ async fn gateway_requires_wallet_write_permission_for_batch_balance_adjustments( ); users_and_wallets.token.allowed_ips = None; users_and_wallets.token.permissions = Some(json!(["admin:users:write", "admin:wallets:write"])); + let mut wallets_admin = sample_management_token( + "token-users-wallet-admin", + &token_owner.id, + &token_owner.username, + true, + ); + wallets_admin.token.allowed_ips = None; + wallets_admin.token.permissions = Some(json!(["admin:users:write", "admin:wallets:admin"])); let token_repository = Arc::new(InMemoryManagementTokenRepository::seed_with_hashes( - vec![users_only, users_and_wallets], + vec![users_only, users_and_wallets, wallets_admin], vec![ ( hash_management_token(users_write_token), @@ -205,6 +214,10 @@ async fn gateway_requires_wallet_write_permission_for_batch_balance_adjustments( hash_management_token(wallet_write_token), "token-users-and-wallet-write".to_string(), ), + ( + hash_management_token(wallet_admin_token), + "token-users-wallet-admin".to_string(), + ), ], )); let user_repository = Arc::new(InMemoryUserReadRepository::seed_auth_users(vec![ @@ -235,6 +248,12 @@ async fn gateway_requires_wallet_write_permission_for_batch_balance_adjustments( .await .expect("users-only management token request should complete"); assert_eq!(denied.status(), StatusCode::FORBIDDEN); + let denied_payload: Value = denied.json().await.expect("response should parse"); + assert_eq!( + denied_payload["required_permissions"], + json!(["admin:wallets:write", "admin:wallets:admin"]) + ); + assert_eq!(denied_payload["permission_mode"], "any_of"); assert_eq!( wallet_detail(&client, &gateway_url, "user-1").await["balance"], 10.0 @@ -256,6 +275,25 @@ async fn gateway_requires_wallet_write_permission_for_batch_balance_adjustments( 15.0 ); + let allowed_with_wallet_admin = client + .post(format!("{gateway_url}/api/admin/users/batch-action")) + .header(crate::constants::GATEWAY_HEADER, "rust-phase3b") + .bearer_auth(wallet_admin_token) + .json(&payload) + .send() + .await + .expect("wallet-admin management token request should complete"); + assert_eq!(allowed_with_wallet_admin.status(), StatusCode::OK); + let result: Value = allowed_with_wallet_admin + .json() + .await + .expect("response should parse"); + assert_eq!(result["success"], 1); + assert_eq!( + wallet_detail(&client, &gateway_url, "user-1").await["balance"], + 20.0 + ); + gateway_handle.abort(); } @@ -313,6 +351,56 @@ async fn gateway_reports_completed_uncertain_and_unprocessed_users_after_adjustm gateway_handle.abort(); } +#[tokio::test] +async fn gateway_reports_wallet_lookup_failure_as_unprocessed() { + let state = AppState::new() + .expect("gateway should build") + .with_auth_users_for_tests([ + sample_user("user-1"), + sample_user("user-2"), + sample_user("user-3"), + ]) + .with_auth_wallets_for_tests([ + sample_wallet("user-1", 10.0, 0.0), + sample_wallet("user-2", 20.0, 0.0), + sample_wallet("user-3", 30.0, 0.0), + ]) + .fail_auth_wallet_lookup_for_tests("user-2"); + let (gateway_url, gateway_handle) = start_server(build_router_with_state(state)).await; + let client = Client::new(); + + let response = post_batch_action( + &client, + &gateway_url, + json!({ + "selection": { "user_ids": ["user-1", "user-2", "user-3"] }, + "action": "adjust_wallet_balance", + "payload": { "operation": "add", "amount": 5.0 } + }), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let result: Value = response.json().await.expect("response should parse"); + assert_eq!(result["interrupted"], true); + assert_eq!(result["completed_user_ids"], json!(["user-1"])); + assert_eq!(result["uncertain_user_ids"], json!([])); + assert_eq!(result["unprocessed_user_ids"], json!(["user-2", "user-3"])); + assert_eq!( + wallet_detail(&client, &gateway_url, "user-1").await["balance"], + 15.0 + ); + assert_eq!( + wallet_detail(&client, &gateway_url, "user-2").await["balance"], + 20.0 + ); + assert_eq!( + wallet_detail(&client, &gateway_url, "user-3").await["balance"], + 30.0 + ); + + gateway_handle.abort(); +} + #[tokio::test] async fn gateway_reports_missing_wallet_and_skips_zero_delta_for_non_positive_balance() { let state = AppState::new()