mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-08-03 11:01:22 +08:00
feat: 新增 S3 兼容云存储同步 (#1351)
* Add S3 Cloud Sync design document Design for adding AWS S3 as a new Cloud Sync backend alongside WebDAV. Hybrid approach: extract shared sync protocol, add independent S3 transport. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Add S3 cloud sync implementation design (reqwest + Sig V4) Updated design based on 2026-03-06 draft: switches from rust-s3 crate to hand-rolled AWS Sig V4 on existing reqwest for broader S3-compatible service support (AWS, MinIO, R2, Alibaba OSS, Tencent COS, Huawei OBS). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Add S3 cloud sync implementation plan (11 tasks, TDD) Detailed step-by-step plan covering: sync_protocol extraction, S3 Sig V4 transport, settings, sync/auto-sync modules, Tauri commands, frontend presets/dynamic form, and i18n. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * deps: add hmac crate for S3 Sig V4 signing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * refactor: extract sync_protocol.rs from webdav_sync.rs for shared use Move transport-agnostic sync protocol logic (constants, types, snapshot building, manifest validation, artifact verification, snapshot application, utilities) into a new shared sync_protocol module so both WebDAV and the upcoming S3 transport can reuse it. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: use transport-neutral error keys in sync_protocol Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 transport layer with AWS Sig V4 signing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3SyncSettings to AppSettings Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 sync module with upload/download/fetch Implements the S3 sync protocol layer (s3_sync.rs) that combines the shared sync_protocol with the S3 transport. Mirrors the WebDAV sync module structure with independent sync mutex, connection check, upload, download, fetch_remote_info, and sync status persistence. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 auto sync worker with debounce Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 sync Tauri commands and auto sync worker startup Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 sync TypeScript types and API layer Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 sync i18n translations (en/zh/ja) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add S3 sync presets and dynamic form to sync settings Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: preserve HTTP scheme for S3 custom endpoints (MinIO support) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * test: add live S3 integration tests (env-var driven, --ignored) Run with: S3_TEST_AK=... S3_TEST_SK=... S3_TEST_BUCKET=... cargo test --lib services::s3::integration_tests -- --ignored Verifies test_connection, put_object, get_object, head_object, and 404 handling against a real S3 bucket using the project's own Sig V4 signing. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * chore: remove internal design docs before PR * fix: wire S3 auto-sync to DB hook & sync UI state on async load - P1: Add s3_auto_sync::notify_db_changed call in SQLite update_hook so S3 auto-sync worker receives DB change signals (was only wired for WebDAV, leaving S3 worker idle) - P2: Add useEffect to update syncType selector when s3Config loads asynchronously, preventing stale "webdav" default for S3 users * fix: satisfy clippy for s3 sync * fix: address s3 sync review feedback --------- Co-authored-by: Keith (via OpenClaw) <keithyt06@users.noreply.github.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com> Co-authored-by: Jason <farion1231@gmail.com>
This commit is contained in:
Keith Yu
@@ -0,0 +1,926 @@
|
||||
//! S3 HTTP transport layer.
|
||||
//!
|
||||
//! Low-level HTTP primitives for S3 operations (PUT, GET, HEAD).
|
||||
//! Implements AWS Signature Version 4 request signing.
|
||||
//! The sync protocol logic lives in the upcoming `s3_sync` module.
|
||||
|
||||
use reqwest::StatusCode;
|
||||
use std::time::Duration;
|
||||
use url::Url;
|
||||
|
||||
use crate::error::AppError;
|
||||
use crate::proxy::http_client;
|
||||
use futures::StreamExt;
|
||||
|
||||
const DEFAULT_TIMEOUT_SECS: u64 = 30;
|
||||
/// Timeout for large file transfers (PUT / GET of db.sql, skills.zip, etc.).
|
||||
const TRANSFER_TIMEOUT_SECS: u64 = 300;
|
||||
|
||||
// ─── Credentials ─────────────────────────────────────────────
|
||||
|
||||
/// S3-compatible storage credentials.
|
||||
pub(crate) struct S3Credentials {
|
||||
pub access_key_id: String,
|
||||
pub secret_access_key: String,
|
||||
pub region: String,
|
||||
pub bucket: String,
|
||||
/// Custom endpoint host (e.g. `minio.example.com:9000`).
|
||||
/// Empty string means AWS official endpoint.
|
||||
pub endpoint: String,
|
||||
}
|
||||
|
||||
// ─── URL construction ────────────────────────────────────────
|
||||
|
||||
/// Returns `true` for AWS official endpoints (empty or contains `amazonaws.com`).
|
||||
fn is_aws_endpoint(endpoint: &str) -> bool {
|
||||
endpoint.is_empty() || endpoint.contains("amazonaws.com")
|
||||
}
|
||||
|
||||
/// Split an endpoint into its scheme and host-with-port parts.
|
||||
///
|
||||
/// Preserves the original scheme when one is provided, defaulting to `"https"`
|
||||
/// when the endpoint is bare (no `://` prefix).
|
||||
///
|
||||
/// ```text
|
||||
/// "http://minio:9000" → ("http", "minio:9000")
|
||||
/// "https://storage.example.com" → ("https", "storage.example.com")
|
||||
/// "minio:9000" → ("https", "minio:9000")
|
||||
/// "storage.example.com" → ("https", "storage.example.com")
|
||||
/// ```
|
||||
fn split_scheme_host(endpoint: &str) -> (&str, &str) {
|
||||
if let Some(rest) = endpoint.strip_prefix("http://") {
|
||||
("http", rest.trim_end_matches('/'))
|
||||
} else if let Some(rest) = endpoint.strip_prefix("https://") {
|
||||
("https", rest.trim_end_matches('/'))
|
||||
} else {
|
||||
("https", endpoint.trim_end_matches('/'))
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the full URL for an S3 object.
|
||||
///
|
||||
/// - AWS endpoints use virtual-hosted style: `https://{bucket}.s3.{region}.amazonaws.com/{key}`
|
||||
/// - Custom endpoints use path style: `https://{endpoint}/{bucket}/{key}`
|
||||
fn build_object_url(creds: &S3Credentials, key: &str) -> String {
|
||||
let key = key.trim_start_matches('/');
|
||||
if is_aws_endpoint(&creds.endpoint) {
|
||||
format!(
|
||||
"https://{}.s3.{}.amazonaws.com/{}",
|
||||
creds.bucket, creds.region, key
|
||||
)
|
||||
} else {
|
||||
let (scheme, host) = split_scheme_host(&creds.endpoint);
|
||||
format!("{}://{}/{}/{}", scheme, host, creds.bucket, key)
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the bucket-level URL (for HEAD bucket / test connection).
|
||||
fn build_bucket_url(creds: &S3Credentials) -> String {
|
||||
if is_aws_endpoint(&creds.endpoint) {
|
||||
format!(
|
||||
"https://{}.s3.{}.amazonaws.com/",
|
||||
creds.bucket, creds.region
|
||||
)
|
||||
} else {
|
||||
let (scheme, host) = split_scheme_host(&creds.endpoint);
|
||||
format!("{}://{}/{}/", scheme, host, creds.bucket)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Cryptographic helpers ───────────────────────────────────
|
||||
|
||||
fn hmac_sha256(key: &[u8], data: &[u8]) -> Vec<u8> {
|
||||
use hmac::{Hmac, Mac};
|
||||
type HmacSha256 = Hmac<sha2::Sha256>;
|
||||
let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts any key length");
|
||||
mac.update(data);
|
||||
mac.finalize().into_bytes().to_vec()
|
||||
}
|
||||
|
||||
fn sha256_hex(data: &[u8]) -> String {
|
||||
use sha2::{Digest, Sha256};
|
||||
format!("{:x}", Sha256::digest(data))
|
||||
}
|
||||
|
||||
/// Percent-encode following AWS Sig V4 rules (RFC 3986 unreserved characters only).
|
||||
fn uri_encode(input: &str, encode_slash: bool) -> String {
|
||||
let mut out = String::with_capacity(input.len());
|
||||
for byte in input.bytes() {
|
||||
match byte {
|
||||
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
|
||||
out.push(byte as char);
|
||||
}
|
||||
b'/' if !encode_slash => out.push('/'),
|
||||
_ => {
|
||||
use std::fmt::Write;
|
||||
let _ = write!(out, "%{:02X}", byte);
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
// ─── AWS Signature V4 signing ────────────────────────────────
|
||||
|
||||
/// Sign an HTTP request using AWS Signature Version 4.
|
||||
///
|
||||
/// Mutates `headers` by adding `host`, `x-amz-date`, `x-amz-content-sha256`,
|
||||
/// and the final `authorization` header.
|
||||
fn sign_request(
|
||||
method: &str,
|
||||
url: &Url,
|
||||
headers: &mut reqwest::header::HeaderMap,
|
||||
body_hash: &str,
|
||||
creds: &S3Credentials,
|
||||
now: chrono::DateTime<chrono::Utc>,
|
||||
) {
|
||||
let timestamp = now.format("%Y%m%dT%H%M%SZ").to_string();
|
||||
let datestamp = now.format("%Y%m%d").to_string();
|
||||
|
||||
// ── Step 1: Add required headers ──
|
||||
let host_value = match url.port() {
|
||||
Some(port) => format!("{}:{}", url.host_str().unwrap_or_default(), port),
|
||||
None => url.host_str().unwrap_or_default().to_string(),
|
||||
};
|
||||
headers.insert("host", host_value.parse().unwrap());
|
||||
headers.insert("x-amz-date", timestamp.parse().unwrap());
|
||||
headers.insert("x-amz-content-sha256", body_hash.parse().unwrap());
|
||||
|
||||
// ── Step 2: Build canonical request ──
|
||||
|
||||
// Canonical URI (already percent-encoded by the url crate).
|
||||
let canonical_uri = url.path();
|
||||
let canonical_uri = if canonical_uri.is_empty() {
|
||||
"/"
|
||||
} else {
|
||||
canonical_uri
|
||||
};
|
||||
|
||||
// Canonical query string — sorted, re-encoded per Sig V4 rules.
|
||||
let mut query_pairs: Vec<(String, String)> = url
|
||||
.query_pairs()
|
||||
.map(|(k, v)| (k.into_owned(), v.into_owned()))
|
||||
.collect();
|
||||
query_pairs.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1)));
|
||||
let canonical_query = if query_pairs.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
query_pairs
|
||||
.iter()
|
||||
.map(|(k, v)| format!("{}={}", uri_encode(k, true), uri_encode(v, true)))
|
||||
.collect::<Vec<_>>()
|
||||
.join("&")
|
||||
};
|
||||
|
||||
// Canonical headers — sorted by lowercase name.
|
||||
let mut header_names: Vec<String> = headers.keys().map(|k| k.as_str().to_lowercase()).collect();
|
||||
header_names.sort();
|
||||
header_names.dedup();
|
||||
|
||||
let canonical_headers: String = header_names
|
||||
.iter()
|
||||
.map(|name| {
|
||||
let value = headers
|
||||
.get(name.as_str())
|
||||
.map(|v| v.to_str().unwrap_or("").trim().to_string())
|
||||
.unwrap_or_default();
|
||||
format!("{}:{}\n", name, value)
|
||||
})
|
||||
.collect();
|
||||
|
||||
let signed_headers = header_names.join(";");
|
||||
|
||||
let canonical_request = format!(
|
||||
"{}\n{}\n{}\n{}\n{}\n{}",
|
||||
method, canonical_uri, canonical_query, canonical_headers, signed_headers, body_hash
|
||||
);
|
||||
|
||||
// ── Step 3: Build string to sign ──
|
||||
let scope = format!("{}/{}/s3/aws4_request", datestamp, creds.region);
|
||||
let string_to_sign = format!(
|
||||
"AWS4-HMAC-SHA256\n{}\n{}\n{}",
|
||||
timestamp,
|
||||
scope,
|
||||
sha256_hex(canonical_request.as_bytes())
|
||||
);
|
||||
|
||||
// ── Step 4: Derive signing key ──
|
||||
// HMAC(HMAC(HMAC(HMAC("AWS4"+secret, date), region), "s3"), "aws4_request")
|
||||
let k_date = hmac_sha256(
|
||||
format!("AWS4{}", creds.secret_access_key).as_bytes(),
|
||||
datestamp.as_bytes(),
|
||||
);
|
||||
let k_region = hmac_sha256(&k_date, creds.region.as_bytes());
|
||||
let k_service = hmac_sha256(&k_region, b"s3");
|
||||
let k_signing = hmac_sha256(&k_service, b"aws4_request");
|
||||
|
||||
// ── Step 5: Compute signature ──
|
||||
let sig_bytes = hmac_sha256(&k_signing, string_to_sign.as_bytes());
|
||||
let signature: String = sig_bytes.iter().map(|b| format!("{:02x}", b)).collect();
|
||||
|
||||
// ── Step 6: Add Authorization header ──
|
||||
let authorization = format!(
|
||||
"AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
|
||||
creds.access_key_id, scope, signed_headers, signature
|
||||
);
|
||||
headers.insert("authorization", authorization.parse().unwrap());
|
||||
}
|
||||
|
||||
// ─── Error helpers ───────────────────────────────────────────
|
||||
|
||||
/// Redact a URL for safe inclusion in error messages (strips query parameters).
|
||||
fn redact_url(raw: &str) -> String {
|
||||
match Url::parse(raw) {
|
||||
Ok(parsed) => {
|
||||
let mut out = format!("{}://", parsed.scheme());
|
||||
if let Some(host) = parsed.host_str() {
|
||||
out.push_str(host);
|
||||
}
|
||||
if let Some(port) = parsed.port() {
|
||||
out.push(':');
|
||||
out.push_str(&port.to_string());
|
||||
}
|
||||
out.push_str(parsed.path());
|
||||
out
|
||||
}
|
||||
Err(_) => raw.split('?').next().unwrap_or(raw).to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn s3_transport_error(
|
||||
key: &'static str,
|
||||
op_zh: &str,
|
||||
op_en: &str,
|
||||
err: &reqwest::Error,
|
||||
) -> AppError {
|
||||
let (zh_reason, en_reason) = if err.is_timeout() {
|
||||
("请求超时", "request timed out")
|
||||
} else if err.is_connect() {
|
||||
("连接失败", "connection failed")
|
||||
} else if err.is_request() {
|
||||
("请求构造失败", "request build failed")
|
||||
} else {
|
||||
("网络请求失败", "network request failed")
|
||||
};
|
||||
|
||||
AppError::localized(
|
||||
key,
|
||||
format!("S3 {op_zh}失败（{zh_reason}）"),
|
||||
format!("S3 {op_en} failed ({en_reason})"),
|
||||
)
|
||||
}
|
||||
|
||||
fn s3_status_error(op: &str, status: StatusCode, url: &str) -> AppError {
|
||||
let safe_url = redact_url(url);
|
||||
let mut zh = format!("S3 {op} 失败: {status} ({safe_url})");
|
||||
let mut en = format!("S3 {op} failed: {status} ({safe_url})");
|
||||
|
||||
if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
|
||||
zh.push_str("。请检查 Access Key ID 和 Secret Access Key。");
|
||||
en.push_str(". Please verify your Access Key ID and Secret Access Key.");
|
||||
} else if status == StatusCode::NOT_FOUND && op == "HEAD bucket" {
|
||||
zh.push_str("。请检查存储桶名称和区域是否正确。");
|
||||
en.push_str(". Please check the bucket name and region.");
|
||||
}
|
||||
|
||||
AppError::localized("s3.http.status", zh, en)
|
||||
}
|
||||
|
||||
fn response_too_large_error(url: &str, max_bytes: usize) -> AppError {
|
||||
let max_mb = max_bytes / 1024 / 1024;
|
||||
AppError::localized(
|
||||
"s3.response_too_large",
|
||||
format!("S3 响应体超过上限（{} MB）: {}", max_mb, redact_url(url)),
|
||||
format!(
|
||||
"S3 response body exceeds limit ({} MB): {}",
|
||||
max_mb,
|
||||
redact_url(url)
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
fn ensure_content_length_within_limit(
|
||||
headers: &reqwest::header::HeaderMap,
|
||||
max_bytes: usize,
|
||||
url: &str,
|
||||
) -> Result<(), AppError> {
|
||||
let Some(cl) = headers.get(reqwest::header::CONTENT_LENGTH) else {
|
||||
return Ok(());
|
||||
};
|
||||
let Ok(raw) = cl.to_str() else {
|
||||
return Ok(());
|
||||
};
|
||||
let Ok(value) = raw.parse::<u64>() else {
|
||||
return Ok(());
|
||||
};
|
||||
if value > max_bytes as u64 {
|
||||
return Err(response_too_large_error(url, max_bytes));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ─── Transport functions ─────────────────────────────────────
|
||||
|
||||
/// Test S3 connectivity by sending HEAD to the bucket root.
|
||||
pub(crate) async fn test_connection(creds: &S3Credentials) -> Result<(), AppError> {
|
||||
let url_str = build_bucket_url(creds);
|
||||
let url = Url::parse(&url_str).map_err(|e| {
|
||||
AppError::localized(
|
||||
"s3.url.invalid",
|
||||
format!("S3 URL 无效: {e}"),
|
||||
format!("Invalid S3 URL: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let client = http_client::get();
|
||||
let body_hash = sha256_hex(b"");
|
||||
let mut headers = reqwest::header::HeaderMap::new();
|
||||
sign_request(
|
||||
"HEAD",
|
||||
&url,
|
||||
&mut headers,
|
||||
&body_hash,
|
||||
creds,
|
||||
chrono::Utc::now(),
|
||||
);
|
||||
|
||||
let resp = client
|
||||
.head(url.as_str())
|
||||
.headers(headers)
|
||||
.timeout(Duration::from_secs(DEFAULT_TIMEOUT_SECS))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| s3_transport_error("s3.connection_failed", "连接", "connection", &e))?;
|
||||
|
||||
if resp.status().is_success() {
|
||||
return Ok(());
|
||||
}
|
||||
Err(s3_status_error("HEAD bucket", resp.status(), &url_str))
|
||||
}
|
||||
|
||||
/// Upload bytes to an S3 object.
|
||||
pub(crate) async fn put_object(
|
||||
creds: &S3Credentials,
|
||||
key: &str,
|
||||
bytes: Vec<u8>,
|
||||
content_type: &str,
|
||||
) -> Result<(), AppError> {
|
||||
let url_str = build_object_url(creds, key);
|
||||
let url = Url::parse(&url_str).map_err(|e| {
|
||||
AppError::localized(
|
||||
"s3.url.invalid",
|
||||
format!("S3 URL 无效: {e}"),
|
||||
format!("Invalid S3 URL: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let client = http_client::get();
|
||||
let body_hash = sha256_hex(&bytes);
|
||||
let mut headers = reqwest::header::HeaderMap::new();
|
||||
headers.insert("content-type", content_type.parse().unwrap());
|
||||
sign_request(
|
||||
"PUT",
|
||||
&url,
|
||||
&mut headers,
|
||||
&body_hash,
|
||||
creds,
|
||||
chrono::Utc::now(),
|
||||
);
|
||||
|
||||
let resp = client
|
||||
.put(url.as_str())
|
||||
.headers(headers)
|
||||
.body(bytes)
|
||||
.timeout(Duration::from_secs(TRANSFER_TIMEOUT_SECS))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| s3_transport_error("s3.put_failed", "PUT 请求", "PUT request", &e))?;
|
||||
|
||||
if resp.status().is_success() {
|
||||
return Ok(());
|
||||
}
|
||||
Err(s3_status_error("PUT", resp.status(), &url_str))
|
||||
}
|
||||
|
||||
/// Download an S3 object. Returns `None` if the object does not exist (404).
|
||||
///
|
||||
/// On success returns `(body_bytes, optional_etag)`.
|
||||
pub(crate) async fn get_object(
|
||||
creds: &S3Credentials,
|
||||
key: &str,
|
||||
max_bytes: usize,
|
||||
) -> Result<Option<(Vec<u8>, Option<String>)>, AppError> {
|
||||
let url_str = build_object_url(creds, key);
|
||||
let url = Url::parse(&url_str).map_err(|e| {
|
||||
AppError::localized(
|
||||
"s3.url.invalid",
|
||||
format!("S3 URL 无效: {e}"),
|
||||
format!("Invalid S3 URL: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let client = http_client::get();
|
||||
let body_hash = sha256_hex(b"");
|
||||
let mut headers = reqwest::header::HeaderMap::new();
|
||||
sign_request(
|
||||
"GET",
|
||||
&url,
|
||||
&mut headers,
|
||||
&body_hash,
|
||||
creds,
|
||||
chrono::Utc::now(),
|
||||
);
|
||||
|
||||
let resp = client
|
||||
.get(url.as_str())
|
||||
.headers(headers)
|
||||
.timeout(Duration::from_secs(TRANSFER_TIMEOUT_SECS))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| s3_transport_error("s3.get_failed", "GET 请求", "GET request", &e))?;
|
||||
|
||||
if resp.status() == StatusCode::NOT_FOUND {
|
||||
return Ok(None);
|
||||
}
|
||||
if !resp.status().is_success() {
|
||||
return Err(s3_status_error("GET", resp.status(), &url_str));
|
||||
}
|
||||
ensure_content_length_within_limit(resp.headers(), max_bytes, &url_str)?;
|
||||
|
||||
let etag = resp
|
||||
.headers()
|
||||
.get("etag")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string());
|
||||
|
||||
let mut bytes = Vec::new();
|
||||
let mut stream = resp.bytes_stream();
|
||||
while let Some(chunk) = stream.next().await {
|
||||
let chunk = chunk.map_err(|e| {
|
||||
AppError::localized(
|
||||
"s3.response_read_failed",
|
||||
format!("读取 S3 响应失败: {e}"),
|
||||
format!("Failed to read S3 response: {e}"),
|
||||
)
|
||||
})?;
|
||||
if bytes.len().saturating_add(chunk.len()) > max_bytes {
|
||||
return Err(response_too_large_error(&url_str, max_bytes));
|
||||
}
|
||||
bytes.extend_from_slice(&chunk);
|
||||
}
|
||||
Ok(Some((bytes, etag)))
|
||||
}
|
||||
|
||||
/// Retrieve the ETag of an S3 object via HEAD. Returns `None` on 404.
|
||||
pub(crate) async fn head_object(
|
||||
creds: &S3Credentials,
|
||||
key: &str,
|
||||
) -> Result<Option<String>, AppError> {
|
||||
let url_str = build_object_url(creds, key);
|
||||
let url = Url::parse(&url_str).map_err(|e| {
|
||||
AppError::localized(
|
||||
"s3.url.invalid",
|
||||
format!("S3 URL 无效: {e}"),
|
||||
format!("Invalid S3 URL: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let client = http_client::get();
|
||||
let body_hash = sha256_hex(b"");
|
||||
let mut headers = reqwest::header::HeaderMap::new();
|
||||
sign_request(
|
||||
"HEAD",
|
||||
&url,
|
||||
&mut headers,
|
||||
&body_hash,
|
||||
creds,
|
||||
chrono::Utc::now(),
|
||||
);
|
||||
|
||||
let resp = client
|
||||
.head(url.as_str())
|
||||
.headers(headers)
|
||||
.timeout(Duration::from_secs(DEFAULT_TIMEOUT_SECS))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| s3_transport_error("s3.head_failed", "HEAD 请求", "HEAD request", &e))?;
|
||||
|
||||
if resp.status() == StatusCode::NOT_FOUND {
|
||||
return Ok(None);
|
||||
}
|
||||
if !resp.status().is_success() {
|
||||
return Err(s3_status_error("HEAD", resp.status(), &url_str));
|
||||
}
|
||||
Ok(resp
|
||||
.headers()
|
||||
.get("etag")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string()))
|
||||
}
|
||||
|
||||
// ─── Tests ───────────────────────────────────────────────────
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use chrono::TimeZone;
|
||||
|
||||
// ── Crypto helpers ──
|
||||
|
||||
#[test]
|
||||
fn sha256_hex_empty_body() {
|
||||
assert_eq!(
|
||||
sha256_hex(b""),
|
||||
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sha256_hex_known_value() {
|
||||
// From the AWS S3 documentation PUT Object example body.
|
||||
assert_eq!(
|
||||
sha256_hex(b"Welcome to Amazon S3."),
|
||||
"44ce7dd67c959e0d3524ffac1771dfbba87d2b6b4b4e99e42034a8b803f8b072"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hmac_sha256_rfc2104_test_vector() {
|
||||
// HMAC-SHA256("key", "The quick brown fox jumps over the lazy dog")
|
||||
let result = hmac_sha256(b"key", b"The quick brown fox jumps over the lazy dog");
|
||||
let hex: String = result.iter().map(|b| format!("{:02x}", b)).collect();
|
||||
assert_eq!(
|
||||
hex,
|
||||
"f7bc83f430538424b13298e6aa6fb143ef4d59a14946175997479dbc2d1a3cd8"
|
||||
);
|
||||
}
|
||||
|
||||
// ── URL construction: virtual-hosted vs path-style ──
|
||||
|
||||
#[test]
|
||||
fn build_object_url_virtual_hosted_style_aws() {
|
||||
let creds = test_creds("", "us-east-1", "mybucket");
|
||||
assert_eq!(
|
||||
build_object_url(&creds, "path/to/file.json"),
|
||||
"https://mybucket.s3.us-east-1.amazonaws.com/path/to/file.json"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_virtual_hosted_explicit_aws_endpoint() {
|
||||
let creds = test_creds(
|
||||
"s3.ap-northeast-1.amazonaws.com",
|
||||
"ap-northeast-1",
|
||||
"mybucket",
|
||||
);
|
||||
assert_eq!(
|
||||
build_object_url(&creds, "key.txt"),
|
||||
"https://mybucket.s3.ap-northeast-1.amazonaws.com/key.txt"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_path_style_custom_endpoint() {
|
||||
let creds = test_creds("minio.example.com:9000", "us-east-1", "mybucket");
|
||||
assert_eq!(
|
||||
build_object_url(&creds, "path/to/file.json"),
|
||||
"https://minio.example.com:9000/mybucket/path/to/file.json"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_strips_leading_slash_from_key() {
|
||||
let creds = test_creds("", "us-east-1", "b");
|
||||
assert_eq!(
|
||||
build_object_url(&creds, "/leading/slash.txt"),
|
||||
"https://b.s3.us-east-1.amazonaws.com/leading/slash.txt"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_bucket_url_aws() {
|
||||
let creds = test_creds("", "us-west-2", "testbucket");
|
||||
assert_eq!(
|
||||
build_bucket_url(&creds),
|
||||
"https://testbucket.s3.us-west-2.amazonaws.com/"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_bucket_url_custom_endpoint() {
|
||||
let creds = test_creds("storage.example.com", "us-east-1", "data");
|
||||
assert_eq!(
|
||||
build_bucket_url(&creds),
|
||||
"https://storage.example.com/data/"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_endpoint_with_trailing_slash() {
|
||||
let creds = test_creds("minio.local:9000/", "us-east-1", "b");
|
||||
assert_eq!(
|
||||
build_object_url(&creds, "k"),
|
||||
"https://minio.local:9000/b/k"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_endpoint_with_scheme_prefix() {
|
||||
let creds = test_creds("https://minio.local:9000", "us-east-1", "b");
|
||||
assert_eq!(
|
||||
build_object_url(&creds, "k"),
|
||||
"https://minio.local:9000/b/k"
|
||||
);
|
||||
}
|
||||
|
||||
// ── HTTP scheme preservation (MinIO support) ──
|
||||
|
||||
#[test]
|
||||
fn build_object_url_preserves_http_scheme() {
|
||||
let creds = test_creds("http://minio:9000", "us-east-1", "mybucket");
|
||||
let url = build_object_url(&creds, "path/to/file.json");
|
||||
assert!(
|
||||
url.starts_with("http://"),
|
||||
"expected http:// scheme, got: {url}"
|
||||
);
|
||||
assert_eq!(url, "http://minio:9000/mybucket/path/to/file.json");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_preserves_https_scheme() {
|
||||
let creds = test_creds("https://storage.example.com", "us-east-1", "mybucket");
|
||||
let url = build_object_url(&creds, "path/to/file.json");
|
||||
assert!(
|
||||
url.starts_with("https://"),
|
||||
"expected https:// scheme, got: {url}"
|
||||
);
|
||||
assert_eq!(
|
||||
url,
|
||||
"https://storage.example.com/mybucket/path/to/file.json"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_object_url_bare_endpoint_defaults_to_https() {
|
||||
let creds = test_creds("minio:9000", "us-east-1", "mybucket");
|
||||
let url = build_object_url(&creds, "path/to/file.json");
|
||||
assert!(
|
||||
url.starts_with("https://"),
|
||||
"bare endpoint should default to https://, got: {url}"
|
||||
);
|
||||
assert_eq!(url, "https://minio:9000/mybucket/path/to/file.json");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_bucket_url_preserves_http_scheme() {
|
||||
let creds = test_creds("http://minio:9000", "us-east-1", "data");
|
||||
let url = build_bucket_url(&creds);
|
||||
assert!(
|
||||
url.starts_with("http://"),
|
||||
"expected http:// scheme, got: {url}"
|
||||
);
|
||||
assert_eq!(url, "http://minio:9000/data/");
|
||||
}
|
||||
|
||||
// ── Endpoint detection ──
|
||||
|
||||
#[test]
|
||||
fn is_aws_endpoint_detection() {
|
||||
assert!(is_aws_endpoint(""), "empty string = AWS");
|
||||
assert!(is_aws_endpoint("s3.us-east-1.amazonaws.com"));
|
||||
assert!(is_aws_endpoint("s3.amazonaws.com"));
|
||||
assert!(!is_aws_endpoint("minio.example.com"));
|
||||
assert!(!is_aws_endpoint("storage.googleapis.com"));
|
||||
assert!(!is_aws_endpoint("r2.cloudflarestorage.com"));
|
||||
}
|
||||
|
||||
// ── URI encoding ──
|
||||
|
||||
#[test]
|
||||
fn uri_encode_preserves_unreserved_chars() {
|
||||
assert_eq!(
|
||||
uri_encode("hello-world_test.txt~v2", true),
|
||||
"hello-world_test.txt~v2"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn uri_encode_encodes_spaces_and_special_chars() {
|
||||
assert_eq!(uri_encode("hello world", true), "hello%20world");
|
||||
assert_eq!(uri_encode("a+b=c&d", true), "a%2Bb%3Dc%26d");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn uri_encode_slash_handling() {
|
||||
assert_eq!(uri_encode("path/to/file", false), "path/to/file");
|
||||
assert_eq!(uri_encode("path/to/file", true), "path%2Fto%2Ffile");
|
||||
}
|
||||
|
||||
// ── AWS Signature V4 signing ──
|
||||
|
||||
#[test]
|
||||
fn sig_v4_signing_against_aws_test_vector() {
|
||||
// Based on the AWS documentation example: GET /?lifecycle
|
||||
// https://docs.aws.amazon.com/AmazonS3/latest/API/sig-v4-header-based-auth.html
|
||||
let creds = S3Credentials {
|
||||
access_key_id: "AKIAIOSFODNN7EXAMPLE".to_string(),
|
||||
secret_access_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_string(),
|
||||
region: "us-east-1".to_string(),
|
||||
bucket: "examplebucket".to_string(),
|
||||
endpoint: String::new(),
|
||||
};
|
||||
|
||||
let now = chrono::Utc.with_ymd_and_hms(2013, 5, 24, 0, 0, 0).unwrap();
|
||||
let url = Url::parse("https://examplebucket.s3.amazonaws.com/?lifecycle").unwrap();
|
||||
let body_hash = sha256_hex(b"");
|
||||
|
||||
let mut headers = reqwest::header::HeaderMap::new();
|
||||
sign_request("GET", &url, &mut headers, &body_hash, &creds, now);
|
||||
|
||||
let auth = headers.get("authorization").unwrap().to_str().unwrap();
|
||||
|
||||
// Verify credential scope
|
||||
assert!(
|
||||
auth.starts_with(
|
||||
"AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request"
|
||||
),
|
||||
"unexpected credential scope in: {auth}"
|
||||
);
|
||||
// Verify signed headers
|
||||
assert!(
|
||||
auth.contains("SignedHeaders=host;x-amz-content-sha256;x-amz-date"),
|
||||
"unexpected signed headers in: {auth}"
|
||||
);
|
||||
// Verify exact signature from AWS documentation
|
||||
assert!(
|
||||
auth.contains(
|
||||
"Signature=fea454ca298b7da1c68078a5d1bdbfbbe0d65c699e0f91ac7a200a0136783543"
|
||||
),
|
||||
"signature mismatch in: {auth}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sig_v4_includes_content_type_when_present() {
|
||||
let creds = S3Credentials {
|
||||
access_key_id: "TESTKEY".to_string(),
|
||||
secret_access_key: "TESTSECRET".to_string(),
|
||||
region: "us-east-1".to_string(),
|
||||
bucket: "b".to_string(),
|
||||
endpoint: String::new(),
|
||||
};
|
||||
|
||||
let now = chrono::Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap();
|
||||
let url = Url::parse("https://b.s3.us-east-1.amazonaws.com/key.json").unwrap();
|
||||
let body_hash = sha256_hex(b"{}");
|
||||
|
||||
let mut headers = reqwest::header::HeaderMap::new();
|
||||
headers.insert("content-type", "application/json".parse().unwrap());
|
||||
sign_request("PUT", &url, &mut headers, &body_hash, &creds, now);
|
||||
|
||||
let auth = headers.get("authorization").unwrap().to_str().unwrap();
|
||||
// content-type must appear in the signed headers
|
||||
assert!(
|
||||
auth.contains("content-type"),
|
||||
"content-type should be in signed headers: {auth}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sig_v4_signing_key_derivation() {
|
||||
// Verify the signing key derivation chain independently.
|
||||
// Using the same AWS example credentials and date.
|
||||
let secret = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY";
|
||||
let datestamp = "20130524";
|
||||
let region = "us-east-1";
|
||||
|
||||
let k_date = hmac_sha256(format!("AWS4{}", secret).as_bytes(), datestamp.as_bytes());
|
||||
let k_region = hmac_sha256(&k_date, region.as_bytes());
|
||||
let k_service = hmac_sha256(&k_region, b"s3");
|
||||
let k_signing = hmac_sha256(&k_service, b"aws4_request");
|
||||
|
||||
// The signing key should be a 32-byte value (256 bits).
|
||||
assert_eq!(k_signing.len(), 32);
|
||||
|
||||
// Verify it is deterministic — computing again yields the same result.
|
||||
let k_date2 = hmac_sha256(format!("AWS4{}", secret).as_bytes(), datestamp.as_bytes());
|
||||
let k_region2 = hmac_sha256(&k_date2, region.as_bytes());
|
||||
let k_service2 = hmac_sha256(&k_region2, b"s3");
|
||||
let k_signing2 = hmac_sha256(&k_service2, b"aws4_request");
|
||||
assert_eq!(k_signing, k_signing2);
|
||||
}
|
||||
|
||||
// ── Redact URL ──
|
||||
|
||||
#[test]
|
||||
fn redact_url_strips_query_params() {
|
||||
let r = redact_url(
|
||||
"https://mybucket.s3.us-east-1.amazonaws.com/file.txt?X-Amz-Credential=AKID&X-Amz-Signature=abc",
|
||||
);
|
||||
assert!(!r.contains("AKID"));
|
||||
assert!(!r.contains("abc"));
|
||||
assert!(r.contains("mybucket.s3.us-east-1.amazonaws.com"));
|
||||
assert!(r.contains("/file.txt"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn redact_url_preserves_path() {
|
||||
let r = redact_url("https://minio.local:9000/bucket/path/to/file.json");
|
||||
assert_eq!(r, "https://minio.local:9000/bucket/path/to/file.json");
|
||||
}
|
||||
|
||||
// ── Content-length check ──
|
||||
|
||||
#[test]
|
||||
fn ensure_content_length_within_limit_rejects_oversized() {
|
||||
use reqwest::header::{HeaderMap, HeaderValue, CONTENT_LENGTH};
|
||||
let mut h = HeaderMap::new();
|
||||
h.insert(CONTENT_LENGTH, HeaderValue::from_static("2048"));
|
||||
assert!(ensure_content_length_within_limit(&h, 1024, "https://example.com").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ensure_content_length_within_limit_accepts_within_bounds() {
|
||||
use reqwest::header::{HeaderMap, HeaderValue, CONTENT_LENGTH};
|
||||
let mut h = HeaderMap::new();
|
||||
h.insert(CONTENT_LENGTH, HeaderValue::from_static("512"));
|
||||
assert!(ensure_content_length_within_limit(&h, 1024, "https://example.com").is_ok());
|
||||
|
||||
let empty = HeaderMap::new();
|
||||
assert!(ensure_content_length_within_limit(&empty, 1024, "https://example.com").is_ok());
|
||||
}
|
||||
|
||||
// ── Helper ──
|
||||
|
||||
fn test_creds(endpoint: &str, region: &str, bucket: &str) -> S3Credentials {
|
||||
S3Credentials {
|
||||
access_key_id: String::new(),
|
||||
secret_access_key: String::new(),
|
||||
region: region.to_string(),
|
||||
bucket: bucket.to_string(),
|
||||
endpoint: endpoint.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Live integration tests (run with --ignored) ─────────────
|
||||
|
||||
#[cfg(test)]
|
||||
mod integration_tests {
|
||||
use super::*;
|
||||
|
||||
fn test_creds() -> S3Credentials {
|
||||
S3Credentials {
|
||||
access_key_id: std::env::var("S3_TEST_AK").expect("S3_TEST_AK env required"),
|
||||
secret_access_key: std::env::var("S3_TEST_SK").expect("S3_TEST_SK env required"),
|
||||
region: std::env::var("S3_TEST_REGION").unwrap_or_else(|_| "us-east-1".to_string()),
|
||||
bucket: std::env::var("S3_TEST_BUCKET").expect("S3_TEST_BUCKET env required"),
|
||||
endpoint: std::env::var("S3_TEST_ENDPOINT").unwrap_or_default(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
async fn live_s3_connection() {
|
||||
crate::proxy::http_client::init(None).ok();
|
||||
let creds = test_creds();
|
||||
let result = test_connection(&creds).await;
|
||||
assert!(result.is_ok(), "Connection failed: {:?}", result.err());
|
||||
println!("PASS: test_connection OK");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
async fn live_s3_put_get_head_roundtrip() {
|
||||
crate::proxy::http_client::init(None).ok();
|
||||
let creds = test_creds();
|
||||
let key = "cc-switch-sync/v2/default/_integration_test.json";
|
||||
let data = br#"{"test":true,"ts":12345}"#;
|
||||
|
||||
// PUT
|
||||
let r = put_object(&creds, key, data.to_vec(), "application/json").await;
|
||||
assert!(r.is_ok(), "PUT failed: {:?}", r.err());
|
||||
println!("PASS: put_object {} bytes", data.len());
|
||||
|
||||
// GET
|
||||
let r = get_object(&creds, key, 1 << 20).await;
|
||||
assert!(r.is_ok(), "GET failed: {:?}", r.err());
|
||||
let (body, etag) = r.unwrap().expect("should exist");
|
||||
assert_eq!(body, data);
|
||||
println!("PASS: get_object {} bytes, etag={:?}", body.len(), etag);
|
||||
|
||||
// HEAD
|
||||
let r = head_object(&creds, key).await;
|
||||
assert!(r.is_ok(), "HEAD failed: {:?}", r.err());
|
||||
assert!(r.unwrap().is_some());
|
||||
println!("PASS: head_object OK");
|
||||
|
||||
// 404
|
||||
let r = get_object(&creds, "cc-switch-sync/_no_such_key", 1024).await;
|
||||
assert!(r.is_ok());
|
||||
assert!(r.unwrap().is_none());
|
||||
println!("PASS: get_object(404) returned None");
|
||||
|
||||
println!("ALL LIVE S3 TESTS PASSED");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user