diff --git a/backend/src/admin/accounts/routes.rs b/backend/src/admin/accounts/routes.rs index 5fb65a0d8..746a5a5dd 100644 --- a/backend/src/admin/accounts/routes.rs +++ b/backend/src/admin/accounts/routes.rs @@ -2,10 +2,10 @@ use axum::{ body::Body, - extract::{Extension, Query, State}, + extract::{Query, State}, http::{ header::{CACHE_CONTROL, CONNECTION, CONTENT_TYPE}, - HeaderMap, StatusCode, + StatusCode, }, response::{IntoResponse, Response}, Json, @@ -14,7 +14,7 @@ use std::collections::HashMap; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -use serde_json::{json, Value}; +use serde_json::Value; use sqlx::{QueryBuilder, Row, Sqlite}; use crate::{ @@ -31,13 +31,8 @@ use crate::{ AdminAccountRefreshOutcome, AdminAccountRefreshResult, AdminAccountUpdate, OAuthExchangeInput, }, - monitoring::{ - account_usage_service::AdminUsageRecord, - billing, - usage_record_model::{UsageRecord, UsageRecordLevel}, - }, + monitoring::{account_usage_service::AdminUsageRecord, billing}, }, - http::middleware::request_id::RequestId, infra::{ format::{ format_cost, format_percent, format_plain_number, format_tokens, nonnegative_i64_to_u64, @@ -552,55 +547,23 @@ pub(crate) async fn create_account( /// `GET /api/admin/accounts/export` pub(crate) async fn export_accounts( State(state): State, - request_id: Option>, _auth: AdminAuth, - headers: HeaderMap, Query(query): Query, ) -> Result { - let request_id = request_id.map(|Extension(request_id)| request_id.as_str().to_string()); - let source = account_action_source(&headers); let ids = account_export_ids(query.ids.as_deref()); if query.confirm.as_deref() != Some(ACCOUNT_EXPORT_CONFIRMATION) { - record_account_export_audit( - &state, - request_id.as_deref(), - &source, - &ids, - AccountExportAuditOutcome::MissingConfirmation, - ) - .await; return Err(AdminError::bad_request( "account export requires confirm=export_sensitive_accounts", )); } - match state.services.admin_accounts.export(ids.clone()).await { - Ok(result) => { - record_account_export_audit( - &state, - request_id.as_deref(), - &source, - &ids, - AccountExportAuditOutcome::Success, - ) - .await; - Ok(AdminResponse::new( - StatusCode::OK, - AdminEnvelope::ok(result), - )) - } - Err(error) => { - record_account_export_audit( - &state, - request_id.as_deref(), - &source, - &ids, - AccountExportAuditOutcome::from_error(&error), - ) - .await; - Err(account_error(&error)) - } + match state.services.admin_accounts.export(ids).await { + Ok(result) => Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(result), + )), + Err(error) => Err(account_error(&error)), } } @@ -1228,96 +1191,6 @@ fn account_not_found() -> AdminError { AdminError::not_found("Account not found") } -enum AccountExportAuditOutcome { - Success, - MissingConfirmation, - EmptyIds, - NotFound, - Failed, -} - -impl AccountExportAuditOutcome { - fn from_error(error: &AdminAccountError) -> Self { - match error { - AdminAccountError::EmptyIds => Self::EmptyIds, - AdminAccountError::NotFound => Self::NotFound, - _ => Self::Failed, - } - } - - fn level(&self) -> UsageRecordLevel { - match self { - Self::Success => UsageRecordLevel::Warn, - Self::MissingConfirmation | Self::EmptyIds | Self::NotFound => UsageRecordLevel::Warn, - Self::Failed => UsageRecordLevel::Error, - } - } - - fn message(&self) -> &'static str { - match self { - Self::Success => "Admin account export succeeded", - Self::MissingConfirmation => "Admin account export missing confirmation", - Self::EmptyIds => "Admin account export missing account ids", - Self::NotFound => "Admin account export account not found", - Self::Failed => "Admin account export failed", - } - } - - fn status_code(&self) -> i64 { - match self { - Self::Success => i64::from(StatusCode::OK.as_u16()), - Self::MissingConfirmation | Self::EmptyIds => { - i64::from(StatusCode::BAD_REQUEST.as_u16()) - } - Self::NotFound => i64::from(StatusCode::NOT_FOUND.as_u16()), - Self::Failed => i64::from(StatusCode::INTERNAL_SERVER_ERROR.as_u16()), - } - } - - fn failure_class(&self) -> Option<&'static str> { - match self { - Self::Success => None, - Self::MissingConfirmation => Some("missing_confirmation"), - Self::EmptyIds => Some("empty_ids"), - Self::NotFound => Some("account_not_found"), - Self::Failed => Some("export_failed"), - } - } -} - -async fn record_account_export_audit( - state: &AppState, - request_id: Option<&str>, - source: &str, - ids: &[String], - outcome: AccountExportAuditOutcome, -) { - let mut record = UsageRecord::new("admin_account_export", outcome.level(), outcome.message()); - record.request_id = request_id.map(ToString::to_string); - record.route = Some("/api/admin/accounts/export".to_string()); - record.status_code = Some(outcome.status_code()); - record.failure_class = outcome.failure_class().map(ToString::to_string); - record.metadata = json!({ - "source": source, - "accountIds": ids, - "accountCount": ids.len(), - }); - - if let Err(error) = state.services.usage_records.record_audit(record).await { - tracing::warn!(%error, "failed to record account export audit event"); - } -} - -fn account_action_source(headers: &HeaderMap) -> String { - headers - .get("x-real-ip") - .and_then(|value| value.to_str().ok()) - .map(str::trim) - .filter(|value| !value.is_empty()) - .unwrap_or("unknown") - .to_string() -} - fn account_export_ids(value: Option<&str>) -> Vec { value .into_iter() diff --git a/backend/src/admin/accounts/service/importing.rs b/backend/src/admin/accounts/service/importing.rs index b6d817d2d..b7205783d 100644 --- a/backend/src/admin/accounts/service/importing.rs +++ b/backend/src/admin/accounts/service/importing.rs @@ -143,8 +143,15 @@ impl AdminAccountService { chatgpt_user_id.as_deref(), ) .await?; - let next_refresh_at = access_token_expires_at - .map(|expires_at| self.next_refresh_at_for_expires_at(&account_id, expires_at)); + let refresh_token = resolved_tokens + .refresh_token + .map(|token| SecretString::new(token.into())); + let next_refresh_at = if refresh_token.is_some() { + access_token_expires_at + .map(|expires_at| self.next_refresh_at_for_expires_at(&account_id, expires_at)) + } else { + None + }; if plan_type.is_none() { plan_type = supplemental.plan_type; } @@ -163,9 +170,7 @@ impl AdminAccountService { label, plan_type, access_token: SecretString::new(resolved_tokens.access_token.into()), - refresh_token: resolved_tokens - .refresh_token - .map(|token| SecretString::new(token.into())), + refresh_token, access_token_expires_at, status, added_at: None, diff --git a/backend/src/admin/accounts/service/lifecycle.rs b/backend/src/admin/accounts/service/lifecycle.rs index 5cc43d27e..75703b19d 100644 --- a/backend/src/admin/accounts/service/lifecycle.rs +++ b/backend/src/admin/accounts/service/lifecycle.rs @@ -3,7 +3,7 @@ use std::{ hash::{Hash, Hasher}, }; -use chrono::Utc; +use chrono::{DateTime, Utc}; use futures::{stream, StreamExt}; use secrecy::{ExposeSecret, SecretString}; use tokio::time::sleep; @@ -147,8 +147,12 @@ impl AdminAccountService { .refresh_token_for_existing .map(|token| SecretString::new(token.into())), access_token_expires_at: Some(claims.expires_at), - next_refresh_at: Some( - self.next_refresh_at_for_expires_at(&existing.id, claims.expires_at), + next_refresh_at: create_next_refresh_at( + self, + &existing.id, + claims.expires_at, + tokens.refresh_token_for_new.is_some() + || existing.refresh_token.is_some(), ), status: crate::upstream::accounts::model::AccountStatus::Active, }, @@ -460,6 +464,15 @@ impl AdminAccountService { } } +fn create_next_refresh_at( + service: &AdminAccountService, + account_id: &str, + expires_at: DateTime, + has_refresh_token: bool, +) -> Option> { + has_refresh_token.then(|| service.next_refresh_at_for_expires_at(account_id, expires_at)) +} + fn stable_jittered_millis(account_id: &str, base_millis: u64, variance: f64) -> u64 { if base_millis == 0 { return 0; diff --git a/backend/src/admin/accounts/service/testing.rs b/backend/src/admin/accounts/service/testing.rs index 96f1c036f..9c7fac4ab 100644 --- a/backend/src/admin/accounts/service/testing.rs +++ b/backend/src/admin/accounts/service/testing.rs @@ -7,7 +7,9 @@ use serde_json::{json, Value}; use tokio::sync::mpsc; use crate::upstream::{ + accounts::{model::Account, store::StoredAccount}, models::CodexModelInfo, + models::ModelRefreshPlanAccount, protocol::{ responses::CodexResponsesRequest, sse::{encode_sse_event, parse_sse_events, SseEvent}, @@ -37,19 +39,43 @@ impl AdminAccountService { .await .map_err(|_| AdminAccountError::Inspect)? .ok_or(AdminAccountError::NotFound)?; - let plan_type = account.plan_type.as_deref().unwrap_or("default"); - let catalog = self.models.catalog().await; - let models = catalog - .models_for_plan(plan_type) - .iter() - .map(account_model_option) - .collect::>(); + let plan_type = account_plan_type(&account); + let mut models = self.models.catalog().await.models_for_plan(&plan_type); + if models.is_empty() { + self.refresh_account_plan_models(&account, &plan_type).await; + models = self.models.catalog().await.models_for_plan(&plan_type); + } + let models = models.iter().map(account_model_option).collect::>(); if models.is_empty() { return Err(AdminAccountError::NoModels); } Ok(models) } + async fn refresh_account_plan_models(&self, account: &StoredAccount, plan_type: &str) { + let request_id = uuid::Uuid::new_v4().to_string(); + let plan_account = ModelRefreshPlanAccount { + plan_type: plan_type.to_string(), + account: stored_account_to_model_refresh_account(account.clone()), + }; + if let Err(error) = self + .models + .refresh_backend_models_with_installation_id( + &[plan_account], + &request_id, + self.installation_id.as_deref(), + ) + .await + { + tracing::warn!( + account_id = %account.id, + plan_type, + error = %error, + "failed to refresh account plan models" + ); + } + } + pub async fn test_connection_stream( &self, account_id: &str, @@ -143,6 +169,57 @@ fn account_model_option(model: &CodexModelInfo) -> AccountModelOption { } } +fn account_plan_type(account: &StoredAccount) -> String { + account + .plan_type + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or("default") + .to_string() +} + +fn stored_account_to_model_refresh_account(stored: StoredAccount) -> Account { + Account { + id: stored.id, + email: stored.email, + account_id: stored.account_id, + user_id: stored.user_id, + label: stored.label, + plan_type: stored.plan_type, + access_token: stored.access_token.expose_secret().to_string(), + refresh_token: stored + .refresh_token + .map(|token| token.expose_secret().to_string()), + access_token_expires_at: stored.access_token_expires_at, + next_refresh_at: stored.next_refresh_at, + status: stored.status, + quota_limit_reached: false, + quota_verify_required: false, + quota_cooldown_until: None, + cloudflare_cooldown_until: None, + request_count: 0, + empty_response_count: 0, + image_input_tokens: 0, + image_output_tokens: 0, + image_request_count: 0, + image_request_failed_count: 0, + window_request_count: 0, + window_input_tokens: 0, + window_output_tokens: 0, + window_cached_tokens: 0, + window_image_input_tokens: 0, + window_image_output_tokens: 0, + window_image_request_count: 0, + window_image_request_failed_count: 0, + window_started_at: None, + window_reset_at: None, + limit_window_seconds: None, + added_at: stored.added_at, + last_used_at: None, + } +} + fn test_responses_request(model: String) -> CodexResponsesRequest { CodexResponsesRequest { model, diff --git a/backend/src/admin/auth/session.rs b/backend/src/admin/auth/session.rs index 91b83063e..244bedae4 100644 --- a/backend/src/admin/auth/session.rs +++ b/backend/src/admin/auth/session.rs @@ -1,21 +1,18 @@ //! 管理端会话处理器。 use axum::{ - extract::{Extension, FromRequestParts, State}, + extract::{FromRequestParts, State}, http::{header::SET_COOKIE, request::Parts, HeaderMap, HeaderValue, StatusCode}, response::{IntoResponse, Response}, Json, }; use serde::{Deserialize, Serialize}; -use serde_json::json; use crate::{ admin::{ auth::service::AdminSessionError, - monitoring::usage_record_model::{UsageRecord, UsageRecordLevel}, response::{AdminEnvelope, AdminError, AdminResponse}, }, - http::middleware::request_id::RequestId, infra::time::china_rfc3339, runtime::state::AppState, }; @@ -60,16 +57,10 @@ struct AdminSessionStatusData { /// `POST /api/admin/login` pub(crate) async fn login( State(state): State, - request_id: Option>, headers: HeaderMap, Json(payload): Json, ) -> Result { let source = admin_login_source(&headers); - let request_id = request_id.map(|Extension(request_id)| request_id.as_str().to_string()); - let username_provided = payload - .username - .as_deref() - .is_some_and(|username| !username.trim().is_empty()); let login_result = state .services .admin_sessions @@ -80,50 +71,10 @@ pub(crate) async fn login( ) .await; let session = match login_result { - Ok(Some(session)) => { - record_admin_login_audit( - &state, - request_id.as_deref(), - &source, - username_provided, - AdminLoginAuditOutcome::Success, - ) - .await; - session - } - Ok(None) => { - record_admin_login_audit( - &state, - request_id.as_deref(), - &source, - username_provided, - AdminLoginAuditOutcome::InvalidCredentials, - ) - .await; - return Err(AdminError::invalid_admin_credentials()); - } - Err(AdminSessionError::LoginThrottled) => { - record_admin_login_audit( - &state, - request_id.as_deref(), - &source, - username_provided, - AdminLoginAuditOutcome::Throttled, - ) - .await; - return Err(AdminError::too_many_login_attempts()); - } - Err(_) => { - record_admin_login_audit( - &state, - request_id.as_deref(), - &source, - username_provided, - AdminLoginAuditOutcome::InternalError, - ) - .await; - return Err(AdminError::internal("Failed to create admin session")); - } + Ok(Some(session)) => session, + Ok(None) => return Err(AdminError::invalid_admin_credentials()), + Err(AdminSessionError::LoginThrottled) => return Err(AdminError::too_many_login_attempts()), + Err(_) => return Err(AdminError::internal("Failed to create admin session")), }; let mut response = AdminResponse::new( @@ -145,72 +96,6 @@ pub(crate) async fn login( Ok(response) } -enum AdminLoginAuditOutcome { - Success, - InvalidCredentials, - Throttled, - InternalError, -} - -impl AdminLoginAuditOutcome { - fn level(&self) -> UsageRecordLevel { - match self { - Self::Success => UsageRecordLevel::Info, - Self::InvalidCredentials | Self::Throttled => UsageRecordLevel::Warn, - Self::InternalError => UsageRecordLevel::Error, - } - } - - fn message(&self) -> &'static str { - match self { - Self::Success => "Admin login succeeded", - Self::InvalidCredentials => "Admin login failed", - Self::Throttled => "Admin login throttled", - Self::InternalError => "Admin login failed internally", - } - } - - fn status_code(&self) -> i64 { - match self { - Self::Success => i64::from(StatusCode::OK.as_u16()), - Self::InvalidCredentials => i64::from(StatusCode::UNAUTHORIZED.as_u16()), - Self::Throttled => i64::from(StatusCode::TOO_MANY_REQUESTS.as_u16()), - Self::InternalError => i64::from(StatusCode::INTERNAL_SERVER_ERROR.as_u16()), - } - } - - fn failure_class(&self) -> Option<&'static str> { - match self { - Self::Success => None, - Self::InvalidCredentials => Some("invalid_credentials"), - Self::Throttled => Some("login_throttled"), - Self::InternalError => Some("session_error"), - } - } -} - -async fn record_admin_login_audit( - state: &AppState, - request_id: Option<&str>, - source: &str, - username_provided: bool, - outcome: AdminLoginAuditOutcome, -) { - let mut record = UsageRecord::new("admin_auth", outcome.level(), outcome.message()); - record.request_id = request_id.map(ToString::to_string); - record.route = Some("/api/admin/login".to_string()); - record.status_code = Some(outcome.status_code()); - record.failure_class = outcome.failure_class().map(ToString::to_string); - record.metadata = json!({ - "source": source, - "usernameProvided": username_provided, - }); - - if let Err(error) = state.services.usage_records.record_audit(record).await { - tracing::warn!(%error, "failed to record admin login audit event"); - } -} - /// `GET /api/admin/auth/status` pub(crate) async fn session_status( State(state): State, diff --git a/backend/src/admin/monitoring/usage_record_service.rs b/backend/src/admin/monitoring/usage_record_service.rs index 38620e336..16aed423b 100644 --- a/backend/src/admin/monitoring/usage_record_service.rs +++ b/backend/src/admin/monitoring/usage_record_service.rs @@ -277,12 +277,6 @@ impl AdminUsageRecordService { self.append_with_settings(event, settings).await } - /// 记录安全审计事件，不受普通使用记录开关影响。 - pub async fn record_audit(&self, event: UsageRecord) -> Result<(), AdminUsageRecordError> { - let settings = *self.settings.read().await; - self.append_with_settings(event, settings).await - } - async fn append_with_settings( &self, mut event: UsageRecord, diff --git a/backend/src/upstream/accounts/importing.rs b/backend/src/upstream/accounts/importing.rs index b4779824e..98f72d0a1 100644 --- a/backend/src/upstream/accounts/importing.rs +++ b/backend/src/upstream/accounts/importing.rs @@ -1,7 +1,7 @@ //! 账号导入逻辑。 use chrono::{DateTime, Utc}; -use serde_json::Value; +use serde_json::{Map, Value}; use crate::upstream::accounts::model::AccountStatus; use crate::upstream::accounts::token_refresh::{jwt_expiry, JwtExpiry}; @@ -44,6 +44,8 @@ pub enum AccountImportSource { Cpr, /// Sub2API 格式。 Sub2Api, + /// CLIProxyAPI auth 文件格式。 + CliProxyApi, } impl AccountImportSource { @@ -52,6 +54,7 @@ impl AccountImportSource { match self { Self::Cpr => "cpr", Self::Sub2Api => "sub2api", + Self::CliProxyApi => "cliproxyapi", } } } @@ -70,6 +73,7 @@ const ACCOUNT_IMPORT_ACCOUNT_KEYS: &[&str] = &[ "planType", "plan_type", "token", + "at", "accessToken", "access_token", "refreshToken", @@ -82,34 +86,6 @@ const ACCOUNT_IMPORT_ACCOUNT_KEYS: &[&str] = &[ "updatedAt", "updated_at", ]; -const SUB2API_ACCOUNT_IMPORT_KEYS: &[&str] = &[ - "id", - "email", - "accountId", - "account_id", - "userId", - "user_id", - "label", - "planType", - "plan_type", - "token", - "accessToken", - "access_token", - "refreshToken", - "refresh_token", - "status", - "addedAt", - "added_at", - "cachedQuota", - "cached_quota", - "quotaFetchedAt", - "quota_fetched_at", - "quotaVerifyRequired", - "quota_verify_required", - "proxyApiKey", - "proxy_api_key", - "usage", -]; /// 解析后的账号导入载荷。 #[derive(Debug, Clone)] @@ -122,74 +98,132 @@ pub struct ParsedAccountImport { /// 解析账号导入载荷。 pub fn parse_account_import_payload(payload: &Value) -> Result { - let payload = payload + let payload = admin_envelope_data(payload)?.unwrap_or(payload); + match explicit_account_import_source(payload)? { + Some(AccountImportSource::Cpr) => parse_cpr_payload(payload), + Some(AccountImportSource::Sub2Api) => parse_sub2api_payload(payload), + Some(AccountImportSource::CliProxyApi) => parse_cli_proxy_api_payload(payload), + None if account_import_payload_looks_sub2api(payload) => parse_sub2api_payload(payload), + None if account_import_payload_looks_cli_proxy_api(payload) => { + parse_cli_proxy_api_payload(payload) + } + None => parse_cpr_payload(payload), + } +} + +fn admin_envelope_data(payload: &Value) -> Result, &'static str> { + let Some(data) = payload .get("data") .filter(|data| data.get("accounts").is_some()) - .map(|data| -> Result<&Value, &'static str> { - ensure_account_import_keys(payload, ACCOUNT_IMPORT_ENVELOPE_KEYS)?; - Ok(data) - }) - .transpose()? - .unwrap_or(payload); + else { + return Ok(None); + }; + ensure_account_import_keys(payload, ACCOUNT_IMPORT_ENVELOPE_KEYS)?; + Ok(Some(data)) +} - if let Some(accounts) = payload.get("accounts") { - ensure_account_import_keys(payload, ACCOUNT_IMPORT_CONTAINER_KEYS)?; - let accounts = accounts.as_array().ok_or("no importable accounts")?; - let source = account_import_source(payload, accounts)?; - return Ok(ParsedAccountImport { - source, - entries: parse_account_import_entries(accounts, source)?, - }); - } - if let Some(accounts) = payload.as_array() { - let source = account_import_source(payload, accounts)?; - return Ok(ParsedAccountImport { - source, - entries: parse_account_import_entries(accounts, source)?, - }); +fn explicit_account_import_source( + value: &Value, +) -> Result, &'static str> { + let Some(source_format) = first_string(value, &["sourceFormat", "source_format"]) else { + return Ok(None); + }; + match source_format.trim().to_ascii_lowercase().as_str() { + "cpr" => Ok(Some(AccountImportSource::Cpr)), + "sub2api" => Ok(Some(AccountImportSource::Sub2Api)), + "cliproxyapi" | "cli_proxy_api" | "cli-proxy-api" | "cpa" => { + Ok(Some(AccountImportSource::CliProxyApi)) + } + _ => Err("no importable accounts"), } +} - let source = account_import_source(payload, std::slice::from_ref(payload))?; +fn parse_cpr_payload(payload: &Value) -> Result { Ok(ParsedAccountImport { - source, - entries: account_import_entry_from_value(payload, source)? - .into_iter() - .collect(), + source: AccountImportSource::Cpr, + entries: parse_cpr_entries(payload)?, }) } -fn parse_account_import_entries( +fn parse_cpr_entries(payload: &Value) -> Result, &'static str> { + if let Some(accounts) = payload.get("accounts") { + ensure_account_import_keys(payload, ACCOUNT_IMPORT_CONTAINER_KEYS)?; + let accounts = accounts.as_array().ok_or("no importable accounts")?; + return parse_entries(accounts, cpr_account_import_entry_from_value); + } + if let Some(accounts) = payload.as_array() { + return parse_entries(accounts, cpr_account_import_entry_from_value); + } + cpr_account_import_entry_from_value(payload).map(option_to_vec) +} + +fn parse_sub2api_payload(payload: &Value) -> Result { + Ok(ParsedAccountImport { + source: AccountImportSource::Sub2Api, + entries: parse_sub2api_entries(payload)?, + }) +} + +fn parse_sub2api_entries(payload: &Value) -> Result, &'static str> { + if let Some(accounts) = payload.get("accounts") { + let accounts = accounts.as_array().ok_or("no importable accounts")?; + return parse_entries(accounts, sub2api_account_import_entry_from_value); + } + if let Some(accounts) = payload.as_array() { + return parse_entries(accounts, sub2api_account_import_entry_from_value); + } + sub2api_account_import_entry_from_value(payload).map(option_to_vec) +} + +fn parse_cli_proxy_api_payload(payload: &Value) -> Result { + Ok(ParsedAccountImport { + source: AccountImportSource::CliProxyApi, + entries: parse_cli_proxy_api_entries(payload)?, + }) +} + +fn parse_cli_proxy_api_entries(payload: &Value) -> Result, &'static str> { + if let Some(accounts) = payload.get("accounts") { + let accounts = accounts.as_array().ok_or("no importable accounts")?; + return parse_entries(accounts, cli_proxy_api_account_import_entry_from_value); + } + if let Some(accounts) = payload.as_array() { + return parse_entries(accounts, cli_proxy_api_account_import_entry_from_value); + } + cli_proxy_api_account_import_entry_from_value(payload).map(option_to_vec) +} + +fn parse_entries( accounts: &[Value], - source: AccountImportSource, + parser: fn(&Value) -> Result, &'static str>, ) -> Result, &'static str> { let mut entries = Vec::new(); for account in accounts { - if let Some(entry) = account_import_entry_from_value(account, source)? { + if let Some(entry) = parser(account)? { entries.push(entry); } } Ok(entries) } -fn account_import_entry_from_value( +fn option_to_vec(value: Option) -> Vec { + value.into_iter().collect() +} + +fn cpr_account_import_entry_from_value( value: &Value, - source: AccountImportSource, ) -> Result, &'static str> { let Some(account) = value.as_object() else { return Ok(None); }; - let allowed_keys = match source { - AccountImportSource::Cpr => ACCOUNT_IMPORT_ACCOUNT_KEYS, - AccountImportSource::Sub2Api => SUB2API_ACCOUNT_IMPORT_KEYS, - }; if account .keys() - .any(|key| !allowed_keys.contains(&key.as_str())) + .any(|key| !ACCOUNT_IMPORT_ACCOUNT_KEYS.contains(&key.as_str())) { return Err("no importable accounts"); } - let token = first_string(value, &["token", "accessToken", "access_token"]); + let token = first_string(value, &["token", "at", "accessToken", "access_token"]); let refresh_token = first_string(value, &["refreshToken", "refresh_token"]); if token.is_none() && refresh_token.is_none() { return Ok(None); @@ -209,15 +243,247 @@ fn account_import_entry_from_value( &["accessTokenExpiresAt", "access_token_expires_at"], ), status: first_string(value, &["status"]), - cached_quota: (source == AccountImportSource::Sub2Api) - .then(|| first_value(value, &["cachedQuota", "cached_quota"])) - .flatten(), - quota_fetched_at: (source == AccountImportSource::Sub2Api) - .then(|| first_string(value, &["quotaFetchedAt", "quota_fetched_at"])) - .flatten(), - quota_verify_required: (source == AccountImportSource::Sub2Api) - .then(|| first_bool(value, &["quotaVerifyRequired", "quota_verify_required"])) - .flatten(), + cached_quota: None, + quota_fetched_at: None, + quota_verify_required: None, + })) +} + +fn sub2api_account_import_entry_from_value( + value: &Value, +) -> Result, &'static str> { + if sub2api_account_backup_entry(value) { + return Ok(sub2api_backup_account_entry(value)); + } + Ok(sub2api_codex_session_or_flat_entry(value)) +} + +fn sub2api_backup_account_entry(value: &Value) -> Option { + let account = value.as_object()?; + if !optional_string_field_matches(account, "platform", "openai") + || !optional_string_field_matches(account, "type", "oauth") + { + return None; + } + let credentials = value.get("credentials")?; + let token = first_path_string( + credentials, + [ + &["access_token"], + &["accessToken"], + &["at"], + &["token", "access_token"], + &["token", "accessToken"], + &["token", "at"], + ], + ); + let refresh_token = first_path_string( + credentials, + [ + &["refresh_token"], + &["refreshToken"], + &["rt"], + &["token", "refresh_token"], + &["token", "refreshToken"], + &["token", "rt"], + ], + ); + if token.is_none() && refresh_token.is_none() { + return None; + } + + Some(AccountImportEntry { + id: first_string(value, &["id"]), + email: first_path_string(credentials, [&["email"], &["user", "email"]]), + account_id: first_path_string( + credentials, + [ + &["chatgpt_account_id"], + &["chatgptAccountId"], + &["account_id"], + &["accountId"], + &["account", "id"], + &["account", "account_id"], + &["account", "chatgpt_account_id"], + ], + ), + user_id: first_path_string( + credentials, + [ + &["chatgpt_user_id"], + &["chatgptUserId"], + &["user_id"], + &["userId"], + &["user", "id"], + ], + ), + label: first_string(value, &["label", "name"]), + plan_type: first_path_string(credentials, [&["plan_type"], &["planType"]]), + token, + refresh_token, + access_token_expires_at: first_datetime_string( + credentials, + [&["expires_at"], &["expiresAt"], &["expired"], &["expire"]], + ), + status: first_string(value, &["status"]), + cached_quota: first_value(value, &["cachedQuota", "cached_quota"]), + quota_fetched_at: first_string(value, &["quotaFetchedAt", "quota_fetched_at"]), + quota_verify_required: first_bool(value, &["quotaVerifyRequired", "quota_verify_required"]), + }) +} + +fn sub2api_codex_session_or_flat_entry(value: &Value) -> Option { + let _account = value.as_object()?; + let token = first_path_string( + value, + [ + &["tokens", "access_token"], + &["tokens", "accessToken"], + &["tokens", "at"], + &["access_token"], + &["accessToken"], + &["token"], + &["at"], + ], + ); + let refresh_token = first_path_string( + value, + [ + &["tokens", "refresh_token"], + &["tokens", "refreshToken"], + &["tokens", "rt"], + &["refresh_token"], + &["refreshToken"], + &["rt"], + ], + ); + if token.is_none() && refresh_token.is_none() { + return None; + } + + Some(AccountImportEntry { + id: first_string(value, &["id"]), + email: first_path_string(value, [&["email"], &["user", "email"]]), + account_id: first_path_string( + value, + [ + &["chatgpt_account_id"], + &["chatgptAccountId"], + &["account_id"], + &["accountId"], + &["account", "id"], + &["account", "account_id"], + &["account", "chatgpt_account_id"], + ], + ), + user_id: first_path_string( + value, + [ + &["chatgpt_user_id"], + &["chatgptUserId"], + &["user_id"], + &["userId"], + &["user", "id"], + ], + ), + label: first_path_string(value, [&["label"], &["name"], &["user", "name"]]), + plan_type: first_path_string( + value, + [ + &["plan_type"], + &["planType"], + &["account", "plan_type"], + &["account", "planType"], + ], + ), + token, + refresh_token, + access_token_expires_at: first_datetime_string( + value, + [ + &["tokens", "expires_at"], + &["tokens", "expiresAt"], + &["expires_at"], + &["expiresAt"], + ], + ), + status: first_string(value, &["status"]), + cached_quota: first_value(value, &["cachedQuota", "cached_quota"]), + quota_fetched_at: first_string(value, &["quotaFetchedAt", "quota_fetched_at"]), + quota_verify_required: first_bool(value, &["quotaVerifyRequired", "quota_verify_required"]), + }) +} + +fn cli_proxy_api_account_import_entry_from_value( + value: &Value, +) -> Result, &'static str> { + let Some(account) = value.as_object() else { + return Ok(None); + }; + if !cli_proxy_api_provider_is_codex(account) { + return Ok(None); + } + + let token = first_path_string( + value, + [ + &["access_token"], + &["accessToken"], + &["at"], + &["token", "access_token"], + &["token", "accessToken"], + &["token", "at"], + &["token"], + ], + ); + let refresh_token = first_path_string( + value, + [ + &["refresh_token"], + &["refreshToken"], + &["rt"], + &["token", "refresh_token"], + &["token", "refreshToken"], + &["token", "rt"], + ], + ); + if token.is_none() && refresh_token.is_none() { + return Ok(None); + } + + Ok(Some(AccountImportEntry { + id: None, + email: first_string(value, &["email"]), + account_id: first_string( + value, + &["chatgpt_account_id", "chatgptAccountId", "account_id"], + ), + user_id: first_string( + value, + &["chatgpt_user_id", "chatgptUserId", "user_id", "userId"], + ), + label: first_string(value, &["label", "name"]), + plan_type: first_path_string( + value, + [&["plan_type"], &["planType"], &["attributes", "plan_type"]], + ), + token, + refresh_token, + access_token_expires_at: first_datetime_string( + value, + [ + &["expired"], + &["expire"], + &["expires_at"], + &["expiresAt"], + &["expiry"], + &["expires"], + ], + ), + status: cli_proxy_api_status(account), + cached_quota: None, + quota_fetched_at: None, + quota_verify_required: None, })) } @@ -235,42 +501,87 @@ fn ensure_account_import_keys(value: &Value, allowed_keys: &[&str]) -> Result<() } } -fn account_import_source( - value: &Value, - accounts: &[Value], -) -> Result { - if let Some(source_format) = first_string(value, &["sourceFormat", "source_format"]) { - return match source_format.trim().to_ascii_lowercase().as_str() { - "cpr" => Ok(AccountImportSource::Cpr), - "sub2api" => Ok(AccountImportSource::Sub2Api), - _ => Err("no importable accounts"), - }; +fn account_import_payload_looks_sub2api(value: &Value) -> bool { + if value.get("proxies").is_some() || value.get("exported_at").is_some() { + return true; } - - if accounts.iter().any(account_import_entry_looks_sub2api) { - Ok(AccountImportSource::Sub2Api) - } else { - Ok(AccountImportSource::Cpr) + if let Some(accounts) = value.get("accounts").and_then(Value::as_array) { + return accounts.iter().any(sub2api_account_looks_known); } + if let Some(accounts) = value.as_array() { + return accounts.iter().any(sub2api_account_looks_known); + } + sub2api_account_looks_known(value) } -fn account_import_entry_looks_sub2api(value: &Value) -> bool { +fn sub2api_account_looks_known(value: &Value) -> bool { let Some(account) = value.as_object() else { return false; }; - [ - "proxyApiKey", - "proxy_api_key", - "usage", - "cachedQuota", - "cached_quota", - "quotaFetchedAt", - "quota_fetched_at", - "quotaVerifyRequired", - "quota_verify_required", - ] - .iter() - .any(|key| account.contains_key(*key)) + sub2api_account_backup_entry(value) + || value.get("tokens").is_some() + || [ + "proxyApiKey", + "proxy_api_key", + "usage", + "cachedQuota", + "cached_quota", + "quotaFetchedAt", + "quota_fetched_at", + "quotaVerifyRequired", + "quota_verify_required", + ] + .iter() + .any(|key| account.contains_key(*key)) +} + +fn sub2api_account_backup_entry(value: &Value) -> bool { + value.get("credentials").is_some() + && (value.get("platform").is_some() || value.get("type").is_some()) +} + +fn account_import_payload_looks_cli_proxy_api(value: &Value) -> bool { + if let Some(accounts) = value.get("accounts").and_then(Value::as_array) { + return accounts.iter().any(cli_proxy_api_account_looks_known); + } + if let Some(accounts) = value.as_array() { + return accounts.iter().any(cli_proxy_api_account_looks_known); + } + cli_proxy_api_account_looks_known(value) +} + +fn cli_proxy_api_account_looks_known(value: &Value) -> bool { + value + .as_object() + .is_some_and(cli_proxy_api_provider_is_codex) +} + +fn cli_proxy_api_provider_is_codex(account: &Map) -> bool { + let provider = account + .get("type") + .or_else(|| account.get("provider")) + .and_then(Value::as_str) + .map(str::trim) + .unwrap_or_default() + .to_ascii_lowercase(); + matches!(provider.as_str(), "codex" | "openai") +} + +fn cli_proxy_api_status(account: &Map) -> Option { + account + .get("disabled") + .and_then(Value::as_bool) + .filter(|disabled| *disabled) + .map(|_| "disabled".to_string()) +} + +fn optional_string_field_matches(account: &Map, key: &str, expected: &str) -> bool { + account + .get(key) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .is_none_or(|value| value.eq_ignore_ascii_case(expected)) } /// 解析导入的状态字符串。 @@ -360,6 +671,22 @@ fn first_string(value: &Value, keys: &[&str]) -> Option { .map(ToString::to_string) } +fn first_path_string(value: &Value, paths: [&[&str]; N]) -> Option { + paths + .iter() + .find_map(|path| value_at_path(value, path).and_then(Value::as_str)) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string) +} + +fn first_datetime_string(value: &Value, paths: [&[&str]; N]) -> Option { + paths + .iter() + .find_map(|path| value_at_path(value, path)) + .and_then(datetime_value_to_rfc3339) +} + fn first_bool(value: &Value, keys: &[&str]) -> Option { keys.iter() .find_map(|key| value.get(key).and_then(Value::as_bool)) @@ -370,3 +697,29 @@ fn first_value(value: &Value, keys: &[&str]) -> Option { .find_map(|key| value.get(key).filter(|value| !value.is_null())) .cloned() } + +fn value_at_path<'a>(value: &'a Value, path: &[&str]) -> Option<&'a Value> { + let mut current = value; + for key in path { + current = current.get(*key)?; + } + Some(current) +} + +fn datetime_value_to_rfc3339(value: &Value) -> Option { + if let Some(value) = value.as_str() { + return normalize_nonempty_str(Some(value)).map(ToString::to_string); + } + if let Some(value) = value.as_i64() { + return DateTime::::from_timestamp(value, 0).map(|time| time.to_rfc3339()); + } + if let Some(value) = value.as_u64().and_then(|value| i64::try_from(value).ok()) { + return DateTime::::from_timestamp(value, 0).map(|time| time.to_rfc3339()); + } + value + .as_f64() + .filter(|value| value.is_finite()) + .map(|value| value.trunc() as i64) + .and_then(|value| DateTime::::from_timestamp(value, 0)) + .map(|time| time.to_rfc3339()) +} diff --git a/backend/src/upstream/accounts/token_refresh/runtime.rs b/backend/src/upstream/accounts/token_refresh/runtime.rs index 50f5c419b..11596cb5e 100644 --- a/backend/src/upstream/accounts/token_refresh/runtime.rs +++ b/backend/src/upstream/accounts/token_refresh/runtime.rs @@ -151,12 +151,15 @@ where let mut expired_recovery_index = 0; for account in accounts { - if account.refresh_token.is_none() - || matches!( - account.status, - AccountStatus::Disabled | AccountStatus::Banned - ) - { + if self.skip_account_without_refresh_token(&account).await? { + summary.skipped += 1; + continue; + } + + if matches!( + account.status, + AccountStatus::Disabled | AccountStatus::Banned + ) { summary.skipped += 1; continue; } @@ -240,12 +243,18 @@ where account_id.to_string(), )); }; - if account.refresh_token.is_none() - || matches!( - account.status, - AccountStatus::Disabled | AccountStatus::Banned - ) - { + if account.refresh_token.is_none() { + self.clear_scheduled_timer(account_id).await; + if account.next_refresh_at.is_some() { + persist_next_refresh_at(&self.store, account_id, None).await?; + } + return Ok(()); + } + + if matches!( + account.status, + AccountStatus::Disabled | AccountStatus::Banned + ) { return Ok(()); } @@ -272,6 +281,11 @@ where }; for account in accounts { + if self.skip_account_without_refresh_token(&account).await? { + summary.skipped += 1; + continue; + } + if account.next_refresh_at.is_some_and(|value| value > now) { summary.skipped += 1; continue; @@ -336,6 +350,22 @@ where (refresh_at - now).to_std().ok() } + async fn skip_account_without_refresh_token( + &self, + account: &Account, + ) -> TokenRefreshServiceResult { + if account.refresh_token.is_some() { + return Ok(false); + } + + self.clear_scheduled_timer(&account.id).await; + if account.next_refresh_at.is_some() { + self.persist_next_refresh_at_if_changed(account, None) + .await?; + } + Ok(true) + } + fn computed_refresh_at(&self, account: &Account) -> Option> { let expires_at = account .access_token_expires_at @@ -440,6 +470,10 @@ where )); }; let account = stored_account_to_refresh_account(account); + if self.skip_account_without_refresh_token(&account).await? { + return Ok(TokenRefreshOutcome::Skipped); + } + if account.next_refresh_at.is_some_and(|value| value > now) { return Ok(TokenRefreshOutcome::Skipped); } diff --git a/backend/tests/admin/accounts/exporting.rs b/backend/tests/admin/accounts/exporting.rs index 488eb9aa8..b3c2b8f73 100644 --- a/backend/tests/admin/accounts/exporting.rs +++ b/backend/tests/admin/accounts/exporting.rs @@ -1,5 +1,4 @@ use super::*; -use sqlx::Row; const EXPORT_CONFIRM: &str = "confirm=export_sensitive_accounts"; @@ -295,28 +294,22 @@ async fn admin_accounts_export_should_require_explicit_sensitive_export_confirma .unwrap(); let status = response.status(); let body = response_json(response).await; - let audit = account_export_audit_row(&pool, "req_export_missing_confirmation").await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!( body["message"], "account export requires confirm=export_sensitive_accounts" ); - assert_eq!(audit.level, "warn"); - assert_eq!(audit.status_code, Some(400)); assert_eq!( - audit.failure_class, - Some("missing_confirmation".to_string()) + usage_record_count_by_request_id(&pool, "req_export_missing_confirmation").await, + 0 ); - assert_eq!(audit.metadata["accountIds"][0], "acct_export_confirm"); - assert!(audit.metadata.get("token").is_none()); - assert!(audit.metadata.get("refreshToken").is_none()); } #[tokio::test] -async fn admin_accounts_export_should_record_success_audit_event() { +async fn admin_accounts_export_should_not_write_usage_record_on_success() { let (app, _state, pool, _dir) = - admin_accounts_test_app("admin-accounts-export-audit.sqlite", 137).await; + admin_accounts_test_app("admin-accounts-export-usage.sqlite", 137).await; seed_account( &pool, NewAccount { @@ -358,48 +351,17 @@ async fn admin_accounts_export_should_record_success_audit_event() { .await .unwrap(); assert_eq!(response.status(), StatusCode::OK); - let audit = account_export_audit_row(&pool, "req_export_success_audit").await; - assert_eq!(audit.kind, "admin_account_export"); - assert_eq!(audit.level, "warn"); - assert_eq!(audit.status_code, Some(200)); - assert_eq!(audit.failure_class, None); - assert_eq!(audit.message, "Admin account export succeeded"); - assert_eq!(audit.metadata["accountCount"], 1); - assert_eq!(audit.metadata["accountIds"][0], "acct_export_audit"); - assert!(audit.metadata.get("token").is_none()); - assert!(audit.metadata.get("refreshToken").is_none()); + assert_eq!( + usage_record_count_by_request_id(&pool, "req_export_success_audit").await, + 0 + ); } -#[derive(Debug)] -struct AccountExportAuditRow { - kind: String, - level: String, - status_code: Option, - failure_class: Option, - message: String, - metadata: Value, -} - -async fn account_export_audit_row(pool: &SqlitePool, request_id: &str) -> AccountExportAuditRow { - let row = sqlx::query( - "select kind, level, status_code, failure_class, message, metadata_json \ - from usage_records \ - where kind = 'admin_account_export' and request_id = ? \ - order by created_at desc, id desc \ - limit 1", - ) - .bind(request_id) - .fetch_one(pool) - .await - .unwrap(); - - AccountExportAuditRow { - kind: row.get("kind"), - level: row.get("level"), - status_code: row.get("status_code"), - failure_class: row.get("failure_class"), - message: row.get("message"), - metadata: serde_json::from_str(&row.get::("metadata_json")).unwrap(), - } +async fn usage_record_count_by_request_id(pool: &SqlitePool, request_id: &str) -> i64 { + sqlx::query_scalar("select count(*) from usage_records where request_id = ?") + .bind(request_id) + .fetch_one(pool) + .await + .unwrap() } diff --git a/backend/tests/admin/accounts/importing.rs b/backend/tests/admin/accounts/importing.rs index d17a97554..f38e2d272 100644 --- a/backend/tests/admin/accounts/importing.rs +++ b/backend/tests/admin/accounts/importing.rs @@ -81,6 +81,256 @@ async fn admin_accounts_import_should_store_cpr_account_tokens() { assert_eq!(stored.status, AccountStatus::Disabled); } +#[tokio::test] +async fn admin_accounts_import_should_store_cpr_access_token_from_at_field() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "account_id": "cpr-at-account", + "user_id": "cpr-at-user", + "email": "cpr-at@example.com", + "plan_type": "k12", + "rate_limit": { + "allowed": true, + "limit_reached": false, + "primary_window": { + "used_percent": 9, + "reset_at": 1_800_000_000, + "limit_window_seconds": 18_000 + } + } + }))) + .mount(&server) + .await; + let (app, _state, pool, _dir) = admin_accounts_test_app_with_api_base_url( + "admin-accounts-import-cpr-at.sqlite", + 129, + format!("{}/backend-api", server.uri()), + ) + .await; + let access_token = unsigned_jwt(&json!({ + "exp": 4_102_444_800i64, + "https://api.openai.com/auth": {}, + "https://api.openai.com/profile": {}, + })); + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/admin/accounts/import") + .header("content-type", "application/json") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_accounts_import_cpr_at") + .body(Body::from( + json!({ + "accounts": [{ + "id": "acct_import_cpr_at", + "at": access_token.clone(), + "status": "active" + }] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = response_json(response).await; + let stored = SqliteAccountStore::new(pool) + .get("acct_import_cpr_at") + .await + .unwrap() + .unwrap(); + + assert_eq!(status, StatusCode::OK); + assert_eq!(body["data"]["imported"], 1); + assert_eq!(body["data"]["sourceFormat"], "cpr"); + assert_eq!(stored.account_id.as_deref(), Some("cpr-at-account")); + assert_eq!(stored.user_id.as_deref(), Some("cpr-at-user")); + assert_eq!(stored.email.as_deref(), Some("cpr-at@example.com")); + assert_eq!(stored.plan_type.as_deref(), Some("k12")); + assert_eq!(stored.access_token.expose_secret(), access_token); + assert!(stored.refresh_token.is_none()); + assert!(stored.next_refresh_at.is_none()); +} + +#[tokio::test] +async fn admin_accounts_import_should_read_sub2api_backup_access_token_only_accounts() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "account_id": "sub2api-at-account", + "user_id": "sub2api-at-user", + "email": "sub2api-at@example.com", + "plan_type": "k12", + "rate_limit": { + "allowed": true, + "limit_reached": false, + "primary_window": { + "used_percent": 7, + "reset_at": 1_800_000_000, + "limit_window_seconds": 18_000 + } + } + }))) + .mount(&server) + .await; + let (app, _state, pool, _dir) = admin_accounts_test_app_with_api_base_url( + "admin-accounts-import-sub2api-backup-at.sqlite", + 127, + format!("{}/backend-api", server.uri()), + ) + .await; + let access_token = unsigned_jwt(&json!({ + "exp": 4_102_444_800i64, + "https://api.openai.com/auth": {}, + "https://api.openai.com/profile": {}, + })); + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/admin/accounts/import") + .header("content-type", "application/json") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_accounts_import_sub2api_backup_at") + .body(Body::from( + json!({ + "sourceFormat": "sub2api", + "exported_at": "2026-07-03T15:46:38.717Z", + "proxies": [], + "accounts": [{ + "name": "sub2api-at@example.com", + "platform": "openai", + "type": "oauth", + "credentials": { + "at": access_token.clone(), + "refresh_token": "", + "expires_at": 4_102_444_800i64 + }, + "extra": {}, + "concurrency": 3, + "priority": 50, + "rate_multiplier": null, + "auto_pause_on_expired": true + }] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = response_json(response).await; + let store = SqliteAccountStore::new(pool); + let stored_metadata = store + .list_metadata_page(1, 10, None) + .await + .unwrap() + .items + .into_iter() + .next() + .unwrap(); + let stored = store.get(&stored_metadata.id).await.unwrap().unwrap(); + + assert_eq!(status, StatusCode::OK); + assert_eq!(body["data"]["imported"], 1); + assert_eq!(body["data"]["sourceFormat"], "sub2api"); + assert_eq!(stored.account_id.as_deref(), Some("sub2api-at-account")); + assert_eq!(stored.user_id.as_deref(), Some("sub2api-at-user")); + assert_eq!(stored.email.as_deref(), Some("sub2api-at@example.com")); + assert_eq!(stored.plan_type.as_deref(), Some("k12")); + assert_eq!(stored.access_token.expose_secret(), access_token); + assert!(stored.refresh_token.is_none()); + assert!(stored.next_refresh_at.is_none()); +} + +#[tokio::test] +async fn admin_accounts_import_should_read_cliproxyapi_codex_auth_files() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "account_id": "cliproxyapi-account", + "user_id": "cliproxyapi-user", + "email": "cliproxyapi@example.com", + "plan_type": "plus", + "rate_limit": { + "allowed": true, + "limit_reached": false, + "primary_window": { + "used_percent": 3, + "reset_at": 1_800_000_000, + "limit_window_seconds": 18_000 + } + } + }))) + .mount(&server) + .await; + let (app, _state, pool, _dir) = admin_accounts_test_app_with_api_base_url( + "admin-accounts-import-cliproxyapi.sqlite", + 128, + format!("{}/backend-api", server.uri()), + ) + .await; + let access_token = unsigned_jwt(&json!({ + "exp": 4_102_444_800i64, + "https://api.openai.com/auth": {}, + "https://api.openai.com/profile": {}, + })); + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/admin/accounts/import") + .header("content-type", "application/json") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_accounts_import_cliproxyapi") + .body(Body::from( + json!({ + "sourceFormat": "cliproxyapi", + "accounts": [{ + "type": "codex", + "access_token": access_token.clone(), + "refresh_token": "cliproxyapi-refresh", + "expired": "2100-01-01T00:00:00Z", + "email": "cliproxyapi@example.com", + "label": "CLIProxyAPI" + }] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = response_json(response).await; + let stored = SqliteAccountStore::new(pool) + .list_metadata_page(1, 10, None) + .await + .unwrap() + .items + .into_iter() + .next() + .unwrap(); + + assert_eq!(status, StatusCode::OK); + assert_eq!(body["data"]["imported"], 1); + assert_eq!(body["data"]["sourceFormat"], "cliproxyapi"); + assert_eq!(stored.account_id.as_deref(), Some("cliproxyapi-account")); + assert_eq!(stored.user_id.as_deref(), Some("cliproxyapi-user")); + assert_eq!(stored.email.as_deref(), Some("cliproxyapi@example.com")); + assert_eq!(stored.plan_type.as_deref(), Some("plus")); +} + #[tokio::test] async fn admin_accounts_import_should_fetch_wham_usage_for_current_openai_token_identity() { const CURRENT_IMPORT_ACCOUNT_ID: &str = "acct_import_current_token"; @@ -346,6 +596,7 @@ async fn admin_accounts_import_from_refresh_token_should_not_store_consumed_refr ); let account = store.get(&stored.id).await.unwrap().unwrap(); assert!(account.refresh_token.is_none()); + assert!(account.next_refresh_at.is_none()); } #[tokio::test] diff --git a/backend/tests/admin/accounts/lifecycle.rs b/backend/tests/admin/accounts/lifecycle.rs index 7ec978a03..6fcdc5570 100644 --- a/backend/tests/admin/accounts/lifecycle.rs +++ b/backend/tests/admin/accounts/lifecycle.rs @@ -866,6 +866,7 @@ async fn admin_account_manual_create_from_refresh_token_should_not_store_consume assert_eq!(status, StatusCode::OK); assert!(stored.refresh_token.is_none()); + assert!(stored.next_refresh_at.is_none()); } #[tokio::test] diff --git a/backend/tests/admin/accounts/testing.rs b/backend/tests/admin/accounts/testing.rs index d12116d3b..d34296d64 100644 --- a/backend/tests/admin/accounts/testing.rs +++ b/backend/tests/admin/accounts/testing.rs @@ -2,7 +2,7 @@ use super::*; use axum::body::to_bytes; use codex_proxy_rs::upstream::protocol::sse::{encode_sse_event, parse_sse_events}; use wiremock::{ - matchers::{method, path}, + matchers::{header, method, path}, Mock, MockServer, ResponseTemplate, }; @@ -75,6 +75,146 @@ async fn account_models_should_return_database_snapshot_models() { assert_eq!(body["data"]["models"][1]["label"], "GPT 5.4"); } +#[tokio::test] +async fn account_models_should_fetch_missing_plan_snapshot_with_account_token() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/codex/models")) + .and(header("authorization", "Bearer access-plan-b")) + .and(header("chatgpt-account-id", "chatgpt-plan-b")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "models": [ + { + "slug": "gpt-plan-b-live", + "display_name": "GPT Plan B Live", + "description": "Live model", + "is_default": true, + "supported_reasoning_efforts": [{"reasoning_effort": "medium", "description": "medium"}], + "default_reasoning_effort": "medium", + "input_modalities": ["text"], + "output_modalities": ["text"], + "supports_personality": false + } + ] + }))) + .mount(&server) + .await; + let (app, state, pool, _dir) = admin_accounts_test_app_with_api_base_url( + "admin-account-models-unfetched-plan.sqlite", + 93, + format!("{}/backend-api", server.uri()), + ) + .await; + seed_account( + &pool, + NewAccount { + id: "acct_plan_b".to_string(), + email: Some("plan-b@example.com".to_string()), + account_id: Some("chatgpt-plan-b".to_string()), + user_id: None, + label: None, + plan_type: Some("plan-b".to_string()), + access_token: SecretString::new("access-plan-b".to_string().into()), + refresh_token: None, + access_token_expires_at: None, + status: AccountStatus::Active, + added_at: None, + }, + ) + .await; + seed_model_snapshot(&pool, "plus").await; + state + .services + .models + .reload_from_store() + .await + .expect("model catalog should reload from seeded snapshot"); + + let response = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/accounts/models?id=acct_plan_b") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_account_models_unfetched_plan") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let requests = server.received_requests().await.unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(requests.len(), 1, "missing plan should be fetched upstream"); + let body = response_json(response).await; + assert_eq!(body["data"]["models"][0]["id"], "gpt-plan-b-live"); + assert_eq!(body["data"]["models"][0]["label"], "GPT Plan B Live"); + let stored_plan_b: i64 = + sqlx::query_scalar("select count(*) from model_plan_snapshots where plan_type = ?") + .bind("plan-b") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(stored_plan_b, 1); +} + +#[tokio::test] +async fn account_models_should_return_models_for_account_plan_snapshot() { + let server = MockServer::start().await; + let (app, state, pool, _dir) = admin_accounts_test_app_with_api_base_url( + "admin-account-models-matching-plan.sqlite", + 94, + format!("{}/backend-api", server.uri()), + ) + .await; + seed_account( + &pool, + NewAccount { + id: "acct_plan_b_snapshot".to_string(), + email: Some("plan-b-snapshot@example.com".to_string()), + account_id: Some("chatgpt-plan-b-snapshot".to_string()), + user_id: None, + label: None, + plan_type: Some("plan-b".to_string()), + access_token: SecretString::new("access-plan-b-snapshot".to_string().into()), + refresh_token: None, + access_token_expires_at: None, + status: AccountStatus::Active, + added_at: None, + }, + ) + .await; + seed_model_snapshot(&pool, "plan-a").await; + seed_single_model_snapshot(&pool, "plan-b", "gpt-plan-b-only", "GPT Plan B Only").await; + state + .services + .models + .reload_from_store() + .await + .expect("model catalog should reload from seeded snapshots"); + + let response = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/accounts/models?id=acct_plan_b_snapshot") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_account_models_matching_plan") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let requests = server.received_requests().await.unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + assert!(requests.is_empty(), "model list should be loaded from DB"); + let body = response_json(response).await; + assert_eq!(body["data"]["models"].as_array().unwrap().len(), 1); + assert_eq!(body["data"]["models"][0]["id"], "gpt-plan-b-only"); + assert_eq!(body["data"]["models"][0]["label"], "GPT Plan B Only"); +} + async fn seed_model_snapshot(pool: &SqlitePool, plan_type: &str) { let models_json = json!([ { @@ -116,6 +256,39 @@ async fn seed_model_snapshot(pool: &SqlitePool, plan_type: &str) { .unwrap(); } +async fn seed_single_model_snapshot( + pool: &SqlitePool, + plan_type: &str, + model_id: &str, + display_name: &str, +) { + let models_json = json!([ + { + "id": model_id, + "displayName": display_name, + "description": "Test model", + "isDefault": false, + "supportedReasoningEfforts": [{"reasoningEffort": "medium", "description": "medium"}], + "defaultReasoningEffort": "medium", + "inputModalities": ["text"], + "outputModalities": ["text"], + "supportsPersonality": false, + "upgrade": null, + "source": "test" + } + ]) + .to_string(); + sqlx::query( + "insert or replace into model_plan_snapshots (plan_type, models_json, fetched_at) values (?, ?, ?)", + ) + .bind(plan_type) + .bind(models_json) + .bind("2026-06-29T00:00:00Z") + .execute(pool) + .await + .unwrap(); +} + #[tokio::test] async fn account_test_stream_should_translate_upstream_responses_sse() { let server = MockServer::start().await; diff --git a/backend/tests/admin/auth/session.rs b/backend/tests/admin/auth/session.rs index 17adc3190..3a2155639 100644 --- a/backend/tests/admin/auth/session.rs +++ b/backend/tests/admin/auth/session.rs @@ -4,8 +4,6 @@ use axum::{ }; use chrono::Utc; use codex_proxy_rs::infra::{database::connect_sqlite, identity::hash_admin_password}; -use serde_json::Value; -use sqlx::Row; use sqlx::SqlitePool; use tower::util::ServiceExt; @@ -106,29 +104,23 @@ async fn admin_login_should_throttle_repeated_failures_from_same_source() { } #[tokio::test] -async fn admin_login_should_record_success_audit_event() { +async fn admin_login_should_not_write_usage_record_on_success() { let (app, _dir, pool) = - admin_login_test_app("admin-login-success-audit.sqlite", "correct-password").await; + admin_login_test_app("admin-login-success-usage.sqlite", "correct-password").await; - let response = post_login_with_request_id(&app, "correct-password", "req_login_audit_ok").await; + let response = post_login_with_request_id(&app, "correct-password", "req_login_usage_ok").await; assert_eq!(response.status(), StatusCode::OK); - let audit = admin_login_audit_row(&pool, "req_login_audit_ok").await; - assert_eq!(audit.kind, "admin_auth"); - assert_eq!(audit.level, "info"); - assert_eq!(audit.route, Some("/api/admin/login".to_string())); - assert_eq!(audit.status_code, Some(200)); - assert_eq!(audit.failure_class, None); - assert_eq!(audit.message, "Admin login succeeded"); - assert_eq!(audit.metadata["source"], "unknown"); - assert_eq!(audit.metadata["usernameProvided"], false); - assert!(audit.metadata.get("username").is_none()); + assert_eq!( + usage_record_count_by_request_id(&pool, "req_login_usage_ok").await, + 0 + ); } #[tokio::test] -async fn admin_login_should_record_invalid_credentials_audit_event() { +async fn admin_login_should_not_write_usage_record_on_invalid_credentials() { let (app, _dir, pool) = - admin_login_test_app("admin-login-invalid-audit.sqlite", "correct-password").await; + admin_login_test_app("admin-login-invalid-usage.sqlite", "correct-password").await; let response = app .clone() @@ -137,7 +129,7 @@ async fn admin_login_should_record_invalid_credentials_audit_event() { .method("POST") .uri("/api/admin/login") .header("content-type", "application/json") - .header("x-request-id", "req_login_audit_bad") + .header("x-request-id", "req_login_usage_bad") .body(Body::from( r#"{"username":"admin","password":"wrong-password"}"#, )) @@ -146,36 +138,32 @@ async fn admin_login_should_record_invalid_credentials_audit_event() { .await .unwrap(); assert_eq!(response.status(), StatusCode::UNAUTHORIZED); - let audit = admin_login_audit_row(&pool, "req_login_audit_bad").await; - assert_eq!(audit.level, "warn"); - assert_eq!(audit.status_code, Some(401)); - assert_eq!(audit.failure_class, Some("invalid_credentials".to_string())); - assert_eq!(audit.message, "Admin login failed"); - assert_eq!(audit.metadata["usernameProvided"], true); - assert!(audit.metadata.get("username").is_none()); + assert_eq!( + usage_record_count_by_request_id(&pool, "req_login_usage_bad").await, + 0 + ); } #[tokio::test] -async fn admin_login_should_record_throttled_audit_event() { +async fn admin_login_should_not_write_usage_record_when_throttled() { let (app, _dir, pool) = - admin_login_test_app("admin-login-throttled-audit.sqlite", "correct-password").await; + admin_login_test_app("admin-login-throttled-usage.sqlite", "correct-password").await; for index in 0..5 { let response = - post_login_with_request_id(&app, "wrong-password", &format!("req_login_audit_{index}")) + post_login_with_request_id(&app, "wrong-password", &format!("req_login_usage_{index}")) .await; assert_eq!(response.status(), StatusCode::UNAUTHORIZED); } let response = - post_login_with_request_id(&app, "correct-password", "req_login_audit_throttled").await; + post_login_with_request_id(&app, "correct-password", "req_login_usage_throttled").await; assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); - let audit = admin_login_audit_row(&pool, "req_login_audit_throttled").await; - assert_eq!(audit.level, "warn"); - assert_eq!(audit.status_code, Some(429)); - assert_eq!(audit.failure_class, Some("login_throttled".to_string())); - assert_eq!(audit.message, "Admin login throttled"); + assert_eq!( + usage_record_count_by_request_id(&pool, "req_login_usage_throttled").await, + 0 + ); } async fn admin_login_test_app( @@ -250,38 +238,12 @@ async fn post_login_with_request_id( .unwrap() } -#[derive(Debug)] -struct AdminLoginAuditRow { - kind: String, - level: String, - route: Option, - status_code: Option, - failure_class: Option, - message: String, - metadata: Value, -} - -async fn admin_login_audit_row(pool: &SqlitePool, request_id: &str) -> AdminLoginAuditRow { - let row = sqlx::query( - "select kind, level, route, status_code, failure_class, message, metadata_json \ - from usage_records \ - where kind = 'admin_auth' and request_id = ? \ - order by created_at desc, id desc \ - limit 1", - ) - .bind(request_id) - .fetch_one(pool) - .await - .unwrap(); - AdminLoginAuditRow { - kind: row.get("kind"), - level: row.get("level"), - route: row.get("route"), - status_code: row.get("status_code"), - failure_class: row.get("failure_class"), - message: row.get("message"), - metadata: serde_json::from_str(&row.get::("metadata_json")).unwrap(), - } +async fn usage_record_count_by_request_id(pool: &SqlitePool, request_id: &str) -> i64 { + sqlx::query_scalar("select count(*) from usage_records where request_id = ?") + .bind(request_id) + .fetch_one(pool) + .await + .unwrap() } async fn seed_admin_user(pool: &SqlitePool, password: &str) { diff --git a/backend/tests/upstream/accounts/token_refresh/scheduling.rs b/backend/tests/upstream/accounts/token_refresh/scheduling.rs index 6097ffdf4..b19f247b8 100644 --- a/backend/tests/upstream/accounts/token_refresh/scheduling.rs +++ b/backend/tests/upstream/accounts/token_refresh/scheduling.rs @@ -341,6 +341,88 @@ async fn token_refresh_task_should_skip_unauthorized_refresh_without_refresh_tok ); } +#[tokio::test] +async fn token_refresh_task_should_clear_stale_refresh_time_without_refresh_token() { + let dir = tempfile::tempdir().expect("temp dir"); + let db = dir.path().join("token-refresh-no-rt-clear-next.sqlite"); + let pool = connect_sqlite(&format!("sqlite://{}", db.display())) + .await + .expect("sqlite pool"); + let store = SqliteAccountStore::new(pool); + let now = Utc::now(); + let old_access_token = test_jwt((now + Duration::minutes(10)).timestamp()); + store + .insert(NewAccount { + id: "acct-refresh-no-rt-clear-next".to_string(), + email: Some("no-rt-clear-next@example.com".to_string()), + account_id: Some("chatgpt-no-rt-clear-next".to_string()), + user_id: Some("user-no-rt-clear-next".to_string()), + label: None, + plan_type: Some("plus".to_string()), + access_token: SecretString::new(old_access_token.clone().into()), + refresh_token: None, + added_at: None, + access_token_expires_at: Some(now + Duration::minutes(10)), + status: AccountStatus::Active, + }) + .await + .expect("account should be inserted"); + store + .set_next_refresh_at( + "acct-refresh-no-rt-clear-next", + Some(now + Duration::minutes(5)), + ) + .await + .expect("stale future refresh time should persist"); + let refresher = CountingTokenRefresher::default(); + let task = codex_proxy_rs::upstream::accounts::token_refresh::RuntimeTokenRefreshService::new( + store.clone(), + RefreshPolicy { + refresh_margin_seconds: 300, + refresh_concurrency: 1, + }, + refresher.clone(), + ); + + let timer_summary = task + .schedule_account_timers_once_at(now) + .await + .expect("timer scheduling should skip account without refresh token"); + let after_timer_scan = store + .get("acct-refresh-no-rt-clear-next") + .await + .expect("account should load") + .expect("account should exist"); + store + .set_next_refresh_at( + "acct-refresh-no-rt-clear-next", + Some(now - Duration::minutes(1)), + ) + .await + .expect("stale due refresh time should persist"); + + let due_summary = task + .refresh_due_accounts_once_at(now) + .await + .expect("due refresh scan should skip account without refresh token"); + let stored = store + .get("acct-refresh-no-rt-clear-next") + .await + .expect("account should load") + .expect("account should exist"); + + assert_eq!(timer_summary.skipped, 1); + assert_eq!(due_summary.skipped, 1); + assert_eq!(refresher.calls.load(Ordering::SeqCst), 0); + assert!(after_timer_scan.next_refresh_at.is_none()); + assert!(stored.next_refresh_at.is_none()); + assert_eq!(stored.status, AccountStatus::Active); + assert_eq!( + stored.access_token.expose_secret(), + old_access_token.as_str() + ); +} + #[tokio::test] async fn token_refresh_task_should_fire_per_account_timer_at_refresh_time() { let dir = tempfile::tempdir().expect("temp dir"); diff --git a/frontend/src/components/base/BaseScrollbar.vue b/frontend/src/components/base/BaseScrollbar.vue index 51258c650..23b8944c5 100644 --- a/frontend/src/components/base/BaseScrollbar.vue +++ b/frontend/src/components/base/BaseScrollbar.vue @@ -7,12 +7,14 @@ const props = withDefaults( defineProps<{ viewClass?: string maxHeight?: string + height?: string horizontal?: boolean vertical?: boolean }>(), { viewClass: '', maxHeight: undefined, + height: undefined, horizontal: false, vertical: true, }, @@ -60,7 +62,7 @@ const horizontalThumbStyle = computed(() => ({ })) const rootClasses = computed(() => [ 'relative min-h-0 overflow-hidden', - props.maxHeight ? undefined : 'h-full', + props.maxHeight || props.height ? undefined : 'h-full', ]) const wrapClasses = computed(() => [ 'min-h-0 overflow-auto max-h-[inherit] [-ms-overflow-style:none] scrollbar-none [&::-webkit-scrollbar]:h-0 [&::-webkit-scrollbar]:w-0 [&::-webkit-scrollbar]:bg-transparent', @@ -204,6 +206,15 @@ async function scrollToTop() { update() } +async function scrollToBottom() { + const wrap = wrapRef.value + if (wrap) { + wrap.scrollTop = wrap.scrollHeight + } + await nextTick() + update() +} + function handleScroll() { const wrap = wrapRef.value if (!wrap) { @@ -357,6 +368,7 @@ useEventListener(document, 'pointerup', handleHorizontalThumbPointerUp) defineExpose({ update, scrollToTop, + scrollToBottom, wrapRef, }) @@ -364,7 +376,7 @@ defineExpose({