fix(accounts): proxy profile avatars through backend

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-31 09:56:23 +08:00
co-authored by Codex
parent 7ad0ca48c5
commit a62bb196f7
18 changed files with 715 additions and 31 deletions
+1
View File
@@ -1363,6 +1363,7 @@ dependencies = [
"argon2",
"async-trait",
"base64 0.23.1",
"bytes",
"chrono",
"chrono-tz",
"cron",
+1
View File
@@ -15,6 +15,7 @@ path = "tests/main.rs"
argon2 = "0.5.3"
async-trait = "0.1.89"
base64 = "0.23.0"
bytes = "1.11.1"
chrono = "0.4.45"
futures = "0.3.32"
gateway-core = { path = "../gateway-core" }
@@ -1,9 +1,11 @@
//! Provider 管理能力交换的中立 Command 与 Result。
use std::fmt;
use std::{fmt, pin::Pin};
use async_trait::async_trait;
use bytes::Bytes;
use chrono::{DateTime, NaiveDate, Utc};
use futures::Stream;
use gateway_core::{
engine::credential::{OpaqueProviderData, ProviderAccountId, ProviderAccountIdentity},
routing::{ProviderKind, UpstreamModelId},
@@ -679,6 +681,35 @@ pub struct ProviderProfileStatistics {
pub activity_insights: ProviderProfileActivityInsights,
}
/// Provider 头像正文的中立字节流;公共层不解释 MIME,也不限制总字节数。
pub type ProviderProfileAvatarStream =
Pin<Box<dyn Stream<Item = Result<Bytes, ProviderProfileAvatarStreamError>> + Send + 'static>>;
/// Provider 头像流在响应开始后的通用失败,不携带上游 URL 或正文。
#[derive(Debug, thiserror::Error)]
#[error("provider profile avatar stream failed")]
pub struct ProviderProfileAvatarStreamError;
/// Provider 已校验并打开的官方头像响应。
pub struct ProviderProfileAvatar {
pub content_type: Option<String>,
pub content_length: Option<u64>,
pub etag: Option<String>,
pub body: ProviderProfileAvatarStream,
}
impl fmt::Debug for ProviderProfileAvatar {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ProviderProfileAvatar")
.field("content_type", &self.content_type)
.field("content_length", &self.content_length)
.field("etag", &self.etag)
.field("body", &"<stream>")
.finish()
}
}
/// Provider 返回的一张安全主动额度重置卡。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderResetCredit {
@@ -17,8 +17,8 @@ use crate::model::provider_credentials::{
PendingAuthorizationMutation, PrepareCredentialImport, PrepareCredentialRefresh,
PrepareCredentialRotation, PreparedAuthorizationCommit, PreparedCredentialImport,
PreparedCredentialRotation, ProviderExport, ProviderExportCredentialInput, ProviderModels,
ProviderProfileStatistics, ProviderQuota, ProviderQuotaRequest, ProviderResetCreditResult,
ProviderResetCredits,
ProviderProfileAvatar, ProviderProfileStatistics, ProviderQuota, ProviderQuotaRequest,
ProviderResetCreditResult, ProviderResetCredits,
};
/// Provider 管理失败的稳定分类。
@@ -157,6 +157,14 @@ pub trait ProviderAdmin: Send + Sync {
Err(ProviderAdminError::new(ProviderAdminErrorKind::Unsupported))
}
/// 打开 Provider 官方头像字节流;不支持该能力的 Provider 使用默认拒绝。
async fn profile_avatar(
&self,
_account_id: &ProviderAccountId,
) -> Result<ProviderProfileAvatar, ProviderAdminError> {
Err(ProviderAdminError::new(ProviderAdminErrorKind::Unsupported))
}
/// 查询 Provider 主动额度重置卡;不支持该能力的 Provider 使用默认拒绝。
async fn reset_credits(
&self,
@@ -25,8 +25,9 @@ use crate::{
provider_credentials::{
AccountDirectoryItem, AccountDirectoryPage, AccountExportBundle, AccountRefreshResult,
ConsumeProviderResetCredit, PrepareCredentialRefresh, ProviderModels,
ProviderProfileStatistics, ProviderQuota, ProviderQuotaRequest, ProviderQuotaWindow,
ProviderResetCreditResult, ProviderResetCredits, QuotaLocalUsageAttribution,
ProviderProfileAvatar, ProviderProfileStatistics, ProviderQuota, ProviderQuotaRequest,
ProviderQuotaWindow, ProviderResetCreditResult, ProviderResetCredits,
QuotaLocalUsageAttribution,
},
},
ports::{
@@ -90,6 +91,13 @@ pub trait AccountsService: Send + Sync {
Err(AdminError::invalid("当前 Provider 不支持账号统计"))
}
async fn profile_avatar(
&self,
_account_id: &ProviderAccountId,
) -> Result<ProviderProfileAvatar, AdminError> {
Err(AdminError::invalid("当前 Provider 不支持账号头像"))
}
async fn reset_credits(
&self,
_context: &MutationContext,
@@ -566,6 +574,17 @@ impl AccountsService for DefaultAccountsService {
.map_err(|error| map_provider_error(error, "provider profile statistics"))
}
async fn profile_avatar(
&self,
account_id: &ProviderAccountId,
) -> Result<ProviderProfileAvatar, AdminError> {
let (_, provider) = self.provider_for_account(account_id).await?;
provider
.profile_avatar(account_id)
.await
.map_err(|error| map_provider_error(error, "provider profile avatar"))
}
async fn reset_credits(
&self,
context: &MutationContext,
@@ -4,8 +4,9 @@ use std::{collections::BTreeSet, convert::Infallible, fmt};
use axum::{
Router,
body::Body,
extract::State,
http::{HeaderValue, StatusCode, header::CACHE_CONTROL},
http::{HeaderName, HeaderValue, StatusCode, header},
response::{
IntoResponse, Response,
sse::{Event, KeepAlive, Sse},
@@ -28,10 +29,10 @@ use gateway_admin::model::{
AuthorizationStarted, CompleteAuthorization, ConsumeProviderResetCredit,
CredentialDeletion, CredentialDeletionResult, CredentialImportResult, CredentialMutation,
CredentialMutationResult, ImportCredentials, ProviderDocument, ProviderModels,
ProviderProfileActivityInsights, ProviderProfileDailyUsage, ProviderProfileInvocation,
ProviderProfileStatistics, ProviderProfileStatisticsSummary, ProviderQuota,
ProviderQuotaWindow, ProviderResetCredit, ProviderResetCreditResult, ProviderResetCredits,
RotateCredential, StartAuthorization,
ProviderProfileActivityInsights, ProviderProfileAvatar, ProviderProfileDailyUsage,
ProviderProfileInvocation, ProviderProfileStatistics, ProviderProfileStatisticsSummary,
ProviderQuota, ProviderQuotaWindow, ProviderResetCredit, ProviderResetCreditResult,
ProviderResetCredits, RotateCredential, StartAuthorization,
},
};
use gateway_core::{
@@ -60,6 +61,7 @@ const MAX_ID_TOKEN_BYTES: usize = 16 * 1024;
const MAX_CALLBACK_URL_BYTES: usize = 64 * 1024;
const MAX_ACCOUNT_DELETE_BATCH: usize = 200;
const MAX_ACCOUNT_GROUP_BATCH: usize = 1000;
const MAX_AVATAR_VERSION_BYTES: usize = 32;
/// 账号列表查询参数。
#[derive(Debug, Clone, Default, Deserialize)]
@@ -422,6 +424,33 @@ impl AccountIdQuery {
}
}
/// 头像 GET 的固定 query;`version` 只参与浏览器缓存键,不进入上游请求。
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AccountProfileAvatarQuery {
pub account_id: String,
pub version: Option<String>,
}
impl AccountProfileAvatarQuery {
pub fn validate(&self) -> Result<(), WireValidationError> {
require_account_id(&self.account_id, "accountId")?;
if self.version.as_deref().is_some_and(|version| {
version.is_empty()
|| version.len() > MAX_AVATAR_VERSION_BYTES
|| !version.bytes().all(|byte| byte.is_ascii_alphanumeric())
}) {
return Err(WireValidationError::new("version"));
}
Ok(())
}
fn into_id(self) -> Result<ProviderAccountId, WireValidationError> {
self.validate()?;
ProviderAccountId::new(self.account_id).map_err(|_| WireValidationError::new("accountId"))
}
}
/// 敏感导出的固定 query;IDs 使用逗号分隔,禁止隐式导出全部账号。
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
@@ -1291,6 +1320,10 @@ where
"/api/admin/accounts/profile-statistics",
get(account_profile_statistics::<S>),
)
.route(
"/api/admin/accounts/profile-avatar",
get(account_profile_avatar::<S>),
)
.route(
"/api/admin/accounts/reset-credits",
get(account_reset_credits::<S>).post(consume_account_reset_credit::<S>),
@@ -1400,7 +1433,7 @@ where
let mut response = AdminResponse::new(StatusCode::OK, AdminEnvelope::ok(data)).into_response();
response
.headers_mut()
.insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
Ok(response)
}
@@ -1646,6 +1679,69 @@ where
))
}
async fn account_profile_avatar<S>(
_auth: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<AccountProfileAvatarQuery>,
) -> Result<Response, AdminError>
where
S: AdminSessionState + Send + Sync,
{
let account_id = query.into_id().map_err(map_wire_error)?;
let avatar = state
.admin_services()
.accounts()
.profile_avatar(&account_id)
.await
.map_err(map_service_error)?;
Ok(profile_avatar_response(avatar))
}
/// 将 Provider 头像流投影为受保护的同源 HTTP 响应。
#[must_use]
pub fn profile_avatar_response(avatar: ProviderProfileAvatar) -> Response {
let ProviderProfileAvatar {
content_type,
content_length,
etag,
body,
} = avatar;
let mut response = Response::new(Body::from_stream(body));
*response.status_mut() = StatusCode::OK;
let headers = response.headers_mut();
headers.insert(
header::CONTENT_TYPE,
content_type
.and_then(|value| HeaderValue::from_str(&value).ok())
.unwrap_or_else(|| HeaderValue::from_static("application/octet-stream")),
);
if let Some(value) =
content_length.and_then(|length| HeaderValue::from_str(&length.to_string()).ok())
{
headers.insert(header::CONTENT_LENGTH, value);
}
if let Some(value) = etag.and_then(|value| HeaderValue::from_str(&value).ok()) {
headers.insert(header::ETAG, value);
}
headers.insert(
header::CACHE_CONTROL,
HeaderValue::from_static("private, max-age=3600"),
);
headers.insert(
HeaderName::from_static("x-content-type-options"),
HeaderValue::from_static("nosniff"),
);
headers.insert(
HeaderName::from_static("cross-origin-resource-policy"),
HeaderValue::from_static("same-origin"),
);
headers.insert(
HeaderName::from_static("content-security-policy"),
HeaderValue::from_static("sandbox; default-src 'none'"),
);
response
}
async fn refresh_account_quota<S>(
_auth: AdminAuth,
State(state): State<S>,
+2 -1
View File
@@ -58,6 +58,7 @@ async fn admin_not_found() -> AdminError {
async fn no_store(mut response: Response) -> Response {
response
.headers_mut()
.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
.entry(header::CACHE_CONTROL)
.or_insert(HeaderValue::from_static("no-store"));
response
}
@@ -76,7 +76,7 @@ mod profile_statistics {
ProviderProfileActivityInsights, ProviderProfileDailyUsage, ProviderProfileInvocation,
ProviderProfileStatistics, ProviderProfileStatisticsSummary,
};
use gateway_api::admin::accounts::AccountProfileStatisticsData;
use gateway_api::admin::accounts::{AccountProfileAvatarQuery, AccountProfileStatisticsData};
use serde_json::json;
#[test]
@@ -134,6 +134,74 @@ mod profile_statistics {
assert!(value.get("models").is_none());
assert!(value.get("estimatedCost").is_none());
}
#[test]
fn profile_avatar_query_accepts_only_bounded_cache_versions() {
let valid: AccountProfileAvatarQuery = serde_json::from_value(json!({
"accountId": "acct_openai",
"version": "k9m2z1"
}))
.expect("decode avatar query");
valid.validate().expect("validate avatar query");
for version in ["", "contains-dash", "x23456789012345678901234567890123"] {
let query: AccountProfileAvatarQuery = serde_json::from_value(json!({
"accountId": "acct_openai",
"version": version
}))
.expect("decode invalid avatar query");
assert_eq!(
query.validate().expect_err("reject version").field(),
"version"
);
}
assert!(
serde_json::from_value::<AccountProfileAvatarQuery>(json!({
"accountId": "acct_openai",
"sourceUrl": "https://example.test/avatar"
}))
.is_err()
);
}
}
mod profile_avatar {
use axum::body::to_bytes;
use bytes::Bytes;
use futures::stream;
use gateway_admin::model::provider_credentials::ProviderProfileAvatar;
use gateway_api::admin::accounts::profile_avatar_response;
#[tokio::test]
async fn profile_avatar_response_preserves_stream_metadata_and_private_cache() {
let response = profile_avatar_response(ProviderProfileAvatar {
content_type: Some("image/svg+xml".to_owned()),
content_length: Some(6),
etag: Some("\"v1\"".to_owned()),
body: Box::pin(stream::iter([Ok(Bytes::from_static(b"avatar"))])),
});
assert_eq!(response.status(), 200);
assert_eq!(response.headers()["content-type"], "image/svg+xml");
assert_eq!(response.headers()["content-length"], "6");
assert_eq!(response.headers()["etag"], "\"v1\"");
assert_eq!(response.headers()["cache-control"], "private, max-age=3600");
assert_eq!(response.headers()["x-content-type-options"], "nosniff");
assert_eq!(
response.headers()["cross-origin-resource-policy"],
"same-origin"
);
assert_eq!(
response.headers()["content-security-policy"],
"sandbox; default-src 'none'"
);
assert_eq!(
to_bytes(response.into_body(), 16)
.await
.expect("avatar response body"),
Bytes::from_static(b"avatar")
);
}
}
mod batch_update {
+46 -8
View File
@@ -5,6 +5,7 @@ use std::time::SystemTime;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures::TryStreamExt as _;
use gateway_admin::model::accounts::AccountRecord;
use gateway_admin::model::observability::{
CalculatedBillingBreakdown, CurrencyCost, DashboardDesktopRelease, DashboardWireAttribute,
@@ -19,9 +20,10 @@ use gateway_admin::model::provider_credentials::{
PreparedCredentialCreate, PreparedCredentialImport, PreparedCredentialRotation,
PreparedCredentialRotationFacts, ProviderDocument, ProviderExport,
ProviderExportCredentialInput, ProviderModel, ProviderModels, ProviderProfileActivityInsights,
ProviderProfileDailyUsage, ProviderProfileInvocation, ProviderProfileStatistics,
ProviderProfileStatisticsSummary, ProviderQuota, ProviderQuotaRequest, ProviderQuotaWindow,
ProviderQuotaWindowRole, ProviderResetCredit, ProviderResetCreditResult, ProviderResetCredits,
ProviderProfileAvatar, ProviderProfileAvatarStreamError, ProviderProfileDailyUsage,
ProviderProfileInvocation, ProviderProfileStatistics, ProviderProfileStatisticsSummary,
ProviderQuota, ProviderQuotaRequest, ProviderQuotaWindow, ProviderQuotaWindowRole,
ProviderResetCredit, ProviderResetCreditResult, ProviderResetCredits,
QuotaLocalUsageAttribution,
};
use gateway_admin::model::{MutationActor, MutationContext, Revision};
@@ -49,10 +51,10 @@ use crate::credential::{
CodexCredentialAdminService, CodexCredentialCatalogError, CodexCredentialCatalogService,
CodexCredentialProfileService, CodexCredentialQuotaError, CodexCredentialQuotaService,
CodexOAuthAdmin, CodexOAuthAdminError, CodexOAuthPendingClaimOutcome, CodexOAuthPendingStore,
CodexOAuthPendingStoreError, CodexPendingAuthorization, CodexProfileStatisticsError,
CodexQuotaWindow, CodexQuotaWindowKind, CodexQuotaWindowRole, CodexResetCreditsError,
CompleteCodexOAuthAuthorization, CompletedCodexOAuthCredential, ExportManagedCodexCredential,
StartCodexOAuthAuthorization, StoredCodexPendingAuthorization,
CodexOAuthPendingStoreError, CodexPendingAuthorization, CodexProfileAvatarError,
CodexProfileStatisticsError, CodexQuotaWindow, CodexQuotaWindowKind, CodexQuotaWindowRole,
CodexResetCreditsError, CompleteCodexOAuthAuthorization, CompletedCodexOAuthCredential,
ExportManagedCodexCredential, StartCodexOAuthAuthorization, StoredCodexPendingAuthorization,
};
use crate::credential::{
CodexCredentialCodec, CodexOAuthSecret, oauth_owner_ref, parse_access_token_expiration,
@@ -61,7 +63,9 @@ use crate::transport::CodexWebSocketPool;
use crate::transport::profile::{
CodexDesktopReleaseSnapshot, CodexDesktopReleaseStatus, CodexWireProfile, CodexWireProfileState,
};
use crate::transport::{CodexProfileStatistics, OpenAiBillingUsage, openai_billing_breakdown};
use crate::transport::{
CodexProfileAvatar, CodexProfileStatistics, OpenAiBillingUsage, openai_billing_breakdown,
};
const PROVIDER_NAME: &str = "openai";
const PENDING_DOCUMENT_SCHEMA_VERSION: u64 = 3;
@@ -462,6 +466,18 @@ impl ProviderAdmin for OpenAiAdminProvider {
Ok(project_profile_statistics(statistics))
}
async fn profile_avatar(
&self,
account_id: &ProviderAccountId,
) -> Result<ProviderProfileAvatar, ProviderAdminError> {
self.account(account_id).await?;
self.profile_statistics
.profile_avatar(account_id)
.await
.map(project_profile_avatar)
.map_err(map_profile_avatar_error)
}
async fn reset_credits(
&self,
account_id: &ProviderAccountId,
@@ -865,6 +881,15 @@ fn project_profile_statistics(statistics: CodexProfileStatistics) -> ProviderPro
}
}
fn project_profile_avatar(avatar: CodexProfileAvatar) -> ProviderProfileAvatar {
ProviderProfileAvatar {
content_type: avatar.content_type,
content_length: avatar.content_length,
etag: avatar.etag,
body: Box::pin(avatar.body.map_err(|_| ProviderProfileAvatarStreamError)),
}
}
fn project_reset_credits(
credits: crate::transport::CodexRateLimitResetCredits,
) -> ProviderResetCredits {
@@ -1498,6 +1523,19 @@ fn map_profile_statistics_error(error: CodexProfileStatisticsError) -> ProviderA
}
}
fn map_profile_avatar_error(error: CodexProfileAvatarError) -> ProviderAdminError {
match error {
CodexProfileAvatarError::ProfileStatistics(error) => map_profile_statistics_error(error),
CodexProfileAvatarError::Missing => provider_admin_error(ProviderAdminErrorKind::NotFound),
CodexProfileAvatarError::InvalidSource | CodexProfileAvatarError::Upstream { .. } => {
provider_admin_error(ProviderAdminErrorKind::BadGateway)
}
CodexProfileAvatarError::TransportUnavailable => {
provider_admin_error(ProviderAdminErrorKind::Unavailable)
}
}
}
fn map_reset_credits_error(error: CodexResetCreditsError) -> ProviderAdminError {
use CodexResetCreditsError as Error;
@@ -38,7 +38,9 @@ pub use oauth::{
CodexPendingAuthorization, CompleteCodexOAuthAuthorization, CompletedCodexOAuthAuthorization,
CompletedCodexOAuthCredential, StartCodexOAuthAuthorization, StoredCodexPendingAuthorization,
};
pub use profile_statistics::{CodexCredentialProfileService, CodexProfileStatisticsError};
pub use profile_statistics::{
CodexCredentialProfileService, CodexProfileAvatarError, CodexProfileStatisticsError,
};
pub use quota::{
CodexAccountQuotaSnapshot, CodexCredentialQuotaError, CodexCredentialQuotaService,
CodexQuotaFact, CodexQuotaRefreshPolicy, CodexQuotaSyncSummary, CodexQuotaWindow,
@@ -1,6 +1,10 @@
//! OpenAI 当前账号个人资料统计查询编排。
//! OpenAI 当前账号个人资料统计与官方头像查询编排。
use std::time::SystemTime;
use std::{
collections::HashMap,
sync::Mutex,
time::{Duration, Instant, SystemTime},
};
use gateway_core::engine::credential::ProviderAccountId;
use reqwest::Client;
@@ -10,11 +14,14 @@ use uuid::Uuid;
use crate::transport::profile::CodexWireProfileState;
use crate::transport::{
CodexBackendClient, CodexClientError, CodexProfileStatistics, CodexRequestContext,
CodexBackendClient, CodexClientError, CodexProfileAvatar, CodexProfileAvatarFetchError,
CodexProfileStatistics, CodexRequestContext, fetch_profile_avatar,
};
use super::CodexCredentialRepository;
const PROFILE_AVATAR_SOURCE_TTL: Duration = Duration::from_secs(10 * 60);
#[derive(Error)]
pub enum CodexProfileStatisticsError {
#[error("Codex profile-statistics credential data is invalid")]
@@ -59,16 +66,55 @@ impl std::fmt::Debug for CodexProfileStatisticsError {
}
}
#[derive(Error)]
pub enum CodexProfileAvatarError {
#[error(transparent)]
ProfileStatistics(#[from] CodexProfileStatisticsError),
#[error("Codex profile does not provide an avatar")]
Missing,
#[error("Codex profile avatar source is invalid")]
InvalidSource,
#[error("Codex profile avatar upstream returned HTTP {status}")]
Upstream { status: u16 },
#[error("Codex profile avatar transport is unavailable")]
TransportUnavailable,
}
impl std::fmt::Debug for CodexProfileAvatarError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::ProfileStatistics(error) => formatter
.debug_tuple("ProfileStatistics")
.field(error)
.finish(),
Self::Missing => formatter.write_str("Missing"),
Self::InvalidSource => formatter.write_str("InvalidSource"),
Self::Upstream { status } => formatter
.debug_struct("Upstream")
.field("status", status)
.finish(),
Self::TransportUnavailable => formatter.write_str("TransportUnavailable"),
}
}
}
#[derive(Clone)]
struct CachedProfileAvatarSource {
source: String,
expires_at: Instant,
}
pub struct CodexCredentialProfileService {
repository: CodexCredentialRepository,
profile: CodexWireProfileState,
http: Client,
base_url: String,
avatar_sources: Mutex<HashMap<ProviderAccountId, CachedProfileAvatarSource>>,
}
impl CodexCredentialProfileService {
#[must_use]
pub const fn new(
pub fn new(
repository: CodexCredentialRepository,
profile: CodexWireProfileState,
http: Client,
@@ -79,6 +125,7 @@ impl CodexCredentialProfileService {
profile,
http,
base_url,
avatar_sources: Mutex::new(HashMap::new()),
}
}
@@ -114,7 +161,7 @@ impl CodexCredentialProfileService {
.authorization_header()
.map_err(|_| CodexProfileStatisticsError::InvalidCredentialData)?;
let request_id = format!("profile_statistics_{}", Uuid::now_v7().simple());
CodexBackendClient::new(
let statistics = CodexBackendClient::new(
self.http.clone(),
self.base_url.clone(),
self.profile.clone(),
@@ -126,7 +173,72 @@ impl CodexCredentialProfileService {
None,
))
.await
.map_err(map_client_error)
.map_err(map_client_error)?;
self.remember_avatar_source(account_id, statistics.image_url.as_deref());
Ok(statistics)
}
pub async fn profile_avatar(
&self,
account_id: &ProviderAccountId,
) -> Result<CodexProfileAvatar, CodexProfileAvatarError> {
let source = match self.cached_avatar_source(account_id) {
Some(source) => source,
None => self
.profile_statistics(account_id)
.await?
.image_url
.ok_or(CodexProfileAvatarError::Missing)?,
};
let desktop_user_agent = self.profile.snapshot().desktop_user_agent();
fetch_profile_avatar(&self.http, &self.base_url, &desktop_user_agent, &source)
.await
.map_err(map_avatar_fetch_error)
}
fn cached_avatar_source(&self, account_id: &ProviderAccountId) -> Option<String> {
let now = Instant::now();
let mut sources = self
.avatar_sources
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
sources.retain(|_, source| source.expires_at > now);
sources.get(account_id).map(|source| source.source.clone())
}
fn remember_avatar_source(&self, account_id: &ProviderAccountId, source: Option<&str>) {
let now = Instant::now();
let mut sources = self
.avatar_sources
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
sources.retain(|_, source| source.expires_at > now);
match source {
Some(source) => {
sources.insert(
account_id.clone(),
CachedProfileAvatarSource {
source: source.to_owned(),
expires_at: now + PROFILE_AVATAR_SOURCE_TTL,
},
);
}
None => {
sources.remove(account_id);
}
}
}
}
fn map_avatar_fetch_error(error: CodexProfileAvatarFetchError) -> CodexProfileAvatarError {
match error {
CodexProfileAvatarFetchError::InvalidSource => CodexProfileAvatarError::InvalidSource,
CodexProfileAvatarFetchError::Upstream { status } => {
CodexProfileAvatarError::Upstream { status }
}
CodexProfileAvatarFetchError::TransportUnavailable => {
CodexProfileAvatarError::TransportUnavailable
}
}
}
@@ -9,6 +9,7 @@ pub mod diagnostics;
pub mod endpoints;
pub mod headers;
pub mod profile;
pub mod profile_avatar;
pub mod profile_statistics;
pub mod protocol;
pub mod request;
@@ -41,6 +42,10 @@ pub use self::{
WHAM_USAGE_PATH, endpoint_url, usage_endpoint_url,
},
headers::build_codex_base_headers,
profile_avatar::{
CodexProfileAvatar, CodexProfileAvatarFetchError, CodexProfileAvatarStreamError,
fetch_profile_avatar,
},
profile_statistics::{
CodexProfileActivityInsights, CodexProfileDailyUsage, CodexProfileInvocation,
CodexProfileStatistics, CodexProfileStatisticsSummary,
@@ -0,0 +1,141 @@
//! Codex 官方个人资料头像的固定来源流式 transport。
use std::{pin::Pin, time::Duration};
use async_stream::try_stream;
use bytes::Bytes;
use futures::{Stream, StreamExt as _};
use reqwest::{
Client, Url,
header::{ACCEPT, CONTENT_TYPE, ETAG, USER_AGENT},
};
use tokio::time::timeout;
use super::endpoints::endpoint_url;
const OFFICIAL_AVATAR_ORIGIN: &str = "https://chatgpt.com";
const OFFICIAL_AVATAR_PATH_PREFIX: &str = "/backend-api/estuary/public_content/enc/";
const PROVIDER_AVATAR_PATH_PREFIX: &str = "/estuary/public_content/enc/";
const PROFILE_AVATAR_ACCEPT: &str =
"image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8";
const PROFILE_AVATAR_HEADERS_TIMEOUT: Duration = Duration::from_secs(15);
const PROFILE_AVATAR_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
/// 头像正文流;不累积完整正文,也不设置总字节数上限。
pub type CodexProfileAvatarStream =
Pin<Box<dyn Stream<Item = Result<Bytes, CodexProfileAvatarStreamError>> + Send + 'static>>;
/// 响应开始后的头像流失败;不得携带带签名的上游 URL。
#[derive(Debug, thiserror::Error)]
#[error("Codex profile avatar stream failed")]
pub struct CodexProfileAvatarStreamError;
/// 已打开的官方头像响应;MIME 只透传,不做格式白名单判断。
pub struct CodexProfileAvatar {
pub content_type: Option<String>,
pub content_length: Option<u64>,
pub etag: Option<String>,
pub body: CodexProfileAvatarStream,
}
impl std::fmt::Debug for CodexProfileAvatar {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("CodexProfileAvatar")
.field("content_type", &self.content_type)
.field("content_length", &self.content_length)
.field("etag", &self.etag)
.field("body", &"<stream>")
.finish()
}
}
#[derive(Debug, thiserror::Error)]
pub enum CodexProfileAvatarFetchError {
#[error("Codex profile avatar source is invalid")]
InvalidSource,
#[error("Codex profile avatar upstream returned HTTP {status}")]
Upstream { status: u16 },
#[error("Codex profile avatar transport is unavailable")]
TransportUnavailable,
}
/// 校验官方来源后,通过当前 Provider base URL 打开头像字节流。
///
/// # Errors
///
/// 来源不是固定的 ChatGPT Estuary 路径、请求失败或上游返回非成功状态时返回错误。
pub async fn fetch_profile_avatar(
client: &Client,
base_url: &str,
desktop_user_agent: &str,
source: &str,
) -> Result<CodexProfileAvatar, CodexProfileAvatarFetchError> {
let source_path = official_avatar_source_path(source)?;
let target = Url::parse(&endpoint_url(base_url, &source_path))
.map_err(|_| CodexProfileAvatarFetchError::TransportUnavailable)?;
let response = timeout(
PROFILE_AVATAR_HEADERS_TIMEOUT,
client
.get(target)
.header(ACCEPT, PROFILE_AVATAR_ACCEPT)
.header(USER_AGENT, desktop_user_agent)
.send(),
)
.await
.map_err(|_| CodexProfileAvatarFetchError::TransportUnavailable)?
.map_err(|_| CodexProfileAvatarFetchError::TransportUnavailable)?;
let status = response.status();
if !status.is_success() {
return Err(CodexProfileAvatarFetchError::Upstream {
status: status.as_u16(),
});
}
let content_type = response
.headers()
.get(CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let etag = response
.headers()
.get(ETAG)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let content_length = response.content_length();
let mut upstream = Box::pin(response.bytes_stream());
let body = try_stream! {
loop {
match timeout(PROFILE_AVATAR_STREAM_IDLE_TIMEOUT, upstream.next()).await {
Ok(Some(Ok(chunk))) => yield chunk,
Ok(Some(Err(_))) | Err(_) => Err(CodexProfileAvatarStreamError)?,
Ok(None) => break,
}
}
};
Ok(CodexProfileAvatar {
content_type,
content_length,
etag,
body: Box::pin(body),
})
}
fn official_avatar_source_path(source: &str) -> Result<String, CodexProfileAvatarFetchError> {
let source = Url::parse(source).map_err(|_| CodexProfileAvatarFetchError::InvalidSource)?;
if source.origin().ascii_serialization() != OFFICIAL_AVATAR_ORIGIN
|| !source.username().is_empty()
|| source.password().is_some()
|| source.query().is_some()
|| source.fragment().is_some()
{
return Err(CodexProfileAvatarFetchError::InvalidSource);
}
let opaque_path = source
.path()
.strip_prefix(OFFICIAL_AVATAR_PATH_PREFIX)
.filter(|path| !path.is_empty())
.ok_or(CodexProfileAvatarFetchError::InvalidSource)?;
Ok(format!("{PROVIDER_AVATAR_PATH_PREFIX}{opaque_path}"))
}
@@ -1,6 +1,7 @@
use std::sync::Arc;
use chrono::{TimeZone as _, Utc};
use futures::TryStreamExt as _;
use gateway_core::engine::credential::ProviderAccountId;
use provider_openai::credential::{CodexCredentialProfileService, ImportCodexOAuthCredential};
use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
@@ -155,3 +156,62 @@ async fn profile_statistics_keeps_profile_when_stats_are_unavailable() {
assert_eq!(statistics.summary.total_text_tokens, None);
assert_eq!(statistics.daily_usage, None);
}
#[tokio::test]
async fn profile_avatar_reuses_cached_source_and_sends_no_account_credentials() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/wham/profiles/me"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"profile": {
"display_name": "Ada",
"profile_picture_url": "https://chatgpt.com/backend-api/estuary/public_content/enc/avatar-token="
},
"stats": {},
"metadata": {"stats_error": null}
})))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/estuary/public_content/enc/avatar-token="))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "image/svg+xml")
.set_body_bytes(b"<svg xmlns=\"http://www.w3.org/2000/svg\"/>"),
)
.expect(1)
.mount(&server)
.await;
let (store, service) = service("acct_profile_avatar", &server).await;
let account = store
.account("acct_profile_avatar")
.expect("profile account");
service
.profile_statistics(account.id())
.await
.expect("profile statistics");
let avatar = service
.profile_avatar(account.id())
.await
.expect("profile avatar");
let content_type = avatar.content_type.clone();
let chunks = avatar
.body
.try_collect::<Vec<_>>()
.await
.expect("avatar body");
let body = chunks.into_iter().flatten().collect::<Vec<_>>();
let requests = server.received_requests().await.expect("recorded requests");
let avatar_request = requests
.iter()
.find(|request| request.url.path() == "/estuary/public_content/enc/avatar-token=")
.expect("avatar request");
assert_eq!(content_type.as_deref(), Some("image/svg+xml"));
assert_eq!(body, b"<svg xmlns=\"http://www.w3.org/2000/svg\"/>");
assert!(!avatar_request.headers.contains_key("authorization"));
assert!(!avatar_request.headers.contains_key("cookie"));
assert!(!avatar_request.headers.contains_key("chatgpt-account-id"));
}
@@ -54,6 +54,7 @@ mod headers;
mod http_client;
mod latency;
mod profile;
mod profile_avatar;
mod protocol;
mod request;
mod reset_credits;
@@ -0,0 +1,80 @@
use futures::TryStreamExt as _;
use provider_openai::transport::{CodexProfileAvatarFetchError, fetch_profile_avatar};
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
const OFFICIAL_AVATAR_SOURCE: &str =
"https://chatgpt.com/backend-api/estuary/public_content/enc/opaque-token=";
#[tokio::test]
async fn profile_avatar_streams_unrestricted_content_type_and_body_size() {
let server = MockServer::start().await;
let body = vec![b'x'; 1024 * 1024 + 1];
Mock::given(method("GET"))
.and(path("/estuary/public_content/enc/opaque-token="))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "application/x-avatar-fixture")
.insert_header("etag", "\"avatar-v1\"")
.set_body_bytes(body.clone()),
)
.expect(1)
.mount(&server)
.await;
let client = reqwest::Client::builder()
.no_proxy()
.redirect(reqwest::redirect::Policy::none())
.build()
.expect("client");
let avatar = fetch_profile_avatar(
&client,
&server.uri(),
"codex_cli_rs/1.0.0 (linux; x86_64)",
OFFICIAL_AVATAR_SOURCE,
)
.await
.expect("profile avatar");
let content_type = avatar.content_type.clone();
let content_length = avatar.content_length;
let etag = avatar.etag.clone();
let chunks = avatar.body.try_collect::<Vec<_>>().await.expect("body");
let streamed = chunks.into_iter().flatten().collect::<Vec<_>>();
let requests = server.received_requests().await.expect("avatar request");
let headers = &requests.first().expect("one avatar request").headers;
assert_eq!(
content_type.as_deref(),
Some("application/x-avatar-fixture")
);
assert_eq!(content_length, Some(body.len() as u64));
assert_eq!(etag.as_deref(), Some("\"avatar-v1\""));
assert_eq!(streamed, body);
assert_eq!(
headers.get("accept").and_then(|value| value.to_str().ok()),
Some("image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8")
);
assert_eq!(
headers
.get("user-agent")
.and_then(|value| value.to_str().ok()),
Some("codex_cli_rs/1.0.0 (linux; x86_64)")
);
}
#[tokio::test]
async fn profile_avatar_rejects_non_official_sources_before_request() {
let client = reqwest::Client::new();
for source in [
"https://example.com/backend-api/estuary/public_content/enc/token",
"https://chatgpt.com/other/token",
"https://chatgpt.com/backend-api/estuary/public_content/enc/token?next=1",
"https://chatgpt.com/backend-api/estuary/public_content/enc/",
] {
let error = fetch_profile_avatar(&client, "http://127.0.0.1:9", "test-agent", source)
.await
.expect_err("invalid source");
assert!(matches!(error, CodexProfileAvatarFetchError::InvalidSource));
}
}
+15
View File
@@ -368,6 +368,21 @@ export function getAccountProfileStatistics(data: AccountIdParam) {
})
}
export function accountProfileAvatarUrl(accountId: string, sourceUrl: string) {
const params = new URLSearchParams({
accountId,
version: stableAvatarVersion(sourceUrl),
})
return `/api/admin/accounts/profile-avatar?${params.toString()}`
}
function stableAvatarVersion(value: string) {
let hash = 0
for (const character of value)
hash = (hash * 33 + (character.codePointAt(0) ?? 0)) % 2_147_483_647
return hash.toString(36)
}
export function refreshAccountQuota(data: AccountIdParam) {
return request<AccountQuotaResponse>({
url: '/api/admin/accounts/quota/refresh',
@@ -4,6 +4,7 @@ import type { AccountProfileStatisticsResponse } from '@/api'
import { UserRound } from '@lucide/vue'
import { computed, shallowRef, watch } from 'vue'
import { accountProfileAvatarUrl } from '@/api'
import { formatCompactNumber, formatInteger } from '@/utils/number'
import { stablePresetVisualToneClass } from '../../utils/visualTone'
import AccountPlanBadge from '../AccountPlanBadge.vue'
@@ -23,6 +24,10 @@ const initial = computed(() => Array.from(displayName.value.trim())[0]?.toUpperC
const avatarToneClass = computed(() =>
stablePresetVisualToneClass(props.account.id || accountIdentity.value || displayName.value),
)
const avatarUrl = computed(() => {
const sourceUrl = props.profile.imageUrl?.trim()
return sourceUrl ? accountProfileAvatarUrl(props.account.id, sourceUrl) : null
})
const metrics = computed(() => [
{
label: '累计 Token 数',
@@ -49,7 +54,7 @@ const metrics = computed(() => [
])
watch(
() => props.profile.imageUrl,
avatarUrl,
() => {
imageFailed.value = false
},
@@ -95,8 +100,8 @@ function formatDuration(value: number | null) {
:class="avatarToneClass"
>
<img
v-if="profile.imageUrl && !imageFailed"
:src="profile.imageUrl"
v-if="avatarUrl && !imageFailed"
:src="avatarUrl"
:alt="`${displayName} 的头像`"
class="size-full object-cover"
referrerpolicy="no-referrer"