diff --git a/.cargo/audit.toml b/.cargo/audit.toml index 01d324d..a9c3770 100644 --- a/.cargo/audit.toml +++ b/.cargo/audit.toml @@ -1,4 +1,5 @@ -# SQLx locks optional MySQL dependencies even though this project builds only -# the PostgreSQL driver. The vulnerable RSA code is absent from `cargo tree`. +# Reviewed 2026-07-25. SQLx locks optional MySQL dependencies even though this +# project builds only the PostgreSQL driver; the vulnerable RSA code is absent +# from `cargo tree`. [advisories] ignore = ["RUSTSEC-2023-0071"] diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 17f4512..55b8287 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -12,6 +12,21 @@ jobs: - name: Checkout uses: actions/checkout@v4 + - name: Verify migration line endings + run: | + python3 - <<'PY' + from pathlib import Path + + invalid = [] + for path in sorted(Path("migrations").glob("*.sql")): + data = path.read_bytes() + if b"\n" in data.replace(b"\r\n", b""): + invalid.append(str(path)) + + if invalid: + raise SystemExit("migrations must use CRLF: " + ", ".join(invalid)) + PY + - name: Install Rust toolchain uses: dtolnay/rust-toolchain@stable with: diff --git a/Cargo.toml b/Cargo.toml index fa76e20..5e563de 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,7 @@ rust-version = "1.92" [dependencies] axum = { version = "0.8", features = ["multipart"] } tokio = { version = "1", features = ["full"] } -tower-http = { version = "0.6", features = ["cors", "trace", "compression-full", "fs"] } +tower-http = { version = "0.6", features = ["trace", "compression-full", "fs"] } axum-extra = { version = "0.12", features = ["cookie"] } time = "0.3.47" diff --git a/migrations/012_task_query_indexes.sql b/migrations/012_task_query_indexes.sql new file mode 100644 index 0000000..9cbe738 --- /dev/null +++ b/migrations/012_task_query_indexes.sql @@ -0,0 +1,8 @@ +CREATE INDEX IF NOT EXISTS idx_tasks_user_created_at + ON tasks(user_id, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_tasks_user_status_created_at + ON tasks(user_id, status, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_tasks_status_created_at + ON tasks(status, created_at DESC); diff --git a/src/api/admin.rs b/src/api/admin.rs index 4b6a2c5..eda5b1c 100644 --- a/src/api/admin.rs +++ b/src/api/admin.rs @@ -431,13 +431,10 @@ async fn list_tasks( let limit = query.limit.unwrap_or(20).clamp(1, 100); let page = query.page.unwrap_or(1).max(1); let offset = (page - 1) * limit; - let status = query - .status - .map(|s| s.trim().to_string()) - .filter(|s| !s.is_empty()); + let status = super::normalize_task_status(query.status.as_deref())?; - let total: i64 = if let Some(status) = &status { - sqlx::query_scalar("SELECT COUNT(*) FROM tasks WHERE status::text = $1") + let total: i64 = if let Some(status) = status { + sqlx::query_scalar("SELECT COUNT(*) FROM tasks WHERE status = $1::task_status") .bind(status) .fetch_one(&state.db) .await @@ -449,7 +446,7 @@ async fn list_tasks( .map_err(|err| AppError::new(ErrorCode::Internal, "查询任务失败").with_source(err))? }; - let tasks: Vec = if let Some(status) = &status { + let tasks: Vec = if let Some(status) = status { sqlx::query_as::<_, AdminTaskRow>( r#" SELECT @@ -467,7 +464,7 @@ async fn list_tasks( u.email AS user_email FROM tasks t LEFT JOIN users u ON u.id = t.user_id - WHERE t.status::text = $1 + WHERE t.status = $1::task_status ORDER BY t.created_at DESC LIMIT $2 OFFSET $3 "#, diff --git a/src/api/auth.rs b/src/api/auth.rs index 29654d7..9ab5c93 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -267,16 +267,19 @@ async fn login( .fetch_optional(&state.db) .await } - .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))? - .ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "账号或密码错误"))?; + .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?; - if !user.is_active { - return Err(AppError::new(ErrorCode::Forbidden, "账号已被禁用")); - } + let Some(user) = user else { + credentials::consume_dummy_password_work(&req.password).await?; + return Err(AppError::new(ErrorCode::Unauthorized, "账号或密码错误")); + }; if !credentials::verify_password(&req.password, &user.password_hash).await? { return Err(AppError::new(ErrorCode::Unauthorized, "账号或密码错误")); } + if !user.is_active { + return Err(AppError::new(ErrorCode::Forbidden, "账号已被禁用")); + } let verification_required = policy.auth.email_verification_required; let (token, expires_at) = auth::issue_jwt( diff --git a/src/api/compress.rs b/src/api/compress.rs index 7bc4996..220cd92 100644 --- a/src/api/compress.rs +++ b/src/api/compress.rs @@ -79,6 +79,40 @@ fn default_units_charged() -> i32 { 1 } +fn direct_response( + body: B, + format: ImageFmt, + data: &DirectIdempotencyData, +) -> axum::response::Response { + let mut headers = HeaderMap::new(); + headers.insert( + axum::http::header::CONTENT_TYPE, + format.content_type().parse().unwrap(), + ); + headers.insert( + "ImageForge-Original-Size", + data.original_size.to_string().parse().unwrap(), + ); + headers.insert( + "ImageForge-Compressed-Size", + data.compressed_size.to_string().parse().unwrap(), + ); + headers.insert( + "ImageForge-Saved-Bytes", + data.saved_bytes.to_string().parse().unwrap(), + ); + headers.insert( + "ImageForge-Saved-Percent", + format!("{:.2}", data.saved_percent).parse().unwrap(), + ); + headers.insert( + "ImageForge-Units-Charged", + data.units_charged.to_string().parse().unwrap(), + ); + + (StatusCode::OK, headers, body).into_response() +} + fn ensure_target_size_supported( target_size_bytes: Option, format_out: ImageFmt, @@ -492,35 +526,7 @@ async fn compress_direct( })?; let (bytes, fmt) = load_direct_replay_bytes(&state, &principal, data.file_id).await?; - - let mut resp_headers = HeaderMap::new(); - resp_headers.insert( - axum::http::header::CONTENT_TYPE, - fmt.content_type().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Original-Size", - data.original_size.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Compressed-Size", - data.compressed_size.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Saved-Bytes", - data.saved_bytes.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Saved-Percent", - format!("{:.2}", data.saved_percent).parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Units-Charged", - data.units_charged.to_string().parse().unwrap(), - ); - - let response = (StatusCode::OK, resp_headers, bytes).into_response(); - return Ok((jar, response)); + return Ok((jar, direct_response(bytes, fmt, &data))); } idempotency::BeginResult::InProgress => { if let Some((_status, body)) = @@ -533,35 +539,7 @@ async fn compress_direct( })?; let (bytes, fmt) = load_direct_replay_bytes(&state, &principal, data.file_id).await?; - - let mut resp_headers = HeaderMap::new(); - resp_headers.insert( - axum::http::header::CONTENT_TYPE, - fmt.content_type().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Original-Size", - data.original_size.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Compressed-Size", - data.compressed_size.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Saved-Bytes", - data.saved_bytes.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Saved-Percent", - format!("{:.2}", data.saved_percent).parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Units-Charged", - data.units_charged.to_string().parse().unwrap(), - ); - - let response = (StatusCode::OK, resp_headers, bytes).into_response(); - return Ok((jar, response)); + return Ok((jar, direct_response(bytes, fmt, &data))); } return Err(AppError::new( ErrorCode::InvalidRequest, @@ -663,45 +641,17 @@ async fn compress_direct( return Err(err); } - let mut resp_headers = HeaderMap::new(); - resp_headers.insert( - axum::http::header::CONTENT_TYPE, - format_out.content_type().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Original-Size", - original_size.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Compressed-Size", - compressed_size.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Saved-Bytes", - saved_bytes.to_string().parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Saved-Percent", - format!("{saved_percent:.2}").parse().unwrap(), - ); - resp_headers.insert( - "ImageForge-Units-Charged", - if charge_units { "1" } else { "0" }.parse().unwrap(), - ); - - let response = (StatusCode::OK, resp_headers, compressed).into_response(); - Ok(( - response, - DirectIdempotencyData { - file_id, - format_out: format_out.as_str().to_string(), - original_size, - compressed_size, - saved_bytes, - saved_percent, - units_charged: if charge_units { 1 } else { 0 }, - }, - )) + let idem_data = DirectIdempotencyData { + file_id, + format_out: format_out.as_str().to_string(), + original_size, + compressed_size, + saved_bytes, + saved_percent, + units_charged: if charge_units { 1 } else { 0 }, + }; + let response = direct_response(compressed, format_out, &idem_data); + Ok((response, idem_data)) }) .await; @@ -1266,3 +1216,30 @@ async fn charge_one_unit( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn direct_response_has_consistent_compression_headers() { + let data = DirectIdempotencyData { + file_id: Uuid::new_v4(), + format_out: "webp".to_string(), + original_size: 1_000, + compressed_size: 625, + saved_bytes: 375, + saved_percent: 37.5, + units_charged: 1, + }; + + let response = direct_response(bytes::Bytes::from_static(b"result"), ImageFmt::Webp, &data); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["content-type"], "image/webp"); + assert_eq!(response.headers()["imageforge-original-size"], "1000"); + assert_eq!(response.headers()["imageforge-compressed-size"], "625"); + assert_eq!(response.headers()["imageforge-saved-bytes"], "375"); + assert_eq!(response.headers()["imageforge-saved-percent"], "37.50"); + assert_eq!(response.headers()["imageforge-units-charged"], "1"); + } +} diff --git a/src/api/context.rs b/src/api/context.rs index 157ab6a..68b8af7 100644 --- a/src/api/context.rs +++ b/src/api/context.rs @@ -1,6 +1,6 @@ use crate::auth; use crate::error::{AppError, ErrorCode}; -use crate::services::{rate_limit, settings}; +use crate::services::{quota, rate_limit, settings}; use crate::state::AppState; use axum::http::HeaderMap; @@ -113,7 +113,8 @@ pub async fn authenticate( .trim() .to_ascii_lowercase() .starts_with("https://"); - let (jar, session_id) = ensure_session_cookie(jar, cookie_secure); + let (jar, session_id) = + ensure_session_cookie(jar, cookie_secure, &state.config.api_key_pepper)?; Ok((jar, Principal::Anonymous { session_id })) } @@ -142,7 +143,7 @@ pub async fn enforce_anonymous_upload_rate( rate_limit::enforce( state, "anonymous_upload_ip", - &ip.to_string(), + "a::anonymous_ip_scope(ip), limit, 60, "匿名上传请求过于频繁，请稍后再试", @@ -317,12 +318,7 @@ async fn try_api_key( }) .collect::, _>>()?; - let _ = - sqlx::query("UPDATE api_keys SET last_used_at = NOW(), last_used_ip = $2 WHERE id = $1") - .bind(row.id) - .bind(ip.to_string()) - .execute(&state.db) - .await; + touch_api_key_usage(state, row.id, ip).await; Ok(Some(Principal::ApiKey { user_id: row.user_id, @@ -333,16 +329,59 @@ async fn try_api_key( })) } -pub fn ensure_session_cookie(jar: CookieJar, secure: bool) -> (CookieJar, String) { +async fn touch_api_key_usage(state: &AppState, api_key_id: Uuid, ip: IpAddr) { + let touch_key = format!("api_key_last_used:{api_key_id}"); + let mut redis = state.redis.clone(); + let claimed: redis::RedisResult> = redis::cmd("SET") + .arg(&touch_key) + .arg("1") + .arg("EX") + .arg(60) + .arg("NX") + .query_async(&mut redis) + .await; + + let should_update = match claimed { + Ok(Some(_)) => true, + Ok(None) => false, + Err(err) => { + tracing::warn!(error = %err, api_key_id = %api_key_id, "API Key 使用时间限频失败，回退为直接落库"); + true + } + }; + if !should_update { + return; + } + + if let Err(err) = + sqlx::query("UPDATE api_keys SET last_used_at = NOW(), last_used_ip = $2 WHERE id = $1") + .bind(api_key_id) + .bind(ip.to_string()) + .execute(&state.db) + .await + { + tracing::warn!(error = %err, api_key_id = %api_key_id, "更新 API Key 最后使用信息失败"); + let _: redis::RedisResult = redis::cmd("DEL") + .arg(touch_key) + .query_async(&mut redis) + .await; + } +} + +pub fn ensure_session_cookie( + jar: CookieJar, + secure: bool, + secret: &str, +) -> Result<(CookieJar, String), AppError> { if let Some(cookie) = jar.get("if_session") { - let session_id = cookie.value().trim().to_string(); - if !session_id.is_empty() { - return (jar, session_id); + if let Some(session_id) = verify_session_cookie(cookie.value().trim(), secret)? { + return Ok((jar, session_id)); } } let session_id = generate_session_id(); - let cookie = Cookie::build(("if_session", session_id.clone())) + let signed_value = sign_session_cookie(&session_id, secret)?; + let cookie = Cookie::build(("if_session", signed_value)) .path("/") .http_only(true) .secure(secure) @@ -350,7 +389,7 @@ pub fn ensure_session_cookie(jar: CookieJar, secure: bool) -> (CookieJar, String .max_age(TimeDuration::days(7)) .build(); - (jar.add(cookie), session_id) + Ok((jar.add(cookie), session_id)) } fn generate_session_id() -> String { @@ -359,6 +398,49 @@ fn generate_session_id() -> String { URL_SAFE_NO_PAD.encode(bytes) } +fn sign_session_cookie(session_id: &str, secret: &str) -> Result { + type HmacSha256 = Hmac; + + let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).map_err(|err| { + AppError::new(ErrorCode::Internal, "匿名会话签名密钥错误").with_source(err) + })?; + mac.update(b"imageforge-anonymous-session:v1:"); + mac.update(session_id.as_bytes()); + let signature = URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes()); + Ok(format!("v1.{session_id}.{signature}")) +} + +fn verify_session_cookie(value: &str, secret: &str) -> Result, AppError> { + type HmacSha256 = Hmac; + + let mut parts = value.split('.'); + let (Some("v1"), Some(session_id), Some(signature), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return Ok(None); + }; + let Ok(session_bytes) = URL_SAFE_NO_PAD.decode(session_id) else { + return Ok(None); + }; + let Ok(signature) = URL_SAFE_NO_PAD.decode(signature) else { + return Ok(None); + }; + if session_bytes.len() != 32 || signature.len() != 32 { + return Ok(None); + } + + let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).map_err(|err| { + AppError::new(ErrorCode::Internal, "匿名会话签名密钥错误").with_source(err) + })?; + mac.update(b"imageforge-anonymous-session:v1:"); + mac.update(session_id.as_bytes()); + if mac.verify_slice(&signature).is_err() { + return Ok(None); + } + + Ok(Some(session_id.to_string())) +} + pub fn api_key_hash(full_key: &str, pepper: &str) -> Result { type HmacSha256 = Hmac; @@ -415,4 +497,28 @@ mod tests { }; assert!(require_api_permission(&user, &["billing_read"]).is_ok()); } + + #[test] + fn signed_session_cookie_round_trips_and_rejects_tampering() { + let session_id = generate_session_id(); + let signed = sign_session_cookie(&session_id, "test-secret").unwrap(); + assert_eq!( + verify_session_cookie(&signed, "test-secret").unwrap(), + Some(session_id) + ); + + let tampered = signed.replacen("v1.", "v1.A", 1); + assert_eq!( + verify_session_cookie(&tampered, "test-secret").unwrap(), + None + ); + assert_eq!( + verify_session_cookie(&signed, "other-secret").unwrap(), + None + ); + assert_eq!( + verify_session_cookie("legacy-cookie", "test-secret").unwrap(), + None + ); + } } diff --git a/src/api/mod.rs b/src/api/mod.rs index 9e128fb..8d2aaf3 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -74,3 +74,39 @@ fn v1_router() -> Router { .merge(admin_storage::router()) .fallback(response::not_found) } + +fn normalize_task_status(input: Option<&str>) -> Result, AppError> { + match input + .map(str::trim) + .unwrap_or("") + .to_ascii_lowercase() + .as_str() + { + "" | "all" => Ok(None), + "pending" => Ok(Some("pending")), + "processing" => Ok(Some("processing")), + "completed" => Ok(Some("completed")), + "failed" => Ok(Some("failed")), + "cancelled" => Ok(Some("cancelled")), + _ => Err(AppError::new( + ErrorCode::InvalidRequest, + "status 仅支持 pending/processing/completed/failed/cancelled", + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn task_status_filter_is_normalized_and_validated() { + assert_eq!(normalize_task_status(None).unwrap(), None); + assert_eq!(normalize_task_status(Some(" ALL ")).unwrap(), None); + assert_eq!( + normalize_task_status(Some("Completed")).unwrap(), + Some("completed") + ); + assert!(normalize_task_status(Some("unknown")).is_err()); + } +} diff --git a/src/api/user.rs b/src/api/user.rs index 692ce4a..1048e39 100644 --- a/src/api/user.rs +++ b/src/api/user.rs @@ -412,18 +412,17 @@ async fn list_history( let limit = query.limit.unwrap_or(20).clamp(1, 100); let page = query.page.unwrap_or(1).max(1); let offset = (page - 1) * limit; - let status = query - .status - .map(|s| s.trim().to_string()) - .filter(|s| !s.is_empty()); + let status = super::normalize_task_status(query.status.as_deref())?; - let total: i64 = if let Some(status) = &status { - sqlx::query_scalar("SELECT COUNT(*) FROM tasks WHERE user_id = $1 AND status::text = $2") - .bind(user_id) - .bind(status) - .fetch_one(&state.db) - .await - .map_err(|err| AppError::new(ErrorCode::Internal, "查询历史失败").with_source(err))? + let total: i64 = if let Some(status) = status { + sqlx::query_scalar( + "SELECT COUNT(*) FROM tasks WHERE user_id = $1 AND status = $2::task_status", + ) + .bind(user_id) + .bind(status) + .fetch_one(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询历史失败").with_source(err))? } else { sqlx::query_scalar("SELECT COUNT(*) FROM tasks WHERE user_id = $1") .bind(user_id) @@ -445,7 +444,7 @@ async fn list_history( expires_at: DateTime, } - let tasks: Vec = if let Some(status) = &status { + let tasks: Vec = if let Some(status) = status { sqlx::query_as::<_, TaskRow>( r#" SELECT @@ -459,7 +458,7 @@ async fn list_history( completed_at, expires_at FROM tasks - WHERE user_id = $1 AND status::text = $2 + WHERE user_id = $1 AND status = $2::task_status ORDER BY created_at DESC LIMIT $3 OFFSET $4 "#, diff --git a/src/services/compress.rs b/src/services/compress.rs index 03a8975..630c982 100644 --- a/src/services/compress.rs +++ b/src/services/compress.rs @@ -874,11 +874,16 @@ fn encode_jpeg_target( JPEG_TARGET_MIN_QUALITY, 95, deadline, - |img, q| { + |img| { let rgb = jpeg_rgb(img); let (w, h) = rgb.dimensions(); - encode_jpeg_raw(rgb.as_raw(), w, h, q) + TargetPixels { + bytes: rgb.into_raw(), + width: w, + height: h, + } }, + encode_jpeg_raw, ) } @@ -893,11 +898,16 @@ fn encode_webp_target( WEBP_TARGET_MIN_QUALITY, 95, deadline, - |img, q| { + |img| { let rgba = img.to_rgba8(); let (w, h) = rgba.dimensions(); - encode_webp_raw(rgba.as_raw(), w, h, q) + TargetPixels { + bytes: rgba.into_raw(), + width: w, + height: h, + } }, + encode_webp_raw, ) } @@ -912,14 +922,25 @@ fn encode_avif_target( AVIF_TARGET_MIN_QUALITY, 95, deadline, - |img, q| { + |img| { let rgba = img.to_rgba8(); let (w, h) = rgba.dimensions(); - encode_avif_raw(rgba.as_raw(), w, h, q) + TargetPixels { + bytes: rgba.into_raw(), + width: w, + height: h, + } }, + encode_avif_raw, ) } +struct TargetPixels { + bytes: Vec, + width: u32, + height: u32, +} + /// 目标体积压缩（质量优先 + 有边界的降尺寸） /// /// 策略： @@ -927,16 +948,18 @@ fn encode_avif_target( /// 2) 原尺寸失败时先验证最低允许尺寸，快速识别不可达目标； /// 3) 最低尺寸可达时，二分查找满足目标的最高分辨率； /// 4) 严格限制最小缩放比例，避免“过度糊图”。 -fn encode_with_auto_resize( +fn encode_with_auto_resize( image: DynamicImage, target_size: u64, min_q: u8, max_q: u8, deadline: &CompressionDeadline, - mut encode_fn: F, + mut prepare_fn: P, + mut encode_fn: E, ) -> Result, AppError> where - F: FnMut(&DynamicImage, u8) -> Result, AppError>, + P: FnMut(&DynamicImage) -> TargetPixels, + E: FnMut(&[u8], u32, u32, u8) -> Result, AppError>, { deadline.check()?; let (orig_w, orig_h) = image.dimensions(); @@ -947,10 +970,10 @@ where let (full_result, full_dimensions) = encode_target_at_scale( &image, 1.0, - min_q, - max_q, + (min_q, max_q), target_size, deadline, + &mut prepare_fn, &mut encode_fn, )?; if full_result.len() as u64 <= target_size { @@ -968,10 +991,10 @@ where let (min_result, min_dimensions) = encode_target_at_scale( &image, min_scale, - min_q, - max_q, + (min_q, max_q), target_size, deadline, + &mut prepare_fn, &mut encode_fn, )?; if min_result.len() as u64 > target_size { @@ -1006,10 +1029,10 @@ where let (candidate_result, _) = encode_target_at_scale( &image, candidate_scale, - min_q, - max_q, + (min_q, max_q), target_size, deadline, + &mut prepare_fn, &mut encode_fn, )?; if candidate_result.len() as u64 <= target_size { @@ -1040,17 +1063,18 @@ fn dimensions_at_scale(orig_w: u32, orig_h: u32, scale: f64) -> (u32, u32) { ) } -fn encode_target_at_scale( +fn encode_target_at_scale( image: &DynamicImage, scale: f64, - min_q: u8, - max_q: u8, + quality_range: (u8, u8), target_size: u64, deadline: &CompressionDeadline, - encode_fn: &mut F, + prepare_fn: &mut P, + encode_fn: &mut E, ) -> Result<(Vec, (u32, u32)), AppError> where - F: FnMut(&DynamicImage, u8) -> Result, AppError>, + P: FnMut(&DynamicImage) -> TargetPixels, + E: FnMut(&[u8], u32, u32, u8) -> Result, AppError>, { deadline.check()?; let (orig_w, orig_h) = image.dimensions(); @@ -1064,40 +1088,36 @@ where }); deadline.check()?; let candidate = resized.as_ref().unwrap_or(image); - let result = encode_target_quality_with_image( - candidate, - min_q, - max_q, - target_size, - deadline, - encode_fn, - )?; + let pixels = prepare_fn(candidate); + deadline.check()?; + let (min_q, max_q) = quality_range; + let result = encode_target_quality(&pixels, min_q, max_q, target_size, deadline, encode_fn)?; Ok((result, dimensions)) } /// 对给定图片进行二分质量搜索 -fn encode_target_quality_with_image( - image: &DynamicImage, +fn encode_target_quality( + pixels: &TargetPixels, min_q: u8, max_q: u8, target_size: u64, deadline: &CompressionDeadline, - encode_fn: &mut F, + encode_fn: &mut E, ) -> Result, AppError> where - F: FnMut(&DynamicImage, u8) -> Result, AppError>, + E: FnMut(&[u8], u32, u32, u8) -> Result, AppError>, { deadline.check()?; // Start with the highest quality. If it already fits, no lower-quality // encodes can improve the result. - let max_quality = encode_fn(image, max_q)?; + let max_quality = encode_fn(&pixels.bytes, pixels.width, pixels.height, max_q)?; let max_size = max_quality.len() as u64; if max_size <= target_size || min_q == max_q { return Ok(max_quality); } deadline.check()?; - let min_quality = encode_fn(image, min_q)?; + let min_quality = encode_fn(&pixels.bytes, pixels.width, pixels.height, min_q)?; let min_size = min_quality.len() as u64; if min_size > target_size { return Ok(min_quality); @@ -1132,7 +1152,7 @@ where }; let candidate_q = under_q.saturating_add(estimated_offset.clamp(1, quality_span.saturating_sub(1)) as u8); - let bytes = encode_fn(image, candidate_q)?; + let bytes = encode_fn(&pixels.bytes, pixels.width, pixels.height, candidate_q)?; let size = bytes.len() as u64; if size > target_size { over_q = candidate_q; @@ -1149,7 +1169,7 @@ where return Ok(best_under); } let mid = under_q + (over_q - under_q) / 2; - let bytes = encode_fn(image, mid)?; + let bytes = encode_fn(&pixels.bytes, pixels.width, pixels.height, mid)?; let size = bytes.len() as u64; if size > target_size { over_q = mid; @@ -1371,6 +1391,15 @@ fn is_animated_image(input: &[u8], format: ImageFmt) -> Result { mod tests { use super::*; + fn empty_target_pixels(image: &DynamicImage) -> TargetPixels { + let (width, height) = image.dimensions(); + TargetPixels { + bytes: Vec::new(), + width, + height, + } + } + fn crc32(bytes: &[u8]) -> u32 { let mut crc = u32::MAX; for byte in bytes { @@ -1582,7 +1611,8 @@ mod tests { 40, 95, &CompressionDeadline::unlimited(), - |_image, quality| { + empty_target_pixels, + |_raw, _width, _height, quality| { calls.set(calls.get() + 1); Ok(vec![0; quality as usize]) }, @@ -1598,14 +1628,18 @@ mod tests { use std::cell::RefCell; let calls = RefCell::new(Vec::new()); - let image = DynamicImage::new_rgb8(32, 32); - let result = encode_target_quality_with_image( - &image, + let pixels = TargetPixels { + bytes: Vec::new(), + width: 32, + height: 32, + }; + let result = encode_target_quality( + &pixels, 25, 95, 50_000, &CompressionDeadline::unlimited(), - &mut |_image, quality| { + &mut |_raw, _width, _height, quality| { calls.borrow_mut().push(quality); Ok(vec![0; usize::from(quality) * 1_000]) }, @@ -1625,10 +1659,8 @@ mod tests { 40, 40, &CompressionDeadline::unlimited(), - |image, _quality| { - let (width, height) = image.dimensions(); - Ok(vec![0; (width as usize * height as usize) / 10]) - }, + empty_target_pixels, + |_raw, width, height, _quality| Ok(vec![0; (width as usize * height as usize) / 10]), ) .unwrap(); @@ -1646,7 +1678,8 @@ mod tests { 40, 40, &CompressionDeadline::unlimited(), - |_image, _quality| { + empty_target_pixels, + |_raw, _width, _height, _quality| { calls.set(calls.get() + 1); Ok(vec![0; 50_000]) }, @@ -1671,7 +1704,8 @@ mod tests { 40, 40, &deadline, - |_image, _quality| { + empty_target_pixels, + |_raw, _width, _height, _quality| { calls.set(calls.get() + 1); Ok(vec![0; 1]) }, @@ -1704,12 +1738,12 @@ mod tests { 40, 40, &deadline, - |image, _quality| { + empty_target_pixels, + |_raw, width, height, _quality| { calls.set(calls.get() + 1); if calls.get() == 2 { controller.cancel(); } - let (width, height) = image.dimensions(); Ok(vec![0; (width as usize * height as usize) / 10]) }, ) @@ -1719,6 +1753,34 @@ mod tests { assert_eq!(calls.get(), 2); } + #[test] + fn target_encoder_prepares_pixels_once_for_each_scale() { + use std::cell::Cell; + + let preparations = Cell::new(0); + let encodes = Cell::new(0); + let result = encode_with_auto_resize( + DynamicImage::new_rgb8(800, 600), + 50_000, + 25, + 95, + &CompressionDeadline::unlimited(), + |image| { + preparations.set(preparations.get() + 1); + empty_target_pixels(image) + }, + |_raw, _width, _height, quality| { + encodes.set(encodes.get() + 1); + Ok(vec![0; usize::from(quality) * 1_000]) + }, + ) + .unwrap(); + + assert_eq!(result.len(), 50_000); + assert_eq!(preparations.get(), 1); + assert!(encodes.get() > 1); + } + #[test] fn large_images_use_a_smaller_scale_search_budget() { assert_eq!(scale_search_attempts(4_000, 3_000), 3); diff --git a/src/services/credentials.rs b/src/services/credentials.rs index 5344bcd..864cb7b 100644 --- a/src/services/credentials.rs +++ b/src/services/credentials.rs @@ -65,6 +65,21 @@ pub async fn verify_password(password: &str, password_hash: &str) -> Result Result<(), AppError> { + let password = password.to_owned(); + tokio::task::spawn_blocking(move || { + let salt = argon2::password_hash::SaltString::encode_b64(b"imageforge-login-dummy") + .map_err(|err| err.to_string())?; + Argon2::default() + .hash_password(password.as_bytes(), &salt) + .map(|_| ()) + .map_err(|err| err.to_string()) + }) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "密码校验任务失败").with_source(err))? + .map_err(|err| AppError::new(ErrorCode::Internal, "密码校验失败").with_source(err)) +} + pub fn generate_token() -> String { let mut bytes = [0u8; 32]; rand::rngs::OsRng.fill_bytes(&mut bytes); @@ -88,6 +103,13 @@ mod tests { assert!(!verify_password("wrong-password", &hash).await.unwrap()); } + #[tokio::test] + async fn dummy_password_work_completes() { + consume_dummy_password_work("unknown-user-password") + .await + .unwrap(); + } + #[test] fn generated_tokens_are_url_safe_and_unique() { let first = generate_token(); diff --git a/src/services/idempotency.rs b/src/services/idempotency.rs index 6c6babb..80b01ce 100644 --- a/src/services/idempotency.rs +++ b/src/services/idempotency.rs @@ -148,10 +148,9 @@ pub async fn wait_for_replay( max_wait_ms: u64, ) -> Result, AppError> { let started = tokio::time::Instant::now(); - let now = Utc::now(); loop { - let row = get_row(state, scope, idempotency_key, now).await?; + let row = get_row(state, scope, idempotency_key, Utc::now()).await?; let Some(row) = row else { return Ok(None) }; if row.request_hash != request_hash { diff --git a/src/services/quota.rs b/src/services/quota.rs index f1a708d..cd39879 100644 --- a/src/services/quota.rs +++ b/src/services/quota.rs @@ -527,8 +527,25 @@ fn anonymous_session_key(session_id: &str, date: NaiveDate) -> String { format!("anon_quota:{session_id}:{}", date.format("%Y-%m-%d")) } +pub(crate) fn anonymous_ip_scope(ip: IpAddr) -> String { + match ip { + IpAddr::V4(ip) => ip.to_string(), + IpAddr::V6(ip) => { + if let Some(mapped) = ip.to_ipv4_mapped() { + return mapped.to_string(); + } + let network = std::net::Ipv6Addr::from(u128::from(ip) & (u128::MAX << 64)); + format!("{network}/64") + } + } +} + fn anonymous_ip_key(ip: IpAddr, date: NaiveDate) -> String { - format!("anon_quota_ip:{ip}:{}", date.format("%Y-%m-%d")) + format!( + "anon_quota_ip:{}:{}", + anonymous_ip_scope(ip), + date.format("%Y-%m-%d") + ) } fn utc8_date() -> NaiveDate { @@ -601,4 +618,19 @@ mod tests { "anon_quota_ip:127.0.0.1:2026-07-25" ); } + + #[test] + fn anonymous_ipv6_addresses_share_a_slash_64_scope() { + let first: IpAddr = "2001:db8:1234:5678::1".parse().unwrap(); + let second: IpAddr = "2001:db8:1234:5678:ffff::99".parse().unwrap(); + let other: IpAddr = "2001:db8:1234:5679::1".parse().unwrap(); + + assert_eq!(anonymous_ip_scope(first), "2001:db8:1234:5678::/64"); + assert_eq!(anonymous_ip_scope(first), anonymous_ip_scope(second)); + assert_ne!(anonymous_ip_scope(first), anonymous_ip_scope(other)); + assert_eq!( + anonymous_ip_scope("::ffff:203.0.113.7".parse().unwrap()), + "203.0.113.7" + ); + } } diff --git a/src/worker/mod.rs b/src/worker/mod.rs index 0fc8ab9..318b4e1 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -569,14 +569,20 @@ async fn process_task_file( ctx: TaskContext, billing_ctx: Option, ) -> Result<(), AppError> { - if is_task_cancelled(&state, task_id).await? { - return Ok(()); - } - let updated = sqlx::query( - "UPDATE task_files SET status = 'processing' WHERE id = $1 AND status = 'pending'", + r#" + UPDATE task_files AS f + SET status = 'processing' + FROM tasks AS t + WHERE f.id = $1 + AND f.task_id = t.id + AND t.id = $2 + AND f.status = 'pending' + AND t.status IN ('pending', 'processing') + "#, ) .bind(file.id) + .bind(task_id) .execute(&state.db) .await .map_err(|err| AppError::new(ErrorCode::Internal, "更新文件处理状态失败").with_source(err))?; @@ -584,11 +590,6 @@ async fn process_task_file( return Ok(()); } - if is_task_cancelled(&state, task_id).await? { - mark_file_failed(&state, task_id, file.id, "已取消").await?; - return Ok(()); - } - let Some(input_path) = file.input_path.clone() else { mark_file_failed(&state, task_id, file.id, "原文件不存在").await?; return Ok(()); @@ -702,13 +703,6 @@ async fn process_task_file( } } - if is_task_cancelled(&state, task_id).await? { - let _ = storage::delete_object(&state, &stored_locator(&stored)).await; - mark_file_failed(&state, task_id, file.id, "已取消").await?; - let _ = tokio::fs::remove_file(&input_path).await; - return Ok(()); - } - if let Err(err) = finalize_file( &state, &billing_ctx, @@ -766,6 +760,9 @@ async fn finalize_file( .map_err(|err| { AppError::new(ErrorCode::Internal, "锁定任务状态失败").with_source(err) })?; + if matches!(task_status.as_deref(), Some("cancelled")) { + return Err(AppError::new(ErrorCode::InvalidRequest, "已取消")); + } if !matches!(task_status.as_deref(), Some("pending" | "processing")) { return Err(AppError::new(ErrorCode::InvalidRequest, "任务已结束")); }