use std::{collections::BTreeMap, error::Error, fmt, future::Future, sync::Arc, time::Instant}; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; use bytes::Bytes; use hmac::{Hmac, Mac}; use reqwest::Method; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use time::{Duration, OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::{sync::Semaphore, time::sleep}; use tracing::{info, warn}; pub mod client_downloads; pub mod project_snapshots; pub mod template_library; type HmacSha256 = Hmac; pub const DEFAULT_POST_EXPIRE_SECONDS: u64 = 10 * 60; pub const DEFAULT_READ_EXPIRE_SECONDS: u64 = 10 * 60; pub const DEFAULT_POST_MAX_SIZE_BYTES: u64 = 20 * 1024 * 1024; pub const DEFAULT_SUCCESS_ACTION_STATUS: u16 = 200; pub const DEFAULT_METADATA_TOTAL_BYTES_LIMIT: usize = 8 * 1024; pub const DEFAULT_IMMUTABLE_CACHE_CONTROL: &str = "public, max-age=31536000, immutable"; const OSS_V4_ALGORITHM: &str = "OSS4-HMAC-SHA256"; const OSS_V4_REQUEST: &str = "aliyun_v4_request"; const OSS_V4_SERVICE: &str = "oss"; const OSS_UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const OSS_PROVIDER: &str = "aliyun-oss"; pub const LEGACY_PUBLIC_PREFIXES: [&str; 14] = [ "generated-character-drafts", "generated-characters", "generated-animations", "generated-big-fish-assets", "generated-square-hole-assets", "generated-wooden-fish-assets", "generated-match3d-assets", "generated-puzzle-assets", "generated-puzzle-clear-assets", "generated-jump-hop-assets", "generated-custom-world-scenes", "generated-custom-world-covers", "generated-bark-battle-assets", "generated-qwen-sprites", ]; #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum OssObjectAccess { Public, Private, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum LegacyAssetPrefix { EditorAgent, AgcErrorReports, CharacterDrafts, Characters, Animations, BigFishAssets, SquareHoleAssets, WoodenFishAssets, Match3DAssets, PuzzleAssets, PuzzleClearAssets, JumpHopAssets, CustomWorldScenes, CustomWorldCovers, BarkBattleAssets, QwenSprites, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssConfig { bucket: String, endpoint: String, access_key_id: String, access_key_secret: String, default_read_expire_seconds: u64, default_post_expire_seconds: u64, default_post_max_size_bytes: u64, default_success_action_status: u16, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssPostObjectRequest { pub prefix: LegacyAssetPrefix, pub path_segments: Vec, pub file_name: String, pub content_type: Option, pub access: OssObjectAccess, pub metadata: BTreeMap, pub max_size_bytes: Option, pub expire_seconds: Option, pub success_action_status: Option, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssSignedGetObjectUrlRequest { pub object_key: String, pub expire_seconds: Option, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssHeadObjectRequest { pub object_key: String, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssGetObjectRequest { pub object_key: String, pub max_bytes: usize, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssDeleteObjectRequest { pub object_key: String, } /// 服务端专用内部对象写入请求。 /// /// 与 `OssPutObjectRequest` 的区别是对象键由调用方按内部前缀完整给出,不再走 /// `path_segments`/`file_name` 的低位规范化,因此可以保留项目内的原始大小写与 /// 目录层级。键必须落在 `normalize_internal_object_key` 允许的内部前缀下。 #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssInternalPutObjectRequest { pub object_key: String, pub content_type: Option, pub access: OssObjectAccess, pub metadata: BTreeMap, pub body: Vec, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssPutObjectRequest { pub prefix: LegacyAssetPrefix, pub path_segments: Vec, pub file_name: String, pub content_type: Option, pub access: OssObjectAccess, pub metadata: BTreeMap, pub body: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct OssPostObjectResponse { #[serde(rename = "signatureVersion")] pub signature_version: &'static str, pub provider: &'static str, pub bucket: String, pub endpoint: String, pub host: String, #[serde(rename = "objectKey")] pub object_key: String, #[serde(rename = "legacyPublicPath")] pub legacy_public_path: String, #[serde(rename = "contentType", skip_serializing_if = "Option::is_none")] pub content_type: Option, pub access: OssObjectAccess, #[serde(rename = "keyPrefix")] pub key_prefix: String, #[serde(rename = "expiresAt")] pub expires_at: String, #[serde(rename = "maxSizeBytes")] pub max_size_bytes: u64, #[serde(rename = "successActionStatus")] pub success_action_status: u16, #[serde(rename = "formFields")] pub form_fields: OssPostObjectFormFields, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct OssSignedGetObjectUrlResponse { pub provider: &'static str, pub bucket: String, pub endpoint: String, pub host: String, #[serde(rename = "objectKey")] pub object_key: String, #[serde(rename = "expiresAt")] pub expires_at: String, #[serde(rename = "signedUrl")] pub signed_url: String, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssHeadObjectResponse { pub bucket: String, pub object_key: String, pub content_length: u64, pub content_type: Option, pub etag: Option, pub last_modified: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct OssPutObjectResponse { pub provider: &'static str, pub bucket: String, pub endpoint: String, pub host: String, #[serde(rename = "objectKey")] pub object_key: String, #[serde(rename = "legacyPublicPath")] pub legacy_public_path: String, #[serde(rename = "contentType", skip_serializing_if = "Option::is_none")] pub content_type: Option, #[serde(rename = "contentLength")] pub content_length: u64, pub access: OssObjectAccess, #[serde(skip_serializing_if = "Option::is_none")] pub etag: Option, #[serde(rename = "lastModified", skip_serializing_if = "Option::is_none")] pub last_modified: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct OssPostObjectFormFields { pub key: String, pub policy: String, #[serde(rename = "x-oss-signature-version")] pub signature_version: String, #[serde(rename = "x-oss-credential")] pub credential: String, #[serde(rename = "x-oss-date")] pub date: String, #[serde(rename = "x-oss-signature")] pub signature: String, #[serde(rename = "success_action_status")] pub success_action_status: String, #[serde(rename = "Content-Type", skip_serializing_if = "Option::is_none")] pub content_type: Option, #[serde(rename = "Cache-Control", skip_serializing_if = "Option::is_none")] pub cache_control: Option, #[serde(flatten)] pub metadata: BTreeMap, } #[derive(Clone, Debug)] pub struct OssClient { config: OssConfig, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum OssRequestOperation { Put, Head, Get, Delete, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct OssRequestError { pub status: Option, pub timeout: bool, pub connect: bool, pub transport: bool, pub oss_code: Option, pub oss_request_id: Option, pub operation: OssRequestOperation, pub message: String, } #[derive(Debug, PartialEq, Eq)] pub enum OssError { InvalidConfig(String), InvalidRequest(String), ObjectNotFound(String), Request(OssRequestError), SerializePolicy(String), Sign(String), } // 平台 OSS 错误只先归类,不在 platform 层绑定 HTTP status。 #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum OssErrorKind { InvalidConfig, InvalidRequest, ObjectNotFound, Request, SerializePolicy, Sign, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct OssRequestAttemptContext { pub frame_index: usize, pub operation: &'static str, } const CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS: u32 = 3; const CHARACTER_ANIMATION_OSS_RETRY_DELAYS_MS: [u64; 2] = [250, 500]; const CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES: usize = 16 * 1024; const OSS_ERROR_CODE_MAX_BYTES: usize = 128; const OSS_REQUEST_ID_MAX_BYTES: usize = 256; struct PreparedHeadObject { object_key: String, target_url: reqwest::Url, } struct PreparedPutObject { object_key: String, target_url: reqwest::Url, content_type: Option, headers: BTreeMap, content_length: u64, access: OssObjectAccess, body: Bytes, } impl LegacyAssetPrefix { pub fn parse(raw: &str) -> Option { let normalized = raw .trim() .trim_start_matches('/') .trim_end_matches('/') .trim_end_matches('*') .trim_end_matches('/'); match normalized { "editor-agent" => Some(Self::EditorAgent), "agc" => Some(Self::AgcErrorReports), "generated-character-drafts" => Some(Self::CharacterDrafts), "generated-characters" => Some(Self::Characters), "generated-animations" => Some(Self::Animations), "generated-big-fish-assets" => Some(Self::BigFishAssets), "generated-square-hole-assets" => Some(Self::SquareHoleAssets), "generated-wooden-fish-assets" => Some(Self::WoodenFishAssets), "generated-match3d-assets" => Some(Self::Match3DAssets), "generated-puzzle-assets" => Some(Self::PuzzleAssets), "generated-puzzle-clear-assets" => Some(Self::PuzzleClearAssets), "generated-jump-hop-assets" => Some(Self::JumpHopAssets), "generated-custom-world-scenes" => Some(Self::CustomWorldScenes), "generated-custom-world-covers" => Some(Self::CustomWorldCovers), "generated-bark-battle-assets" => Some(Self::BarkBattleAssets), "generated-qwen-sprites" => Some(Self::QwenSprites), _ => None, } } pub fn as_str(&self) -> &'static str { match self { Self::EditorAgent => "editor-agent", Self::AgcErrorReports => "agc", Self::CharacterDrafts => "generated-character-drafts", Self::Characters => "generated-characters", Self::Animations => "generated-animations", Self::BigFishAssets => "generated-big-fish-assets", Self::SquareHoleAssets => "generated-square-hole-assets", Self::WoodenFishAssets => "generated-wooden-fish-assets", Self::Match3DAssets => "generated-match3d-assets", Self::PuzzleAssets => "generated-puzzle-assets", Self::PuzzleClearAssets => "generated-puzzle-clear-assets", Self::JumpHopAssets => "generated-jump-hop-assets", Self::CustomWorldScenes => "generated-custom-world-scenes", Self::CustomWorldCovers => "generated-custom-world-covers", Self::BarkBattleAssets => "generated-bark-battle-assets", Self::QwenSprites => "generated-qwen-sprites", } } pub fn as_public_path_prefix(&self) -> String { format!("/{}", self.as_str()) } pub fn from_object_key(raw: &str) -> Option { let normalized = raw.trim().trim_start_matches('/').trim(); let prefix = normalized.split('/').next()?; match Self::parse(prefix) { // agc/error-reports is a server-only write prefix, never a caller-supplied object key. Some(Self::AgcErrorReports) => None, other => other, } } } impl OssConfig { #[allow(clippy::too_many_arguments)] pub fn new( bucket: String, endpoint: String, access_key_id: String, access_key_secret: String, default_read_expire_seconds: u64, default_post_expire_seconds: u64, default_post_max_size_bytes: u64, default_success_action_status: u16, ) -> Result { let bucket = normalize_required_value(bucket, "OSS bucket 不能为空")?; let endpoint = normalize_endpoint(&endpoint)?; let access_key_id = normalize_required_value(access_key_id, "OSS AccessKeyId 不能为空")?; let access_key_secret = normalize_required_value(access_key_secret, "OSS AccessKeySecret 不能为空")?; if default_read_expire_seconds == 0 { return Err(OssError::InvalidConfig( "OSS 私有读签名有效期必须大于 0".to_string(), )); } if default_post_expire_seconds == 0 { return Err(OssError::InvalidConfig( "OSS PostObject 签名有效期必须大于 0".to_string(), )); } if default_post_max_size_bytes == 0 { return Err(OssError::InvalidConfig( "OSS PostObject 最大上传大小必须大于 0".to_string(), )); } if !(100..=999).contains(&default_success_action_status) { return Err(OssError::InvalidConfig( "OSS success_action_status 必须是三位 HTTP 状态码".to_string(), )); } Ok(Self { bucket, endpoint, access_key_id, access_key_secret, default_read_expire_seconds, default_post_expire_seconds, default_post_max_size_bytes, default_success_action_status, }) } pub fn upload_host(&self) -> String { format!("https://{}.{}", self.bucket, self.endpoint) } pub fn endpoint(&self) -> &str { &self.endpoint } pub fn bucket(&self) -> &str { &self.bucket } pub fn access_key_id(&self) -> &str { &self.access_key_id } pub fn access_key_secret(&self) -> &str { &self.access_key_secret } } impl OssClient { pub fn new(config: OssConfig) -> Self { Self { config } } pub fn config_bucket(&self) -> &str { self.config.bucket() } pub fn sign_post_object( &self, request: OssPostObjectRequest, ) -> Result { let started_at = Instant::now(); let requested_prefix = request.prefix.as_str(); let requested_content_type = request .content_type .as_deref() .map(str::trim) .unwrap_or("") .to_string(); let requested_metadata_count = request.metadata.len(); let result = (|| { let max_size_bytes = request .max_size_bytes .unwrap_or(self.config.default_post_max_size_bytes); let expire_seconds = request .expire_seconds .unwrap_or(self.config.default_post_expire_seconds); let success_action_status = request .success_action_status .unwrap_or(self.config.default_success_action_status); if max_size_bytes == 0 { return Err(OssError::InvalidRequest( "maxSizeBytes 必须大于 0".to_string(), )); } if expire_seconds == 0 { return Err(OssError::InvalidRequest( "expireSeconds 必须大于 0".to_string(), )); } if !(100..=999).contains(&success_action_status) { return Err(OssError::InvalidRequest( "successActionStatus 必须是三位 HTTP 状态码".to_string(), )); } let sanitized_segments = request .path_segments .iter() .map(|segment| sanitize_path_segment(segment)) .filter(|segment| !segment.is_empty()) .collect::>(); let file_name = sanitize_file_name(&request.file_name)?; let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name); let legacy_public_path = format!("/{}", object_key); let content_type = normalize_optional_value(request.content_type); let metadata = normalize_metadata(request.metadata)?; let cache_control = Some(DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string()); let expires_at = OffsetDateTime::now_utc() .checked_add(Duration::seconds(i64::try_from(expire_seconds).map_err( |_| OssError::InvalidRequest("expireSeconds 超出可支持范围".to_string()), )?)) .ok_or_else(|| { OssError::InvalidRequest("expireSeconds 计算结果溢出".to_string()) })?; let expires_at = expires_at.format(&Rfc3339).map_err(|error| { OssError::SerializePolicy(format!("格式化过期时间失败:{error}")) })?; let signed_at = OffsetDateTime::now_utc(); let signature_scope = build_v4_signature_scope(&self.config.endpoint, signed_at)?; let signature_date = build_v4_signature_date(signed_at)?; let credential = format!("{}/{}", self.config.access_key_id, signature_scope); let policy_json = build_policy_json( &self.config.bucket, &object_key, &expires_at, max_size_bytes, success_action_status, content_type.as_deref(), cache_control.as_deref(), &metadata, &credential, &signature_date, ); let policy = serde_json::to_string(&policy_json).map_err(|error| { OssError::SerializePolicy(format!("序列化 policy 失败:{error}")) })?; let encoded_policy = BASE64_STANDARD.encode(policy.as_bytes()); let signature = sign_v4_content( &self.config.access_key_secret, &signature_scope, &encoded_policy, )?; Ok(OssPostObjectResponse { signature_version: "v4", provider: OSS_PROVIDER, bucket: self.config.bucket.clone(), endpoint: self.config.endpoint.clone(), host: self.config.upload_host(), object_key: object_key.clone(), legacy_public_path, content_type: content_type.clone(), access: request.access, key_prefix: build_key_prefix(request.prefix, &sanitized_segments), expires_at, max_size_bytes, success_action_status, form_fields: OssPostObjectFormFields { key: object_key, policy: encoded_policy, signature_version: OSS_V4_ALGORITHM.to_string(), credential, date: signature_date, signature, success_action_status: success_action_status.to_string(), content_type, cache_control, metadata, }, }) })(); match &result { Ok(response) => info!( provider = OSS_PROVIDER, operation = "sign_post_object", bucket = %response.bucket, endpoint = %response.endpoint, object_key = %response.object_key, key_prefix = %response.key_prefix, access = oss_access_label(response.access), content_type = %response.content_type.as_deref().unwrap_or(""), max_size_bytes = response.max_size_bytes, success_action_status = response.success_action_status, metadata_count = response.form_fields.metadata.len(), expires_at = %response.expires_at, elapsed_ms = elapsed_ms(started_at), "OSS PostObject 签名完成" ), Err(error) => warn!( provider = OSS_PROVIDER, operation = "sign_post_object", bucket = %self.config.bucket(), endpoint = %self.config.endpoint(), key_prefix = requested_prefix, content_type = %requested_content_type, metadata_count = requested_metadata_count, error_kind = oss_error_kind_label(error), message = %error, elapsed_ms = elapsed_ms(started_at), "OSS PostObject 签名失败" ), } result } // 私有 bucket 的对象读取统一走短期签名 URL,避免把长期主凭证下发给浏览器。 pub fn sign_get_object_url( &self, request: OssSignedGetObjectUrlRequest, ) -> Result { self.sign_get_object_url_with_normalizer( request, normalize_object_key, "sign_get_object_url", ) } pub fn sign_internal_get_object_url( &self, request: OssSignedGetObjectUrlRequest, ) -> Result { self.sign_get_object_url_with_normalizer( request, normalize_editor_agent_messages_object_key, "sign_internal_get_object_url", ) } fn sign_get_object_url_with_normalizer( &self, request: OssSignedGetObjectUrlRequest, normalize: fn(&str) -> Result, operation: &'static str, ) -> Result { let started_at = Instant::now(); let requested_object_key = request .object_key .trim() .trim_start_matches('/') .trim() .to_string(); let result = (|| { let expire_seconds = request .expire_seconds .unwrap_or(self.config.default_read_expire_seconds); if expire_seconds == 0 { return Err(OssError::InvalidRequest( "expireSeconds 必须大于 0".to_string(), )); } let object_key = normalize(&request.object_key)?; let expires_at = OffsetDateTime::now_utc() .checked_add(Duration::seconds(i64::try_from(expire_seconds).map_err( |_| OssError::InvalidRequest("expireSeconds 超出可支持范围".to_string()), )?)) .ok_or_else(|| { OssError::InvalidRequest("expireSeconds 计算结果溢出".to_string()) })?; let expires_at_text = expires_at .format(&Rfc3339) .map_err(|error| OssError::Sign(format!("格式化过期时间失败:{error}")))?; let signed_at = OffsetDateTime::now_utc(); let signed_at_text = build_v4_signature_date(signed_at)?; let signature_scope = build_v4_signature_scope(&self.config.endpoint, signed_at)?; let credential = format!("{}/{}", self.config.access_key_id, signature_scope); let mut query = BTreeMap::from([ ("x-oss-additional-headers".to_string(), "host".to_string()), ( "x-oss-signature-version".to_string(), OSS_V4_ALGORITHM.to_string(), ), ("x-oss-credential".to_string(), credential), ("x-oss-date".to_string(), signed_at_text), ("x-oss-expires".to_string(), expire_seconds.to_string()), ]); let canonical_uri = build_v4_canonical_uri(&self.config.bucket, Some(&object_key)); let object_url_path = format!("/{}", encode_url_path(&object_key)); let additional_headers = "host"; let canonical_headers = format!("host:{}.{}\n", self.config.bucket(), self.config.endpoint()); let canonical_query = build_canonical_query_string(&query); let canonical_request = build_v4_canonical_request( Method::GET.as_str(), &canonical_uri, &canonical_query, &canonical_headers, additional_headers, OSS_UNSIGNED_PAYLOAD, ); let string_to_sign = build_v4_string_to_sign( query["x-oss-date"].as_str(), &signature_scope, &canonical_request, ); let signature = sign_v4_content( &self.config.access_key_secret, &signature_scope, &string_to_sign, )?; query.insert("x-oss-signature".to_string(), signature); let signed_url = format!( "{}{}?{}", self.config.upload_host(), object_url_path, build_canonical_query_string(&query) ); Ok(OssSignedGetObjectUrlResponse { provider: OSS_PROVIDER, bucket: self.config.bucket.clone(), endpoint: self.config.endpoint.clone(), host: self.config.upload_host(), object_key, expires_at: expires_at_text, signed_url, }) })(); match &result { Ok(response) => info!( provider = OSS_PROVIDER, operation, bucket = %response.bucket, endpoint = %response.endpoint, object_key = %response.object_key, expires_at = %response.expires_at, elapsed_ms = elapsed_ms(started_at), "OSS GetObject 读签名完成" ), Err(error) => warn!( provider = OSS_PROVIDER, operation, bucket = %self.config.bucket(), endpoint = %self.config.endpoint(), object_key = %requested_object_key, error_kind = oss_error_kind_label(error), message = %error, elapsed_ms = elapsed_ms(started_at), "OSS GetObject 读签名失败" ), } result } // 上传完成确认前,服务端必须自己探测一次对象,不能只相信客户端回传的 object_key。 pub async fn head_object( &self, client: &reqwest::Client, request: OssHeadObjectRequest, ) -> Result { let started_at = Instant::now(); let requested_object_key = request .object_key .trim() .trim_start_matches('/') .trim() .to_string(); let mut response_status = None; let result = async { let object_key = normalize_object_key(&request.object_key)?; let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key).map_err( |error| { request_error( OssRequestOperation::Head, &format!("构造 OSS 对象 URL 失败:{error}"), ) }, )?; let response = send_signed_request( client, &self.config, Method::HEAD, Some(&object_key), target_url, OssRequestOperation::Head, ) .await?; response_status = Some(response.status().as_u16()); if response.status() == reqwest::StatusCode::NOT_FOUND { return Err(OssError::ObjectNotFound(format!( "OSS 对象不存在:{}", request.object_key ))); } if !response.status().is_success() { return Err(request_status_error( OssRequestOperation::Head, response.status().as_u16(), format!("OSS HEAD Object 失败,状态码:{}", response.status()), )); } let headers = response.headers(); let content_length = head_object_content_length(headers); let content_type = headers .get(reqwest::header::CONTENT_TYPE) .and_then(|value| value.to_str().ok()) .map(|value| value.to_string()); let etag = headers .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(|value| value.trim_matches('"').to_string()); let last_modified = headers .get(reqwest::header::LAST_MODIFIED) .and_then(|value| value.to_str().ok()) .map(|value| value.to_string()); Ok(OssHeadObjectResponse { bucket: self.config.bucket.clone(), object_key, content_length, content_type, etag, last_modified, }) } .await; match &result { Ok(response) => info!( provider = OSS_PROVIDER, operation = "head_object", bucket = %response.bucket, endpoint = %self.config.endpoint(), object_key = %response.object_key, status = response_status.unwrap_or(reqwest::StatusCode::OK.as_u16()), status_class = http_status_class_from_option(response_status), content_length = response.content_length, content_type = %response.content_type.as_deref().unwrap_or(""), etag_present = response.etag.is_some(), last_modified_present = response.last_modified.is_some(), elapsed_ms = elapsed_ms(started_at), "OSS HEAD Object 完成" ), Err(error) => warn!( provider = OSS_PROVIDER, operation = "head_object", bucket = %self.config.bucket(), endpoint = %self.config.endpoint(), object_key = %requested_object_key, status = response_status.unwrap_or_default(), status_class = http_status_class_from_option(response_status), error_kind = oss_error_kind_label(error), message = %error, elapsed_ms = elapsed_ms(started_at), "OSS HEAD Object 失败" ), } result } pub async fn get_object( &self, client: &reqwest::Client, request: OssGetObjectRequest, ) -> Result, OssError> { let key = normalize_internal_object_key(&request.object_key)?; let target = build_object_url(&self.config.bucket, &self.config.endpoint, &key) .map_err(|e| request_error(OssRequestOperation::Get, &e.to_string()))?; let mut response = send_signed_request( client, &self.config, Method::GET, Some(&key), target, OssRequestOperation::Get, ) .await?; if response.status() == reqwest::StatusCode::NOT_FOUND { return Err(OssError::ObjectNotFound(format!("OSS 对象不存在:{key}"))); } if !response.status().is_success() { return Err(request_status_error( OssRequestOperation::Get, response.status().as_u16(), format!("OSS GET Object 失败,状态码:{}", response.status()), )); } if response .content_length() .is_some_and(|length| length > request.max_bytes as u64) { return Err(OssError::InvalidRequest("OSS 对象超过读取上限".to_string())); } let mut bytes = Vec::new(); while let Some(chunk) = response .chunk() .await .map_err(|e| request_error_from_reqwest(OssRequestOperation::Get, e))? { if bytes.len().saturating_add(chunk.len()) > request.max_bytes { return Err(OssError::InvalidRequest("OSS 对象超过读取上限".to_string())); } bytes.extend_from_slice(&chunk); } Ok(bytes) } pub async fn delete_object( &self, client: &reqwest::Client, request: OssDeleteObjectRequest, ) -> Result<(), OssError> { let key = normalize_internal_object_key(&request.object_key)?; let target = build_object_url(&self.config.bucket, &self.config.endpoint, &key) .map_err(|e| request_error(OssRequestOperation::Delete, &e.to_string()))?; let response = send_signed_request( client, &self.config, Method::DELETE, Some(&key), target, OssRequestOperation::Delete, ) .await?; if response.status() == reqwest::StatusCode::NOT_FOUND || response.status().is_success() { return Ok(()); } Err(request_status_error( OssRequestOperation::Delete, response.status().as_u16(), format!("OSS DELETE Object 失败,状态码:{}", response.status()), )) } /// 按内部前缀写入服务端专用对象;项目快照允许合法的零字节工程文件。 pub async fn put_internal_object( &self, client: &reqwest::Client, request: OssInternalPutObjectRequest, ) -> Result { let object_key = normalize_internal_object_key(&request.object_key)?; if request.body.is_empty() && !project_snapshots::is_snapshot_file_key(&object_key) { return Err(OssError::InvalidRequest( "服务端内部对象内容不能为空".to_string(), )); } let content_type = normalize_optional_value(request.content_type); let headers = build_put_object_headers(request.metadata)?; let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key) .map_err(|error| { request_error( OssRequestOperation::Put, &format!("构造 OSS 对象 URL 失败:{error}"), ) })?; let content_length = u64::try_from(request.body.len()) .map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?; let builder = signed_request_builder( client, &self.config, Method::PUT, Some(&object_key), target_url, content_type.as_deref(), &headers, )? .header(reqwest::header::CONTENT_LENGTH, content_length) .body(request.body); let response = builder .send() .await .map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?; if !response.status().is_success() { return Err(request_status_error( OssRequestOperation::Put, response.status().as_u16(), format!("OSS PutObject 失败,状态码:{}", response.status()), )); } let headers = response.headers(); let etag = headers .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(|value| value.trim_matches('"').to_string()); let last_modified = headers .get(reqwest::header::LAST_MODIFIED) .and_then(|value| value.to_str().ok()) .map(|value| value.to_string()); Ok(OssPutObjectResponse { provider: OSS_PROVIDER, bucket: self.config.bucket.clone(), endpoint: self.config.endpoint.clone(), host: self.config.upload_host(), legacy_public_path: format!("/{object_key}"), object_key, content_type, content_length, access: request.access, etag, last_modified, }) } /// 探测内部对象是否存在。`Ok(None)` 表示确定不存在,其余失败都按上游错误返回, /// 调用方不能把不确定当成"不存在"。 pub async fn head_internal_object( &self, client: &reqwest::Client, object_key: &str, ) -> Result, OssError> { let object_key = normalize_internal_object_key(object_key)?; let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key) .map_err(|error| { request_error( OssRequestOperation::Head, &format!("构造 OSS 对象 URL 失败:{error}"), ) })?; let response = send_signed_request( client, &self.config, Method::HEAD, Some(&object_key), target_url, OssRequestOperation::Head, ) .await?; if response.status() == reqwest::StatusCode::NOT_FOUND { return Ok(None); } if !response.status().is_success() { return Err(request_status_error( OssRequestOperation::Head, response.status().as_u16(), format!("OSS HEAD Object 失败,状态码:{}", response.status()), )); } let headers = response.headers(); Ok(Some(OssHeadObjectResponse { bucket: self.config.bucket.clone(), object_key, content_length: head_object_content_length(headers), content_type: headers .get(reqwest::header::CONTENT_TYPE) .and_then(|value| value.to_str().ok()) .map(|value| value.to_string()), etag: headers .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(|value| value.trim_matches('"').to_string()), last_modified: headers .get(reqwest::header::LAST_MODIFIED) .and_then(|value| value.to_str().ok()) .map(|value| value.to_string()), })) } // AI 生成资源默认由服务端上传 OSS,Web 端只拿签名读地址,不直接持有写权限。 pub async fn put_object( &self, client: &reqwest::Client, request: OssPutObjectRequest, ) -> Result { let started_at = Instant::now(); let requested_prefix = request.prefix.as_str(); let requested_content_type = request .content_type .as_deref() .map(str::trim) .unwrap_or("") .to_string(); let requested_content_length = request.body.len(); let requested_metadata_count = request.metadata.len(); let mut response_status = None; let result = async { if request.body.is_empty() { return Err(OssError::InvalidRequest( "服务端上传对象内容不能为空".to_string(), )); } let sanitized_segments = request .path_segments .iter() .map(|segment| sanitize_path_segment(segment)) .filter(|segment| !segment.is_empty()) .collect::>(); let file_name = sanitize_file_name(&request.file_name)?; let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name); let content_type = normalize_optional_value(request.content_type); let headers = build_put_object_headers(request.metadata)?; let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key).map_err( |error| { request_error( OssRequestOperation::Put, &format!("构造 OSS 对象 URL 失败:{error}"), ) }, )?; let content_length = u64::try_from(request.body.len()) .map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?; let builder = signed_request_builder( client, &self.config, Method::PUT, Some(&object_key), target_url, content_type.as_deref(), &headers, )? .header(reqwest::header::CONTENT_LENGTH, content_length) .body(request.body); let response = builder .send() .await .map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?; response_status = Some(response.status().as_u16()); if !response.status().is_success() { return Err(request_status_error( OssRequestOperation::Put, response.status().as_u16(), format!("OSS PutObject 失败,状态码:{}", response.status()), )); } let headers = response.headers(); let etag = headers .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(|value| value.trim_matches('"').to_string()); let last_modified = headers .get(reqwest::header::LAST_MODIFIED) .and_then(|value| value.to_str().ok()) .map(|value| value.to_string()); Ok(OssPutObjectResponse { provider: OSS_PROVIDER, bucket: self.config.bucket.clone(), endpoint: self.config.endpoint.clone(), host: self.config.upload_host(), legacy_public_path: format!("/{object_key}"), object_key, content_type, content_length, access: request.access, etag, last_modified, }) } .await; match &result { Ok(response) => info!( provider = OSS_PROVIDER, operation = "put_object", bucket = %response.bucket, endpoint = %response.endpoint, object_key = %response.object_key, access = oss_access_label(response.access), status = response_status.unwrap_or(reqwest::StatusCode::OK.as_u16()), status_class = http_status_class_from_option(response_status), content_length = response.content_length, content_type = %response.content_type.as_deref().unwrap_or(""), etag_present = response.etag.is_some(), last_modified_present = response.last_modified.is_some(), elapsed_ms = elapsed_ms(started_at), "OSS PutObject 上传完成" ), Err(error) => warn!( provider = OSS_PROVIDER, operation = "put_object", bucket = %self.config.bucket(), endpoint = %self.config.endpoint(), key_prefix = requested_prefix, content_length = requested_content_length, content_type = %requested_content_type, metadata_count = requested_metadata_count, status = response_status.unwrap_or_default(), status_class = http_status_class_from_option(response_status), error_kind = oss_error_kind_label(error), message = %error, elapsed_ms = elapsed_ms(started_at), "OSS PutObject 上传失败" ), } result } /// 角色动画帧专用的可重试 PUT。调用方传入进程级并发限制器,单次网络 attempt /// 独占一个 permit,退避等待期间不会占用 permit。 pub async fn put_object_with_retry( &self, client: &reqwest::Client, request: OssPutObjectRequest, io_limiter: Arc, attempt_context: OssRequestAttemptContext, ) -> Result { let prepared = self.prepare_put_object(request)?; let object_key = prepared.object_key.clone(); run_animation_request_with_retry(io_limiter, attempt_context, &object_key, || { self.put_object_once(client, &prepared) }) .await } /// 角色动画帧专用的可重试 HEAD。HEAD 与 PUT 独立重试,HEAD 失败不会重新上传 PUT。 pub async fn head_object_with_retry( &self, client: &reqwest::Client, request: OssHeadObjectRequest, io_limiter: Arc, attempt_context: OssRequestAttemptContext, ) -> Result { let prepared = self.prepare_head_object(request)?; let object_key = prepared.object_key.clone(); run_animation_request_with_retry(io_limiter, attempt_context, &object_key, || { self.head_object_once(client, &prepared) }) .await } fn prepare_head_object( &self, request: OssHeadObjectRequest, ) -> Result { let object_key = normalize_object_key(&request.object_key)?; let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key) .map_err(|error| { request_error( OssRequestOperation::Head, &format!("构造 OSS 对象 URL 失败:{error}"), ) })?; Ok(PreparedHeadObject { object_key, target_url, }) } async fn head_object_once( &self, client: &reqwest::Client, prepared: &PreparedHeadObject, ) -> Result<(OssHeadObjectResponse, u16), OssError> { let response = send_signed_request( client, &self.config, Method::HEAD, Some(&prepared.object_key), prepared.target_url.clone(), OssRequestOperation::Head, ) .await?; if response.status() == reqwest::StatusCode::NOT_FOUND { return Err(OssError::ObjectNotFound(format!( "OSS 对象不存在:{}", prepared.object_key ))); } if !response.status().is_success() { return Err(request_status_error( OssRequestOperation::Head, response.status().as_u16(), format!("OSS HEAD Object 失败,状态码:{}", response.status()), )); } let headers = response.headers(); let content_length = headers .get(reqwest::header::CONTENT_LENGTH) .and_then(|value| value.to_str().ok()) .and_then(|value| value.parse::().ok()) .unwrap_or(0); let content_type = headers .get(reqwest::header::CONTENT_TYPE) .and_then(|value| value.to_str().ok()) .map(str::to_string); let etag = headers .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(|value| value.trim_matches('"').to_string()); let last_modified = headers .get(reqwest::header::LAST_MODIFIED) .and_then(|value| value.to_str().ok()) .map(str::to_string); Ok(( OssHeadObjectResponse { bucket: self.config.bucket.clone(), object_key: prepared.object_key.clone(), content_length, content_type, etag, last_modified, }, response.status().as_u16(), )) } fn prepare_put_object( &self, request: OssPutObjectRequest, ) -> Result { if request.body.is_empty() { return Err(OssError::InvalidRequest( "服务端上传对象内容不能为空".to_string(), )); } let sanitized_segments = request .path_segments .iter() .map(|segment| sanitize_path_segment(segment)) .filter(|segment| !segment.is_empty()) .collect::>(); let file_name = sanitize_file_name(&request.file_name)?; let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name); let content_type = normalize_optional_value(request.content_type); let headers = build_put_object_headers(request.metadata)?; let target_url = build_object_url(&self.config.bucket, &self.config.endpoint, &object_key) .map_err(|error| { request_error( OssRequestOperation::Put, &format!("构造 OSS 对象 URL 失败:{error}"), ) })?; let content_length = u64::try_from(request.body.len()) .map_err(|_| OssError::InvalidRequest("上传对象大小超出可支持范围".to_string()))?; Ok(PreparedPutObject { object_key, target_url, content_type, headers, content_length, access: request.access, body: Bytes::from(request.body), }) } async fn put_object_once( &self, client: &reqwest::Client, prepared: &PreparedPutObject, ) -> Result<(OssPutObjectResponse, u16), OssError> { let response = signed_request_builder( client, &self.config, Method::PUT, Some(&prepared.object_key), prepared.target_url.clone(), prepared.content_type.as_deref(), &prepared.headers, )? .header(reqwest::header::CONTENT_LENGTH, prepared.content_length) .body(prepared.body.clone()) .send() .await .map_err(|error| request_error_from_reqwest(OssRequestOperation::Put, error))?; if !response.status().is_success() { return Err(request_status_error_from_oss_put_response(response).await); } let headers = response.headers(); let etag = headers .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(|value| value.trim_matches('"').to_string()); let last_modified = headers .get(reqwest::header::LAST_MODIFIED) .and_then(|value| value.to_str().ok()) .map(str::to_string); Ok(( OssPutObjectResponse { provider: OSS_PROVIDER, bucket: self.config.bucket.clone(), endpoint: self.config.endpoint.clone(), host: self.config.upload_host(), legacy_public_path: format!("/{}", prepared.object_key), object_key: prepared.object_key.clone(), content_type: prepared.content_type.clone(), content_length: prepared.content_length, access: prepared.access, etag, last_modified, }, response.status().as_u16(), )) } } fn request_error(operation: OssRequestOperation, message: &str) -> OssError { OssError::Request(OssRequestError { status: None, timeout: false, connect: false, transport: false, oss_code: None, oss_request_id: None, operation, message: message.to_string(), }) } async fn request_status_error_from_oss_put_response(mut response: reqwest::Response) -> OssError { let status = response.status(); let header_request_id = response .headers() .get("x-oss-request-id") .and_then(|value| value.to_str().ok()) .and_then(|value| normalize_oss_error_field(value, OSS_REQUEST_ID_MAX_BYTES)); let body_read = if status == reqwest::StatusCode::BAD_REQUEST { read_bounded_oss_error_body(&mut response).await } else { OssErrorBodyRead { body: Vec::new(), read_failure: None, } }; request_status_error_from_oss_parts( OssRequestOperation::Put, status.as_u16(), header_request_id, &body_read.body, body_read.read_failure, ) } /// 400 错误响应体读取失败的原因。仅在部分响应体尚未解析出确定性 /// OSS 错误码时参与重试判定,否则只体现在 message 里。 #[derive(Clone, Copy, Debug, PartialEq, Eq)] struct OssErrorBodyReadFailure { timeout: bool, } struct OssErrorBodyRead { body: Vec, read_failure: Option, } async fn read_bounded_oss_error_body(response: &mut reqwest::Response) -> OssErrorBodyRead { let mut body = Vec::new(); let mut read_failure = None; while body.len() < CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES { let chunk = match response.chunk().await { Ok(Some(chunk)) => chunk, Ok(None) => break, Err(error) => { // 断流/超时不丢弃已读字节:部分响应体可能已含确定性错误码。 read_failure = Some(OssErrorBodyReadFailure { timeout: error.is_timeout(), }); break; } }; let remaining = CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES - body.len(); body.extend_from_slice(&chunk[..chunk.len().min(remaining)]); if chunk.len() > remaining { break; } } OssErrorBodyRead { body, read_failure } } fn request_status_error_from_oss_parts( operation: OssRequestOperation, status: u16, header_request_id: Option, body: &[u8], body_read_failure: Option, ) -> OssError { let header_request_id = header_request_id .as_deref() .and_then(|value| normalize_oss_error_field(value, OSS_REQUEST_ID_MAX_BYTES)); let bounded_body = &body[..body.len().min(CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES)]; let oss_code = extract_oss_error_xml_field(bounded_body, "Code", OSS_ERROR_CODE_MAX_BYTES); let xml_request_id = extract_oss_error_xml_field(bounded_body, "RequestId", OSS_REQUEST_ID_MAX_BYTES); let oss_request_id = header_request_id.or(xml_request_id); // 确定性错误码优先:已解析出 Code 时,响应体读取失败只保留在 message 里, // 不改变重试语义;错误码缺失时才按读取失败归类为可重试的超时/传输错误。 let unclassified_read_failure = if oss_code.is_some() { None } else { body_read_failure }; let timeout = (status == reqwest::StatusCode::BAD_REQUEST.as_u16() && oss_code.as_deref() == Some("RequestTimeout")) || unclassified_read_failure.is_some_and(|failure| failure.timeout); let transport = unclassified_read_failure.is_some_and(|failure| !failure.timeout); let mut message = format!("OSS PutObject 失败,状态码:{status}"); if let Some(oss_code) = oss_code.as_deref() { message.push_str(&format!(",OSS 错误码:{oss_code}")); } if let Some(oss_request_id) = oss_request_id.as_deref() { message.push_str(&format!(",OSS Request ID:{oss_request_id}")); } if body_read_failure.is_some() { message.push_str(",错误响应体读取失败"); } OssError::Request(OssRequestError { status: Some(status), timeout, connect: false, transport, oss_code, oss_request_id, operation, message, }) } fn extract_oss_error_xml_field(body: &[u8], field: &str, max_bytes: usize) -> Option { let body = std::str::from_utf8(body).ok()?; let start_tag = format!("<{field}>"); let end_tag = format!(""); let value_start = body.find(&start_tag)? + start_tag.len(); let value_end = value_start + body[value_start..].find(&end_tag)?; normalize_oss_error_field(&body[value_start..value_end], max_bytes) } fn normalize_oss_error_field(value: &str, max_bytes: usize) -> Option { let value = value.trim(); if value.is_empty() || value.len() > max_bytes || value.chars().any(char::is_control) { return None; } Some(value.to_string()) } async fn run_animation_request_with_retry( io_limiter: Arc, attempt_context: OssRequestAttemptContext, object_key: &str, mut attempt_request: F, ) -> Result where F: FnMut() -> Fut, Fut: Future>, { for attempt in 1..=CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS { let attempt_started_at = Instant::now(); let permit_wait_started_at = Instant::now(); let permit = io_limiter .acquire() .await .map_err(|_| OssError::InvalidConfig("角色动画 OSS 并发限制器已关闭".to_string()))?; let permit_wait_ms = elapsed_ms(permit_wait_started_at); let result = attempt_request().await; drop(permit); let retryable = result.as_ref().err().is_some_and(oss_error_is_retryable); let will_retry = retryable && attempt < CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS; let retry_delay_ms = if will_retry { CHARACTER_ANIMATION_OSS_RETRY_DELAYS_MS[(attempt - 1) as usize] } else { 0 }; let success_status = result.as_ref().ok().map(|(_, status)| *status); log_animation_request_attempt( attempt_context, object_key, attempt, retryable, will_retry, retry_delay_ms, permit_wait_ms, elapsed_ms(attempt_started_at), success_status, result.as_ref().err(), ); match result { Ok((response, _status)) => return Ok(response), Err(_error) if will_retry => { sleep(std::time::Duration::from_millis(retry_delay_ms)).await } Err(error) => return Err(error), } } unreachable!("角色动画 OSS 重试循环必须返回结果") } fn request_error_from_reqwest(operation: OssRequestOperation, error: reqwest::Error) -> OssError { let status = error.status().map(|status| status.as_u16()); let timeout = error.is_timeout(); let connect = error.is_connect(); let transport = !timeout && !connect && (error.is_request() || error.is_body()); OssError::Request(OssRequestError { status, timeout, connect, transport, oss_code: None, oss_request_id: None, operation, message: format!("请求 OSS 失败:{error}"), }) } fn request_status_error(operation: OssRequestOperation, status: u16, message: String) -> OssError { OssError::Request(OssRequestError { status: Some(status), timeout: false, connect: false, transport: false, oss_code: None, oss_request_id: None, operation, message, }) } fn oss_error_is_retryable(error: &OssError) -> bool { let OssError::Request(request_error) = error else { return false; }; match (request_error.status, request_error.oss_code.as_deref()) { (Some(400), Some("RequestTimeout")) => true, // 400 是唯一会读取错误响应体的状态码:错误码缺失且响应体读取 // 超时/断流时,无法证明是确定性 400,按传输错误重试。 (Some(400), None) if request_error.timeout || request_error.transport => true, (Some(408 | 429 | 500..=599), _) => true, (Some(_), _) => false, (None, _) => request_error.timeout || request_error.connect || request_error.transport, } } fn request_error_details( error: Option<&OssError>, ) -> (Option, bool, bool, bool, Option<&str>, Option<&str>) { match error { Some(OssError::Request(request_error)) => ( request_error.status, request_error.timeout, request_error.connect, request_error.transport, request_error.oss_code.as_deref(), request_error.oss_request_id.as_deref(), ), Some(OssError::ObjectNotFound(_)) => (Some(404), false, false, false, None, None), _ => (None, false, false, false, None, None), } } fn log_animation_request_attempt( context: OssRequestAttemptContext, object_key: &str, attempt: u32, retryable: bool, will_retry: bool, retry_delay_ms: u64, permit_wait_ms: u64, elapsed_ms: u64, success_status: Option, error: Option<&OssError>, ) { let (error_status, timeout, connect, transport, oss_code, oss_request_id) = request_error_details(error); let status = success_status.or(error_status); if error.is_none() { info!( provider = OSS_PROVIDER, frame_index = context.frame_index, object_key, operation = context.operation, attempt, max_attempts = CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS, retryable, will_retry, retry_delay_ms, permit_wait_ms, timeout, connect, transport, oss_code = oss_code.unwrap_or_default(), oss_request_id = oss_request_id.unwrap_or_default(), status = status.unwrap_or_default(), elapsed_ms, "角色动画 OSS 请求 attempt 完成" ); return; } warn!( provider = OSS_PROVIDER, frame_index = context.frame_index, object_key, operation = context.operation, attempt, max_attempts = CHARACTER_ANIMATION_OSS_MAX_ATTEMPTS, retryable, will_retry, retry_delay_ms, permit_wait_ms, timeout, connect, transport, oss_code = oss_code.unwrap_or_default(), oss_request_id = oss_request_id.unwrap_or_default(), status = status.unwrap_or_default(), elapsed_ms, error_kind = error.map(oss_error_kind_label), message = error.map(ToString::to_string).unwrap_or_default(), "角色动画 OSS 请求 attempt 完成" ); } impl fmt::Display for OssError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::InvalidConfig(message) | Self::InvalidRequest(message) | Self::SerializePolicy(message) | Self::Sign(message) => f.write_str(message), Self::ObjectNotFound(message) => f.write_str(message), Self::Request(error) => f.write_str(&error.message), } } } impl Error for OssError {} impl OssError { pub fn kind(&self) -> OssErrorKind { match self { Self::InvalidConfig(_) => OssErrorKind::InvalidConfig, Self::InvalidRequest(_) => OssErrorKind::InvalidRequest, Self::ObjectNotFound(_) => OssErrorKind::ObjectNotFound, Self::Request(_) => OssErrorKind::Request, Self::SerializePolicy(_) => OssErrorKind::SerializePolicy, Self::Sign(_) => OssErrorKind::Sign, } } } fn elapsed_ms(started_at: Instant) -> u64 { started_at.elapsed().as_millis().min(u64::MAX as u128) as u64 } fn oss_access_label(access: OssObjectAccess) -> &'static str { match access { OssObjectAccess::Public => "public", OssObjectAccess::Private => "private", } } fn oss_error_kind_label(error: &OssError) -> &'static str { match error.kind() { OssErrorKind::InvalidConfig => "invalid_config", OssErrorKind::InvalidRequest => "invalid_request", OssErrorKind::ObjectNotFound => "object_not_found", OssErrorKind::Request => "request", OssErrorKind::SerializePolicy => "serialize_policy", OssErrorKind::Sign => "sign", } } fn http_status_class_from_option(status: Option) -> &'static str { status.map(http_status_class).unwrap_or("unknown") } fn http_status_class(status: u16) -> &'static str { match status { 100..=199 => "1xx", 200..=299 => "2xx", 300..=399 => "3xx", 400..=499 => "4xx", 500..=599 => "5xx", _ => "unknown", } } fn build_policy_json( bucket: &str, object_key: &str, expires_at: &str, max_size_bytes: u64, success_action_status: u16, content_type: Option<&str>, cache_control: Option<&str>, metadata: &BTreeMap, credential: &str, signature_date: &str, ) -> Value { let mut conditions = vec![ json!({ "bucket": bucket }), json!(["eq", "$key", object_key]), json!(["content-length-range", 1, max_size_bytes]), json!([ "eq", "$success_action_status", success_action_status.to_string() ]), json!(["eq", "$x-oss-signature-version", OSS_V4_ALGORITHM]), json!(["eq", "$x-oss-credential", credential]), json!(["eq", "$x-oss-date", signature_date]), ]; if let Some(content_type) = content_type { conditions.push(json!(["eq", "$content-type", content_type])); } if let Some(cache_control) = cache_control { conditions.push(json!(["eq", "$Cache-Control", cache_control])); } for (key, value) in metadata { conditions.push(json!(["eq", format!("${key}"), value])); } json!({ "expiration": expires_at, "conditions": conditions, }) } /// HEAD 响应没有正文,`reqwest::Response::content_length()` 对 HEAD 恒为 0; /// 对象大小只能读响应头,两个 HEAD 入口共用这一处解析。 fn head_object_content_length(headers: &reqwest::header::HeaderMap) -> u64 { headers .get(reqwest::header::CONTENT_LENGTH) .and_then(|value| value.to_str().ok()) .and_then(|value| value.parse::().ok()) .unwrap_or(0) } fn build_object_url( bucket: &str, endpoint: &str, object_key: &str, ) -> Result { // 对象键是原始路径,#、% 和中文不能被 URL 解析器当作片段或已有转义。 let path = encode_url_path(object_key.trim_start_matches('/')); reqwest::Url::parse(&format!("https://{bucket}.{endpoint}/{path}")) .map_err(|error| error.to_string()) } fn build_object_key( prefix: LegacyAssetPrefix, path_segments: &[String], file_name: &str, ) -> String { let mut parts = Vec::with_capacity(path_segments.len() + 2); parts.push(prefix.as_str().to_string()); parts.extend(path_segments.iter().cloned()); parts.push(file_name.to_string()); parts.join("/") } fn normalize_object_key(raw: &str) -> Result { let normalized = raw.trim().trim_start_matches('/').trim().to_string(); if normalized.is_empty() { return Err(OssError::InvalidRequest("objectKey 不能为空".to_string())); } match LegacyAssetPrefix::from_object_key(&normalized) { Some(LegacyAssetPrefix::EditorAgent) | None => { return Err(OssError::InvalidRequest( "objectKey 必须落在受支持的 OSS 前缀下".to_string(), )); } Some(_) => {} } validate_object_key_segments(&normalized)?; Ok(normalized) } fn normalize_editor_agent_messages_object_key(raw: &str) -> Result { let normalized = raw.trim().trim_start_matches('/').trim().to_string(); if normalized.is_empty() { return Err(OssError::InvalidRequest("objectKey 不能为空".to_string())); } validate_object_key_segments(&normalized)?; let mut segments = normalized.split('/'); let prefix = segments.next(); let file_name = segments.next(); if prefix != Some(LegacyAssetPrefix::EditorAgent.as_str()) || segments.next().is_some() || !file_name .map(|value| { value.starts_with("editor-agent-conv-") && value.ends_with(".json") && value.len() > "editor-agent-conv-.json".len() }) .unwrap_or(false) { return Err(OssError::InvalidRequest( "objectKey 必须是 editor-agent 消息文档".to_string(), )); } Ok(normalized) } /// AGC 内部对象前缀:只允许服务端写入,客户端直传票据、公开对象键与 legacy /// 公开路径都不覆盖这些前缀。 pub const AGC_ERROR_REPORTS_INTERNAL_PREFIX: &str = "agc/error-reports/v1/"; /// 项目快照当前前缀:第二层是部署渠道,读写都只落在本部署渠道下。 pub const AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX: &str = "agc/project-snapshots/v2/"; /// 无渠道的历史项目快照前缀:不再写入、不再枚举,但仍必须保持服务端私有。 pub const AGC_PROJECT_SNAPSHOT_LEGACY_INTERNAL_PREFIX: &str = "agc/project-snapshots/v1/"; const AGC_INTERNAL_OBJECT_PREFIXES: [&str; 3] = [ AGC_ERROR_REPORTS_INTERNAL_PREFIX, AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX, AGC_PROJECT_SNAPSHOT_LEGACY_INTERNAL_PREFIX, ]; /// 项目快照渠道名:与 AGC 客户端更新渠道同形(小写字母开头,只含小写字母、 /// 数字与连字符)。渠道在对象键里是第一层目录,非法值直接拒绝而不是回落。 pub fn validate_agc_project_snapshot_channel(raw: &str) -> Result { let allowed = !raw.is_empty() && raw.len() <= 32 && raw.as_bytes()[0].is_ascii_lowercase() && raw .bytes() .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') && !raw.ends_with('-'); if !allowed { return Err(OssError::InvalidRequest("项目快照渠道名非法".to_string())); } Ok(raw.to_string()) } /// 项目快照文件对象键: /// `agc/project-snapshots/v2/{channel}/{user}/{project}/files/{size}-{digest}/{relativePath}`。 /// /// 键里同时带字节数与内容摘要,既让同一内容重复提交落在同一个对象上,也让 /// "对象已存在且长度一致" 可以作为内容一致的判据;相对路径按原始大小写保留, /// 不走 `put_object` 的低位规范化。 pub fn agc_project_snapshot_file_object_key( channel: &str, user_id: &str, project_id: &str, size_bytes: u64, checksum_digest: &str, relative_path: &str, ) -> Result { let channel = validate_agc_project_snapshot_channel(channel)?; let user = validate_internal_key_segment(user_id, "用户标识")?; let project = validate_internal_key_segment(project_id, "项目标识")?; let digest = validate_internal_checksum_digest(checksum_digest)?; let relative_path = validate_internal_relative_path(relative_path)?; Ok(format!( "{AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX}{channel}/{user}/{project}/files/{size_bytes}-{digest}/{relative_path}" )) } /// 项目快照清单对象键: /// `agc/project-snapshots/v2/{channel}/{user}/{project}/manifest.json`。 pub fn agc_project_snapshot_manifest_object_key( channel: &str, user_id: &str, project_id: &str, ) -> Result { let channel = validate_agc_project_snapshot_channel(channel)?; let user = validate_internal_key_segment(user_id, "用户标识")?; let project = validate_internal_key_segment(project_id, "项目标识")?; Ok(format!( "{AGC_PROJECT_SNAPSHOT_INTERNAL_PREFIX}{channel}/{user}/{project}/manifest.json" )) } fn validate_internal_key_segment(raw: &str, label: &str) -> Result { let trimmed = raw.trim(); let allowed = !trimmed.is_empty() && trimmed.len() <= 128 && trimmed != "." && trimmed != ".." && trimmed.chars().all(|character| { character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') }); if !allowed { return Err(OssError::InvalidRequest(format!( "{label}不能作为 OSS 对象键片段" ))); } Ok(trimmed.to_string()) } fn validate_internal_checksum_digest(raw: &str) -> Result { let trimmed = raw.trim(); if trimmed.is_empty() || trimmed.len() > 64 || !trimmed .chars() .all(|character| character.is_ascii_hexdigit()) { return Err(OssError::InvalidRequest( "对象摘要必须是 1 到 64 位十六进制".to_string(), )); } Ok(trimmed.to_ascii_lowercase()) } fn validate_internal_relative_path(raw: &str) -> Result { if raw.is_empty() || raw.len() > 1024 || raw.starts_with('/') || raw.contains('\\') { return Err(OssError::InvalidRequest( "对象相对路径必须是 1 到 1024 字节的正斜杠相对路径".to_string(), )); } for part in raw.split('/') { if part.is_empty() || part == "." || part == ".." || part.chars().any(char::is_control) { return Err(OssError::InvalidRequest( "对象相对路径包含非法片段".to_string(), )); } } Ok(raw.to_string()) } fn normalize_internal_object_key(raw: &str) -> Result { let normalized = raw.trim().trim_start_matches('/').trim().to_string(); validate_object_key_segments(&normalized)?; if AGC_INTERNAL_OBJECT_PREFIXES .iter() .any(|prefix| normalized.starts_with(prefix)) { Ok(normalized) } else { Err(OssError::InvalidRequest( "objectKey 不属于内部对象前缀".to_string(), )) } } fn validate_object_key_segments(normalized: &str) -> Result<(), OssError> { let segments = normalized.split('/').collect::>(); if segments.len() < 2 { return Err(OssError::InvalidRequest( "objectKey 至少需要包含前缀和文件名".to_string(), )); } for segment in &segments { if segment.is_empty() || *segment == "." || *segment == ".." { return Err(OssError::InvalidRequest( "objectKey 包含非法路径片段".to_string(), )); } if segment.contains('\\') { return Err(OssError::InvalidRequest( "objectKey 不能包含反斜杠".to_string(), )); } } Ok(()) } fn build_key_prefix(prefix: LegacyAssetPrefix, path_segments: &[String]) -> String { let mut parts = Vec::with_capacity(path_segments.len() + 1); parts.push(prefix.as_str().to_string()); parts.extend(path_segments.iter().cloned()); parts.join("/") } fn normalize_metadata( metadata: BTreeMap, ) -> Result, OssError> { let mut normalized = BTreeMap::new(); for (key, value) in metadata { let key = key.trim(); let value = value.trim(); if key.is_empty() || value.is_empty() { continue; } let key = normalize_metadata_key(key); normalized.insert(key, value.to_string()); } let total_bytes = normalized .iter() .map(|(key, value)| key.len() + value.len()) .sum::(); if total_bytes > DEFAULT_METADATA_TOTAL_BYTES_LIMIT { return Err(OssError::InvalidRequest(format!( "x-oss-meta-* 总大小不能超过 {} 字节", DEFAULT_METADATA_TOTAL_BYTES_LIMIT ))); } Ok(normalized) } fn build_put_object_headers( metadata: BTreeMap, ) -> Result, OssError> { // 中文注释:生成资产 object key 含会话与 asset id,内容不可变,适合交给浏览器/CDN 长缓存。 let mut headers = BTreeMap::from([( "Cache-Control".to_string(), DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(), )]); headers.extend(normalize_metadata(metadata)?); Ok(headers) } fn normalize_metadata_key(raw: &str) -> String { let stripped = raw .trim() .trim_start_matches("x-oss-meta-") .trim() .to_ascii_lowercase(); let sanitized = stripped .chars() .map(|character| match character { 'a'..='z' | '0'..='9' | '-' => character, '_' | ' ' | '/' | '.' => '-', _ => '-', }) .collect::(); let sanitized = collapse_dashes(&sanitized); format!( "x-oss-meta-{}", if sanitized.is_empty() { "metadata".to_string() } else { sanitized } ) } fn sanitize_path_segment(raw: &str) -> String { let normalized = raw .trim() .to_ascii_lowercase() .chars() .map(|character| match character { 'a'..='z' | '0'..='9' | '-' | '_' => character, _ => '-', }) .collect::(); collapse_dashes(&normalized) } fn sanitize_file_name(raw: &str) -> Result { let trimmed = raw.trim(); if trimmed.is_empty() { return Err(OssError::InvalidRequest("fileName 不能为空".to_string())); } let file_name = trimmed.rsplit(['/', '\\']).next().unwrap_or(trimmed).trim(); if file_name.is_empty() { return Err(OssError::InvalidRequest("fileName 不能为空".to_string())); } let (raw_stem, raw_extension) = match file_name.rsplit_once('.') { Some((stem, extension)) if !stem.trim().is_empty() && !extension.trim().is_empty() => { (stem, Some(extension)) } _ => (file_name, None), }; let stem = raw_stem .trim() .to_ascii_lowercase() .chars() .map(|character| match character { 'a'..='z' | '0'..='9' | '-' | '_' => character, _ => '-', }) .collect::(); let stem = collapse_dashes(&stem); let stem = if stem.is_empty() { "file".to_string() } else { stem }; let extension = raw_extension .map(|extension| { extension .trim() .to_ascii_lowercase() .chars() .filter(|character| character.is_ascii_alphanumeric()) .collect::() }) .filter(|extension| !extension.is_empty()); Ok(match extension { Some(extension) => format!("{stem}.{extension}"), None => stem, }) } fn normalize_required_value(value: String, message: &str) -> Result { let value = value.trim().to_string(); if value.is_empty() { return Err(OssError::InvalidConfig(message.to_string())); } Ok(value) } fn normalize_optional_value(value: Option) -> Option { value.and_then(|value| { let value = value.trim().to_string(); if value.is_empty() { None } else { Some(value) } }) } fn normalize_endpoint(raw: &str) -> Result { let endpoint = raw .trim() .trim_start_matches("https://") .trim_start_matches("http://") .trim_matches('/') .to_string(); if endpoint.is_empty() { return Err(OssError::InvalidConfig("OSS endpoint 不能为空".to_string())); } Ok(endpoint) } fn collapse_dashes(value: &str) -> String { value .chars() .fold( (String::new(), false), |(mut output, last_is_dash), character| { let is_dash = character == '-'; if is_dash && last_is_dash { return (output, true); } output.push(character); (output, is_dash) }, ) .0 .trim_matches('-') .to_string() } async fn send_signed_request( client: &reqwest::Client, config: &OssConfig, method: Method, object_key: Option<&str>, target_url: reqwest::Url, operation: OssRequestOperation, ) -> Result { signed_request_builder( client, config, method, object_key, target_url, None, &BTreeMap::new(), )? .send() .await .map_err(|error| request_error_from_reqwest(operation, error)) } fn signed_request_builder( client: &reqwest::Client, config: &OssConfig, method: Method, object_key: Option<&str>, target_url: reqwest::Url, content_type: Option<&str>, oss_headers: &BTreeMap, ) -> Result { let signed_at = OffsetDateTime::now_utc(); let signed_at_text = build_v4_signature_date(signed_at)?; let signature_scope = build_v4_signature_scope(config.endpoint(), signed_at)?; let object_path = object_key.map(str::trim).filter(|value| !value.is_empty()); let canonical_uri = build_v4_canonical_uri(config.bucket(), object_path); let body_sha256 = OSS_UNSIGNED_PAYLOAD.to_string(); let mut signed_headers = BTreeMap::from([ ( "host".to_string(), format!("{}.{}", config.bucket(), config.endpoint()), ), ("x-oss-content-sha256".to_string(), body_sha256.clone()), ("x-oss-date".to_string(), signed_at_text.clone()), ]); if let Some(content_type) = content_type { signed_headers.insert("content-type".to_string(), content_type.to_string()); } for (key, value) in oss_headers { signed_headers.insert(key.to_ascii_lowercase(), value.trim().to_string()); } let canonical_headers = build_v4_canonical_headers(&signed_headers); let additional_headers = build_v4_additional_headers(&signed_headers); let query = target_url .query_pairs() .into_owned() .collect::>(); let canonical_query = build_canonical_query_string(&query); let canonical_request = build_v4_canonical_request( method.as_str(), &canonical_uri, &canonical_query, &canonical_headers, &additional_headers, &body_sha256, ); let string_to_sign = build_v4_string_to_sign(&signed_at_text, &signature_scope, &canonical_request); let signature = sign_v4_content( config.access_key_secret(), &signature_scope, &string_to_sign, )?; let mut builder = client .request(method, target_url) .header("x-oss-content-sha256", body_sha256) .header("x-oss-date", signed_at_text) .header( "Authorization", format!( "{OSS_V4_ALGORITHM} Credential={}/{},AdditionalHeaders={},Signature={}", config.access_key_id(), signature_scope, additional_headers, signature ), ); if let Some(content_type) = content_type { builder = builder.header(reqwest::header::CONTENT_TYPE, content_type); } for (key, value) in oss_headers { builder = builder.header(key.as_str(), value.as_str()); } Ok(builder) } fn build_v4_signature_scope(endpoint: &str, signed_at: OffsetDateTime) -> Result { let date = format_v4_signature_scope_date(signed_at); let region = extract_oss_region(endpoint)?; Ok(format!("{date}/{region}/{OSS_V4_SERVICE}/{OSS_V4_REQUEST}")) } fn build_v4_signature_date(signed_at: OffsetDateTime) -> Result { // 中文注释:time::Time 的 Display 在小时小于 10 时不会稳定补零,OSS V4 必须使用固定宽度 UTC 时间。 Ok(format!( "{}T{:02}{:02}{:02}Z", format_v4_signature_scope_date(signed_at), signed_at.hour(), signed_at.minute(), signed_at.second() )) } fn format_v4_signature_scope_date(signed_at: OffsetDateTime) -> String { format!( "{:04}{:02}{:02}", signed_at.year(), signed_at.month() as u8, signed_at.day() ) } fn build_v4_canonical_uri(bucket: &str, object_key: Option<&str>) -> String { match object_key.map(str::trim).filter(|value| !value.is_empty()) { Some(object_key) => format!( "/{}/{}", encode_url_query_value(bucket), encode_url_path(object_key.trim_start_matches('/')) ), None => format!("/{}/", encode_url_query_value(bucket)), } } fn extract_oss_region(endpoint: &str) -> Result { endpoint .trim() .trim_start_matches("https://") .trim_start_matches("http://") .split('.') .next() .and_then(|segment| segment.strip_prefix("oss-")) .map(str::to_string) .filter(|region| !region.is_empty()) .ok_or_else(|| { OssError::InvalidConfig(format!("OSS endpoint 无法解析 region,当前值:{endpoint}")) }) } fn sign_v4_content( access_key_secret: &str, signature_scope: &str, content: &str, ) -> Result { let signing_key = build_v4_signing_key(access_key_secret, signature_scope)?; Ok(hex_sha256_hmac(&signing_key, content.as_bytes())) } fn build_v4_signing_key( access_key_secret: &str, signature_scope: &str, ) -> Result, OssError> { let mut parts = signature_scope.split('/'); let date = parts .next() .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少日期".to_string()))?; let region = parts .next() .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 region".to_string()))?; let service = parts .next() .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 service".to_string()))?; let request = parts .next() .ok_or_else(|| OssError::Sign("OSS V4 签名 scope 缺少 request".to_string()))?; let date_key = hmac_sha256_raw(format!("aliyun_v4{access_key_secret}").as_bytes(), date)?; let region_key = hmac_sha256_raw(&date_key, region)?; let service_key = hmac_sha256_raw(®ion_key, service)?; hmac_sha256_raw(&service_key, request) } fn hmac_sha256_raw(key: &[u8], content: &str) -> Result, OssError> { let mut signer = HmacSha256::new_from_slice(key) .map_err(|error| OssError::Sign(format!("初始化 HMAC-SHA256 失败:{error}")))?; signer.update(content.as_bytes()); Ok(signer.finalize().into_bytes().to_vec()) } fn hex_sha256_hmac(key: &[u8], content: &[u8]) -> String { let mut signer = HmacSha256::new_from_slice(key).expect("HMAC-SHA256 accepts keys of any size"); signer.update(content); hex_lower(&signer.finalize().into_bytes()) } fn build_v4_canonical_request( method: &str, canonical_uri: &str, canonical_query: &str, canonical_headers: &str, signed_headers: &str, payload_hash: &str, ) -> String { format!( "{method}\n{canonical_uri}\n{canonical_query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}" ) } fn build_v4_string_to_sign( signature_date: &str, signature_scope: &str, canonical_request: &str, ) -> String { format!( "{OSS_V4_ALGORITHM}\n{signature_date}\n{signature_scope}\n{}", sha256_hex(canonical_request.as_bytes()) ) } fn sha256_hex(content: &[u8]) -> String { let mut hasher = Sha256::new(); hasher.update(content); hex_lower(&hasher.finalize()) } fn hex_lower(bytes: &[u8]) -> String { bytes .iter() .map(|byte| format!("{byte:02x}")) .collect::() } fn build_v4_canonical_headers(headers: &BTreeMap) -> String { headers .iter() .map(|(key, value)| format!("{}:{}\n", key.to_ascii_lowercase(), value.trim())) .collect::() } fn build_v4_additional_headers(headers: &BTreeMap) -> String { let mut additional_headers = headers .keys() .map(|key| key.to_ascii_lowercase()) .filter(|key| key != "content-type" && key != "content-md5" && !key.starts_with("x-oss-")) .collect::>(); additional_headers.sort(); additional_headers.join(";") } fn build_canonical_query_string(params: &BTreeMap) -> String { let mut encoded = params .iter() .map(|(key, value)| (encode_url_query_value(key), encode_url_query_value(value))) .collect::>(); encoded.sort_by(|left, right| left.0.cmp(&right.0)); encoded .into_iter() // OSS V4 的空值子资源只保留名称,例如 versioning;不套用 S3 的尾随等号。 .map(|(key, value)| { if value.is_empty() { key } else { format!("{key}={value}") } }) .collect::>() .join("&") } fn encode_url_path(path: &str) -> String { path.split('/') .map(encode_url_query_value) .collect::>() .join("/") } fn encode_url_query_value(value: &str) -> String { let mut encoded = String::with_capacity(value.len()); for byte in value.bytes() { match byte { b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { encoded.push(byte as char) } _ => { use std::fmt::Write as _; let _ = write!(&mut encoded, "%{byte:02X}"); } } } encoded } #[cfg(test)] mod tests { use super::*; use std::sync::atomic::{AtomicUsize, Ordering}; fn retryable_request_error( status: Option, timeout: bool, connect: bool, transport: bool, ) -> OssError { OssError::Request(OssRequestError { status, timeout, connect, transport, oss_code: None, oss_request_id: None, operation: OssRequestOperation::Put, message: "mock request failure".to_string(), }) } fn animation_attempt_context(operation: &'static str) -> OssRequestAttemptContext { OssRequestAttemptContext { frame_index: 1, operation, } } #[test] fn oss_error_kind_is_stable_for_adapter_mapping() { assert_eq!( OssError::InvalidConfig("bad config".to_string()).kind(), OssErrorKind::InvalidConfig ); assert_eq!( OssError::ObjectNotFound("missing".to_string()).kind(), OssErrorKind::ObjectNotFound ); assert_eq!( OssError::Request(OssRequestError { status: None, timeout: false, connect: false, transport: true, oss_code: None, oss_request_id: None, operation: OssRequestOperation::Put, message: "network".to_string(), }) .kind(), OssErrorKind::Request ); } #[test] fn object_not_found_attempt_log_details_keep_404_status() { let error = OssError::ObjectNotFound("missing".to_string()); assert_eq!( request_error_details(Some(&error)), (Some(404), false, false, false, None, None) ); assert_eq!(oss_error_kind_label(&error), "object_not_found"); assert!(!oss_error_is_retryable(&error)); } #[test] fn oss_request_timeout_400_is_retryable_and_prefers_header_request_id() { let body = br#" RequestTimeoutxml-request-id"#; let error = request_status_error_from_oss_parts( OssRequestOperation::Put, 400, Some("header-request-id".to_string()), body, None, ); let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!(request_error.status, Some(400)); assert!(request_error.timeout); assert_eq!(request_error.oss_code.as_deref(), Some("RequestTimeout")); assert_eq!( request_error.oss_request_id.as_deref(), Some("header-request-id") ); assert!(oss_error_is_retryable(&error)); } #[test] fn oss_request_timeout_400_uses_xml_request_id_when_header_is_missing() { let body = br#" RequestTimeoutxml-request-id "#; let error = request_status_error_from_oss_parts(OssRequestOperation::Put, 400, None, body, None); let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!( request_error.oss_request_id.as_deref(), Some("xml-request-id") ); assert!(oss_error_is_retryable(&error)); } #[test] fn other_oss_400_errors_and_malformed_xml_are_not_retryable() { for body in [ b"InvalidArgument".as_slice(), b"RequestTimeout".as_slice(), b"not xml".as_slice(), ] { let error = request_status_error_from_oss_parts( OssRequestOperation::Put, 400, None, body, None, ); assert!(!oss_error_is_retryable(&error)); } let error = request_status_error_from_oss_parts( OssRequestOperation::Put, 403, None, b"RequestTimeout", None, ); assert!(!oss_error_is_retryable(&error)); } #[test] fn oss_error_xml_fields_beyond_body_limit_are_ignored() { let mut body = vec![b' '; CHARACTER_ANIMATION_OSS_ERROR_BODY_MAX_BYTES]; body.extend_from_slice( b"RequestTimeoutlate", ); let error = request_status_error_from_oss_parts(OssRequestOperation::Put, 400, None, &body, None); let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!(request_error.oss_code, None); assert_eq!(request_error.oss_request_id, None); assert!(!request_error.timeout); assert!(!oss_error_is_retryable(&error)); } #[test] fn oss_400_without_code_and_broken_body_read_is_retryable() { for (read_timeout, expect_transport) in [(true, false), (false, true)] { let error = request_status_error_from_oss_parts( OssRequestOperation::Put, 400, None, b"Request", Some(OssErrorBodyReadFailure { timeout: read_timeout, }), ); let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!(request_error.status, Some(400)); assert_eq!(request_error.oss_code, None); assert_eq!(request_error.timeout, read_timeout); assert_eq!(request_error.transport, expect_transport); assert!(request_error.message.contains("错误响应体读取失败")); assert!(oss_error_is_retryable(&error)); } } #[test] fn oss_400_with_parsed_code_keeps_deterministic_semantics_on_read_failure() { let error = request_status_error_from_oss_parts( OssRequestOperation::Put, 400, None, b"InvalidArgumentpartial", Some(OssErrorBodyReadFailure { timeout: true }), ); let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!(request_error.oss_code.as_deref(), Some("InvalidArgument")); assert!(!request_error.timeout); assert!(!request_error.transport); assert!(request_error.message.contains("错误响应体读取失败")); assert!(!oss_error_is_retryable(&error)); let error = request_status_error_from_oss_parts( OssRequestOperation::Put, 400, None, b"RequestTimeout", Some(OssErrorBodyReadFailure { timeout: false }), ); assert!(oss_error_is_retryable(&error)); } const MOCK_PUT_BODY: &[u8] = b"animation-frame-bytes"; /// 极简 HTTP/1.1 mock:读完整个 PUT 请求后返回 400 与部分 XML 响应体 /// (Content-Length 大于实际发送字节),`stall_before_close` 决定挂住 /// 连接触发客户端读超时,还是直接断开触发传输错误。 async fn spawn_broken_error_body_server(stall_before_close: bool) -> std::net::SocketAddr { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("mock server should bind"); let addr = listener .local_addr() .expect("mock server should expose its addr"); tokio::spawn(async move { let Ok((mut socket, _)) = listener.accept().await else { return; }; let mut received = Vec::new(); let mut buffer = [0u8; 4096]; while !received.ends_with(MOCK_PUT_BODY) { match socket.read(&mut buffer).await { Ok(0) | Err(_) => return, Ok(read) => received.extend_from_slice(&buffer[..read]), } } let response = "HTTP/1.1 400 Bad Request\r\n\ x-oss-request-id: mock-request-id\r\n\ Content-Type: application/xml\r\n\ Content-Length: 4096\r\n\ \r\n\ Request"; if socket.write_all(response.as_bytes()).await.is_err() { return; } let _ = socket.flush().await; if stall_before_close { tokio::time::sleep(std::time::Duration::from_secs(5)).await; } }); addr } #[tokio::test] async fn oss_400_with_broken_error_body_stream_is_retryable_transport() { let addr = spawn_broken_error_body_server(false).await; let response = reqwest::Client::new() .put(format!("http://{addr}/generated-animations/frame01.png")) .body(MOCK_PUT_BODY.to_vec()) .send() .await .expect("response headers should arrive before the body breaks"); assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); let error = request_status_error_from_oss_put_response(response).await; let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!(request_error.status, Some(400)); assert_eq!(request_error.oss_code, None); assert!(!request_error.timeout); assert!(request_error.transport); assert_eq!( request_error.oss_request_id.as_deref(), Some("mock-request-id") ); assert!(request_error.message.contains("错误响应体读取失败")); assert!(oss_error_is_retryable(&error)); } #[tokio::test] async fn oss_400_with_stalled_error_body_stream_is_retryable_timeout() { let addr = spawn_broken_error_body_server(true).await; let client = reqwest::Client::builder() .timeout(std::time::Duration::from_millis(300)) .build() .expect("test client should build"); let response = client .put(format!("http://{addr}/generated-animations/frame01.png")) .body(MOCK_PUT_BODY.to_vec()) .send() .await .expect("response headers should arrive before the body stalls"); assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); let error = request_status_error_from_oss_put_response(response).await; let OssError::Request(request_error) = &error else { panic!("OSS status failure should remain a request error"); }; assert_eq!(request_error.status, Some(400)); assert_eq!(request_error.oss_code, None); assert!(request_error.timeout); assert!(!request_error.transport); assert!(oss_error_is_retryable(&error)); } #[tokio::test] async fn reqwest_builder_error_is_not_retryable_transport() { let error = reqwest::Client::new() .put("https://example.com") .header("x-oss-meta-invalid", "first line\nsecond line") .send() .await .expect_err("invalid header must fail while building the request"); assert!(error.is_builder()); let error = request_error_from_reqwest(OssRequestOperation::Put, error); let OssError::Request(request_error) = &error else { panic!("builder failure should remain an OSS request error"); }; assert_eq!(request_error.status, None); assert!(!request_error.timeout); assert!(!request_error.connect); assert!(!request_error.transport); assert!(!oss_error_is_retryable(&error)); } #[tokio::test] async fn animation_retry_retries_transport_then_succeeds() { let attempts = Arc::new(AtomicUsize::new(0)); let attempts_for_request = attempts.clone(); let result = run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("source_put"), "generated-animations/editor/layer/task/green-screen-frame01.png", move || { let attempts = attempts_for_request.clone(); async move { if attempts.fetch_add(1, Ordering::SeqCst) == 0 { Err(retryable_request_error(None, false, false, true)) } else { Ok(("uploaded", 201)) } } }, ) .await; assert_eq!(result.expect("second attempt should succeed"), "uploaded"); assert_eq!(attempts.load(Ordering::SeqCst), 2); } #[tokio::test] async fn animation_retry_retries_timeout_and_retryable_statuses() { for (status, timeout, connect, transport) in [ (None, true, false, false), (Some(408), false, false, false), (Some(429), false, false, false), (Some(500), false, false, false), (Some(502), false, false, false), (Some(503), false, false, false), (Some(504), false, false, false), ] { let attempts = Arc::new(AtomicUsize::new(0)); let attempts_for_request = attempts.clone(); let result = run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("final_put"), "generated-animations/editor/layer/task/frame01.png", move || { let attempts = attempts_for_request.clone(); async move { if attempts.fetch_add(1, Ordering::SeqCst) == 0 { Err(retryable_request_error(status, timeout, connect, transport)) } else { Ok(((), 204)) } } }, ) .await; assert!(result.is_ok(), "status={status:?} should retry"); assert_eq!(attempts.load(Ordering::SeqCst), 2, "status={status:?}"); } } #[tokio::test] async fn animation_retry_does_not_retry_deterministic_statuses() { for status in [400, 403, 404] { let attempts = Arc::new(AtomicUsize::new(0)); let attempts_for_request = attempts.clone(); let result = run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("final_head"), "generated-animations/editor/layer/task/frame01.png", move || { let attempts = attempts_for_request.clone(); async move { attempts.fetch_add(1, Ordering::SeqCst); Err::<((), u16), _>(retryable_request_error( Some(status), false, false, false, )) } }, ) .await; assert!(result.is_err(), "status={status} should fail"); assert_eq!(attempts.load(Ordering::SeqCst), 1, "status={status}"); } } #[tokio::test] async fn animation_retry_returns_final_error_after_three_attempts() { let attempts = Arc::new(AtomicUsize::new(0)); let attempts_for_request = attempts.clone(); let result = run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("final_put"), "generated-animations/editor/layer/task/frame01.png", move || { let attempts = attempts_for_request.clone(); async move { attempts.fetch_add(1, Ordering::SeqCst); Err::<((), u16), _>(retryable_request_error(Some(503), false, false, false)) } }, ) .await; assert!(result.is_err()); assert_eq!(attempts.load(Ordering::SeqCst), 3); } #[tokio::test] async fn animation_retry_keeps_head_retry_independent_from_successful_put() { let put_attempts = Arc::new(AtomicUsize::new(0)); let put_attempts_for_request = put_attempts.clone(); run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("final_put"), "generated-animations/editor/layer/task/frame01.png", move || { put_attempts_for_request.fetch_add(1, Ordering::SeqCst); async { Ok::<_, OssError>(((), 204)) } }, ) .await .expect("PUT should succeed once"); let head_attempts = Arc::new(AtomicUsize::new(0)); let head_attempts_for_request = head_attempts.clone(); run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("final_head"), "generated-animations/editor/layer/task/frame01.png", move || { let head_attempts = head_attempts_for_request.clone(); async move { if head_attempts.fetch_add(1, Ordering::SeqCst) == 0 { Err(retryable_request_error(Some(503), false, false, false)) } else { Ok(((), 204)) } } }, ) .await .expect("HEAD should succeed on its retry"); assert_eq!(put_attempts.load(Ordering::SeqCst), 1); assert_eq!(head_attempts.load(Ordering::SeqCst), 2); } #[tokio::test] async fn animation_retry_keeps_permit_bound_at_eight_in_flight_requests() { let limiter = Arc::new(Semaphore::new(8)); let current = Arc::new(AtomicUsize::new(0)); let maximum = Arc::new(AtomicUsize::new(0)); let mut tasks = Vec::new(); for frame_index in 0..32 { let limiter = limiter.clone(); let current_for_request = current.clone(); let maximum_for_request = maximum.clone(); tasks.push(tokio::spawn(async move { run_animation_request_with_retry( limiter, OssRequestAttemptContext { frame_index: frame_index + 1, operation: "source_put", }, "generated-animations/editor/layer/task/green-screen-frame01.png", move || { let current = current_for_request.clone(); let maximum = maximum_for_request.clone(); async move { let in_flight = current.fetch_add(1, Ordering::SeqCst) + 1; maximum.fetch_max(in_flight, Ordering::SeqCst); tokio::time::sleep(std::time::Duration::from_millis(5)).await; current.fetch_sub(1, Ordering::SeqCst); Ok::<_, OssError>(((), 204)) } }, ) .await })); } for task in tasks { task.await .expect("mock request task should join") .expect("request should pass"); } assert!(maximum.load(Ordering::SeqCst) <= 8); } #[tokio::test] async fn animation_retry_reuses_object_key_and_body_across_attempts() { let client = OssClient::new( OssConfig::new( "bucket".to_string(), "oss-cn-shanghai.aliyuncs.com".to_string(), "access-key".to_string(), "access-secret".to_string(), 60, 60, 1024, 204, ) .expect("test OSS config should be valid"), ); let prepared = client .prepare_put_object(OssPutObjectRequest { prefix: LegacyAssetPrefix::Animations, path_segments: vec![ "editor".to_string(), "layer".to_string(), "task".to_string(), ], file_name: "frame01.png".to_string(), content_type: Some("image/png".to_string()), access: OssObjectAccess::Private, metadata: BTreeMap::new(), body: vec![1, 2, 3, 4], }) .expect("test request should be prepared"); let expected_key = prepared.object_key.clone(); let expected_body = prepared.body.clone(); let expected_key_for_request = expected_key.clone(); let attempts = Arc::new(AtomicUsize::new(0)); let attempts_for_request = attempts.clone(); let result = run_animation_request_with_retry( Arc::new(Semaphore::new(8)), animation_attempt_context("final_put"), &expected_key_for_request, move || { let attempts = attempts_for_request.clone(); let key = prepared.object_key.clone(); let body = prepared.body.clone(); let expected_key = expected_key.clone(); let expected_body = expected_body.clone(); async move { assert_eq!(key, expected_key); assert_eq!(body, expected_body); if attempts.fetch_add(1, Ordering::SeqCst) == 0 { Err(retryable_request_error(Some(503), false, false, false)) } else { Ok(((), 204)) } } }, ) .await; assert!(result.is_ok()); assert_eq!(attempts.load(Ordering::SeqCst), 2); } #[test] fn structured_log_labels_are_stable() { assert_eq!( oss_error_kind_label(&OssError::InvalidRequest("bad input".to_string())), "invalid_request" ); assert_eq!(oss_access_label(OssObjectAccess::Private), "private"); assert_eq!(http_status_class(204), "2xx"); assert_eq!(http_status_class(404), "4xx"); assert_eq!(http_status_class_from_option(None), "unknown"); } fn build_client() -> OssClient { OssClient::new( OssConfig::new( "genarrative-assets".to_string(), "oss-cn-shanghai.aliyuncs.com".to_string(), "test-access-key-id".to_string(), "test-access-key-secret".to_string(), DEFAULT_READ_EXPIRE_SECONDS, DEFAULT_POST_EXPIRE_SECONDS, DEFAULT_POST_MAX_SIZE_BYTES, DEFAULT_SUCCESS_ACTION_STATUS, ) .expect("OSS config should be valid"), ) } #[test] fn parse_legacy_prefix_accepts_public_style_path() { assert_eq!( LegacyAssetPrefix::parse("/generated-characters/*"), Some(LegacyAssetPrefix::Characters) ); assert_eq!( LegacyAssetPrefix::parse("/generated-puzzle-assets/*"), Some(LegacyAssetPrefix::PuzzleAssets) ); assert_eq!( LegacyAssetPrefix::parse("/generated-match3d-assets/*"), Some(LegacyAssetPrefix::Match3DAssets) ); assert_eq!( LegacyAssetPrefix::parse("/generated-wooden-fish-assets/*"), Some(LegacyAssetPrefix::WoodenFishAssets) ); assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-puzzle-assets")); assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-match3d-assets")); assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-bark-battle-assets")); assert!(LEGACY_PUBLIC_PREFIXES.contains(&"generated-wooden-fish-assets")); assert_eq!(LegacyAssetPrefix::parse("unknown"), None); } #[test] fn build_v4_signature_date_zero_pads_single_digit_time_parts() { let signed_at = OffsetDateTime::from_unix_timestamp(1_771_477_389).expect("timestamp should be valid"); assert_eq!( build_v4_signature_date(signed_at).expect("date should format"), "20260219T050309Z" ); assert_eq!( build_v4_signature_scope("oss-cn-shanghai.aliyuncs.com", signed_at) .expect("scope should format"), "20260219/cn-shanghai/oss/aliyun_v4_request" ); } #[test] fn sign_post_object_returns_bucket_and_object_key_for_private_storage_truth() { let client = build_client(); let mut metadata = BTreeMap::new(); metadata.insert("asset-kind".to_string(), "character-visual".to_string()); metadata.insert("origin".to_string(), "browser-upload".to_string()); let response = client .sign_post_object(OssPostObjectRequest { prefix: LegacyAssetPrefix::Characters, path_segments: vec![ "Hero_001".to_string(), "Visual".to_string(), "Asset_01".to_string(), ], file_name: "Master.PNG".to_string(), content_type: Some("image/png".to_string()), access: OssObjectAccess::Public, metadata, max_size_bytes: Some(5 * 1024 * 1024), expire_seconds: Some(300), success_action_status: Some(200), }) .expect("post object signature should build"); assert_eq!( response.object_key, "generated-characters/hero_001/visual/asset_01/master.png" ); assert_eq!( response.legacy_public_path, "/generated-characters/hero_001/visual/asset_01/master.png" ); assert_eq!(response.bucket, "genarrative-assets".to_string()); assert_eq!( response.form_fields.signature_version, OSS_V4_ALGORITHM.to_string() ); assert!( response .form_fields .credential .starts_with("test-access-key-id/") ); assert!( response .form_fields .credential .ends_with("/cn-shanghai/oss/aliyun_v4_request") ); assert_eq!(response.form_fields.date.len(), "20260507T120000Z".len()); assert_eq!( response.form_fields.metadata.get("x-oss-meta-asset-kind"), Some(&"character-visual".to_string()) ); } #[test] fn sign_post_object_embeds_policy_constraints() { let client = build_client(); let response = client .sign_post_object(OssPostObjectRequest { prefix: LegacyAssetPrefix::QwenSprites, path_segments: vec!["_drafts".to_string(), "master".to_string()], file_name: "candidate-01.png".to_string(), content_type: Some("image/png".to_string()), access: OssObjectAccess::Private, metadata: BTreeMap::new(), max_size_bytes: Some(1024), expire_seconds: Some(60), success_action_status: Some(200), }) .expect("post object signature should build"); let decoded_policy = BASE64_STANDARD .decode(response.form_fields.policy.as_bytes()) .expect("policy should be valid base64"); let policy: Value = serde_json::from_slice(&decoded_policy).expect("policy should be valid json"); assert_eq!( policy["conditions"][0]["bucket"], Value::String("genarrative-assets".to_string()) ); assert_eq!( policy["conditions"][1], json!([ "eq", "$key", "generated-qwen-sprites/_drafts/master/candidate-01.png" ]) ); assert_eq!( policy["conditions"][2], json!(["content-length-range", 1, 1024]) ); assert_eq!( policy["conditions"][3], json!(["eq", "$success_action_status", "200"]) ); assert_eq!( policy["conditions"][4], json!(["eq", "$x-oss-signature-version", "OSS4-HMAC-SHA256"]) ); assert_eq!( policy["conditions"][5], json!(["eq", "$x-oss-credential", response.form_fields.credential]) ); assert_eq!( policy["conditions"][6], json!(["eq", "$x-oss-date", response.form_fields.date]) ); assert_eq!( policy["conditions"][7], json!(["eq", "$content-type", "image/png"]) ); assert_eq!( policy["conditions"][8], json!(["eq", "$Cache-Control", DEFAULT_IMMUTABLE_CACHE_CONTROL]) ); assert_eq!( response.form_fields.cache_control, Some(DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string()) ); assert_eq!(response.bucket, "genarrative-assets".to_string()); } #[test] fn sanitize_file_name_rejects_empty_input() { let error = sanitize_file_name(" ").expect_err("empty file name should fail"); assert_eq!( error, OssError::InvalidRequest("fileName 不能为空".to_string()) ); } #[test] fn sanitize_file_name_falls_back_when_stem_has_no_ascii_body() { assert_eq!( sanitize_file_name("剪贴板素材.png").expect("file name should sanitize"), "file.png" ); } #[test] fn sign_get_object_url_returns_signed_private_read_url() { let client = build_client(); let response = client .sign_get_object_url(OssSignedGetObjectUrlRequest { object_key: "generated-characters/hero_001/visual/asset_01/master.png".to_string(), expire_seconds: Some(300), }) .expect("signed get url should build"); assert_eq!(response.bucket, "genarrative-assets".to_string()); assert_eq!( response.object_key, "generated-characters/hero_001/visual/asset_01/master.png".to_string() ); assert!(response .signed_url .starts_with("https://genarrative-assets.oss-cn-shanghai.aliyuncs.com/generated-characters/hero_001/visual/asset_01/master.png?")); assert!( response .signed_url .contains("x-oss-signature-version=OSS4-HMAC-SHA256") ); assert!( response .signed_url .contains("x-oss-credential=test-access-key-id%2F") ); assert!(response.signed_url.contains("&x-oss-expires=300")); assert!(response.signed_url.contains("&x-oss-signature=")); } #[test] fn sign_get_object_url_uses_square_hole_object_key_without_bucket_prefix() { let client = OssClient::new( OssConfig::new( "xushi-dev".to_string(), "oss-cn-shanghai.aliyuncs.com".to_string(), "test-access-key-id".to_string(), "test-access-key-secret".to_string(), DEFAULT_READ_EXPIRE_SECONDS, DEFAULT_POST_EXPIRE_SECONDS, DEFAULT_POST_MAX_SIZE_BYTES, DEFAULT_SUCCESS_ACTION_STATUS, ) .expect("OSS config should be valid"), ); let response = client .sign_get_object_url(OssSignedGetObjectUrlRequest { object_key: "generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png".to_string(), expire_seconds: Some(300), }) .expect("square hole object key should build signed url"); assert_eq!(response.bucket, "xushi-dev".to_string()); assert_eq!( response.object_key, "generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png".to_string() ); assert!(response .signed_url .starts_with("https://xushi-dev.oss-cn-shanghai.aliyuncs.com/generated-square-hole-assets/square-hole-session-546d881972684be2980a2a882cd0cc71/square-hole-profile-134411276ce1469cbe398f946a25d7f8/square-hole-shape-image/rabbit-option/asset-1777979289912039/image.png?")); } #[test] fn sign_get_object_url_rejects_unsupported_prefix() { let client = build_client(); let error = client .sign_get_object_url(OssSignedGetObjectUrlRequest { object_key: "workflow-cache/task-1.json".to_string(), expire_seconds: Some(300), }) .expect_err("unsupported prefix should fail"); assert_eq!( error, OssError::InvalidRequest("objectKey 必须落在受支持的 OSS 前缀下".to_string()) ); } #[test] fn sign_internal_get_object_url_allows_editor_agent_messages_document() { let client = build_client(); let response = client .sign_internal_get_object_url(OssSignedGetObjectUrlRequest { object_key: "editor-agent/editor-agent-conv-1.json".to_string(), expire_seconds: Some(300), }) .expect("editor agent messages document should build signed url"); assert_eq!( response.object_key, "editor-agent/editor-agent-conv-1.json".to_string() ); assert!(response.signed_url.starts_with( "https://genarrative-assets.oss-cn-shanghai.aliyuncs.com/editor-agent/editor-agent-conv-1.json?" )); } #[test] fn sign_internal_get_object_url_rejects_non_editor_agent_messages_document() { let client = build_client(); for object_key in [ "workflow-cache/task-1.json", "generated-characters/hero/master.png", "editor-agent/editor-agent-conv-1/part.json", "editor-agent/other.json", "editor-agent/editor-agent-conv-1.png", ] { let error = client .sign_internal_get_object_url(OssSignedGetObjectUrlRequest { object_key: object_key.to_string(), expire_seconds: Some(300), }) .expect_err("non editor agent messages document should fail"); assert_eq!( error, OssError::InvalidRequest("objectKey 必须是 editor-agent 消息文档".to_string()) ); } } #[test] fn legacy_prefix_can_be_resolved_from_object_key() { assert_eq!( LegacyAssetPrefix::from_object_key( "generated-custom-world-scenes/profile_01/landmark_01/scene.png" ), Some(LegacyAssetPrefix::CustomWorldScenes) ); assert_eq!( LegacyAssetPrefix::from_object_key( "generated-wooden-fish-assets/session/profile/hit_object/asset/image.png" ), Some(LegacyAssetPrefix::WoodenFishAssets) ); assert_eq!( LegacyAssetPrefix::from_object_key("workflow-cache/demo.json"), None ); assert_eq!( LegacyAssetPrefix::from_object_key( "agc/project-snapshots/v2/dev/user-1/project-1/manifest.json" ), None, "AGC 内部前缀不能经由通用对象键解析变成客户端可写前缀" ); } #[test] fn head_object_content_length_reads_the_response_header_not_the_head_body() { let mut headers = reqwest::header::HeaderMap::new(); headers.insert( reqwest::header::CONTENT_LENGTH, "32".parse().expect("content length header value"), ); assert_eq!( head_object_content_length(&headers), 32, "HEAD 的对象大小必须来自响应头;reqwest 对 HEAD 的 body 长度恒为 0" ); assert_eq!( head_object_content_length(&reqwest::header::HeaderMap::new()), 0 ); } #[test] fn agc_project_snapshot_object_keys_preserve_case_and_reject_traversal() { let file_key = agc_project_snapshot_file_object_key( "dev", "user-1", "gameagent-1a2b3c4d", 1234, "0123456789ABCDEF", "Game/Scenes/Main.HTML", ) .expect("file key"); assert_eq!( file_key, "agc/project-snapshots/v2/dev/user-1/gameagent-1a2b3c4d/files/1234-0123456789abcdef/Game/Scenes/Main.HTML" ); assert_eq!( agc_project_snapshot_manifest_object_key("release", "user-1", "gameagent-1a2b3c4d") .expect("manifest key"), "agc/project-snapshots/v2/release/user-1/gameagent-1a2b3c4d/manifest.json" ); for (user, project, path) in [ ("../escape", "project-1", "game/index.html"), ("user-1", "../escape", "game/index.html"), ("user-1", "project-1", "../outside.txt"), ("user-1", "project-1", "game/../../outside.txt"), ("user-1", "project-1", "game\\index.html"), ] { assert!( agc_project_snapshot_file_object_key("dev", user, project, 1, "abcdef", path) .is_err(), "越界键片段必须被拒绝:{user} {project} {path}" ); } assert!( agc_project_snapshot_file_object_key( "dev", "user-1", "project-1", 1, "not-hex", "game/a.txt" ) .is_err(), "摘要必须是十六进制" ); } /// 渠道是对象键的第一层:非法渠道必须失败关闭,不能悄悄换成一个默认渠道。 #[test] fn agc_project_snapshot_channel_is_validated_before_it_reaches_the_key() { assert_eq!( validate_agc_project_snapshot_channel("release").expect("channel"), "release" ); assert_eq!( validate_agc_project_snapshot_channel("dev-internal-2").expect("channel"), "dev-internal-2" ); for invalid in [ "", " dev", "dev ", "Dev", "dev_internal", "dev/internal", "-dev", "dev-", "2dev", "dev.", &"d".repeat(33), ] { assert!( validate_agc_project_snapshot_channel(invalid).is_err(), "非法渠道必须被拒绝:{invalid}" ); assert!( agc_project_snapshot_manifest_object_key(invalid, "user-1", "project-1").is_err(), "非法渠道不能进入对象键:{invalid}" ); } } #[test] fn internal_object_prefixes_cover_agc_snapshots_but_reject_everything_else() { let file_key = agc_project_snapshot_file_object_key( "dev", "user-1", "project-1", 7, "abcdef", "game/index.html", ) .expect("file key"); assert_eq!( normalize_internal_object_key(&file_key).expect("snapshot key is internal"), file_key ); // 无渠道的历史项目快照对象仍然必须保持服务端私有。 assert_eq!( normalize_internal_object_key( "agc/project-snapshots/v1/user-1/project-1/manifest.json" ) .expect("legacy snapshot key stays internal"), "agc/project-snapshots/v1/user-1/project-1/manifest.json" ); assert_eq!( normalize_internal_object_key("agc/error-reports/v1/batch.zip") .expect("error report key stays internal"), "agc/error-reports/v1/batch.zip" ); for key in [ "generated-characters/hero/master.png", "agc/other-purpose/v1/file.json", ] { assert!( normalize_internal_object_key(key).is_err(), "非内部前缀不能走服务端内部写入:{key}" ); } } #[test] fn put_object_request_reuses_generated_object_key_contract() { let request = OssPutObjectRequest { prefix: LegacyAssetPrefix::CustomWorldCovers, path_segments: vec!["Profile 001".to_string(), "asset_01".to_string()], file_name: "Cover.PNG".to_string(), content_type: Some(" image/png ".to_string()), access: OssObjectAccess::Private, metadata: BTreeMap::from([ ("asset_kind".to_string(), "custom_world_cover".to_string()), ("source job id".to_string(), "job_001".to_string()), ]), body: b"cover-bytes".to_vec(), }; let sanitized_segments = request .path_segments .iter() .map(|segment| sanitize_path_segment(segment)) .filter(|segment| !segment.is_empty()) .collect::>(); let file_name = sanitize_file_name(&request.file_name).expect("file name should sanitize"); let object_key = build_object_key(request.prefix, &sanitized_segments, &file_name); let metadata = normalize_metadata(request.metadata).expect("metadata should normalize"); assert_eq!( object_key, "generated-custom-world-covers/profile-001/asset_01/cover.png" ); assert_eq!( metadata.get("x-oss-meta-asset-kind"), Some(&"custom_world_cover".to_string()) ); assert_eq!( metadata.get("x-oss-meta-source-job-id"), Some(&"job_001".to_string()) ); } #[test] fn canonicalized_oss_headers_matches_sorted_v4_header_shape() { let headers = BTreeMap::from([ ( "Cache-Control".to_string(), DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(), ), ( "x-oss-meta-source-job-id".to_string(), " job_001 ".to_string(), ), ( "x-oss-meta-asset-kind".to_string(), "character_visual".to_string(), ), ]); assert_eq!( build_v4_canonical_headers(&headers), "cache-control:public, max-age=31536000, immutable\nx-oss-meta-asset-kind:character_visual\nx-oss-meta-source-job-id:job_001\n" ); } #[test] fn additional_headers_include_plain_headers_and_skip_oss_managed_headers() { let headers = BTreeMap::from([ ( "host".to_string(), "genarrative-assets.oss-cn-beijing.aliyuncs.com".to_string(), ), ("content-type".to_string(), "image/png".to_string()), ( "Cache-Control".to_string(), DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string(), ), ("x-oss-date".to_string(), "20260507T120000Z".to_string()), ( "x-oss-meta-asset-kind".to_string(), "puzzle-cover".to_string(), ), ]); assert_eq!(build_v4_additional_headers(&headers), "cache-control;host"); } #[test] fn put_object_headers_include_immutable_cache_control_for_generated_assets() { let headers = build_put_object_headers(BTreeMap::from([( "asset-kind".to_string(), "puzzle-cover".to_string(), )])) .expect("headers should build"); assert_eq!( headers.get("Cache-Control"), Some(&DEFAULT_IMMUTABLE_CACHE_CONTROL.to_string()) ); assert_eq!( headers.get("x-oss-meta-asset-kind"), Some(&"puzzle-cover".to_string()) ); } #[tokio::test] async fn put_object_rejects_empty_body_before_calling_oss() { let client = build_client(); let error = client .put_object( &reqwest::Client::new(), OssPutObjectRequest { prefix: LegacyAssetPrefix::Characters, path_segments: vec!["hero".to_string()], file_name: "master.png".to_string(), content_type: Some("image/png".to_string()), access: OssObjectAccess::Private, metadata: BTreeMap::new(), body: Vec::new(), }, ) .await .expect_err("empty server upload should fail before network"); assert_eq!( error, OssError::InvalidRequest("服务端上传对象内容不能为空".to_string()) ); } }