From 8a379e89e6e6e9f798038283b8a7c5edc7e59350 Mon Sep 17 00:00:00 2001 From: turtton Date: Sat, 12 Sep 2026 23:35:10 +0900 Subject: [PATCH 01/13] =?UTF-8?q?fix(drive):=20=E3=82=A2=E3=83=83=E3=83=97?= =?UTF-8?q?=E3=83=AD=E3=83=BC=E3=83=89=E3=81=AE=E3=83=95=E3=82=A1=E3=82=A4?= =?UTF-8?q?=E3=83=AB=E5=90=8D=E6=A4=9C=E8=A8=BC=E3=82=92=E3=83=95=E3=82=A9?= =?UTF-8?q?=E3=83=AB=E3=83=80=E5=90=8D=E8=A6=8F=E7=B4=84=E3=81=AB=E6=8F=83?= =?UTF-8?q?=E3=81=88=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/drive/files.rs | 2 +- core/src/drive/folders.rs | 8 +------ core/src/drive/mod.rs | 45 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 8 deletions(-) diff --git a/core/src/drive/files.rs b/core/src/drive/files.rs index 6faa215..1a935e3 100644 --- a/core/src/drive/files.rs +++ b/core/src/drive/files.rs @@ -140,8 +140,8 @@ async fn upload_file( ) -> Result<(StatusCode, Json), AppError> { let name = query .name - .filter(|name| !name.trim().is_empty()) .ok_or_else(|| AppError::Validation("file name is required".into()))?; + super::validate_name(&name, "file")?; let mime = query .mime .filter(|mime| !mime.trim().is_empty()) diff --git a/core/src/drive/folders.rs b/core/src/drive/folders.rs index cfce009..6f492fc 100644 --- a/core/src/drive/folders.rs +++ b/core/src/drive/folders.rs @@ -174,13 +174,7 @@ fn folder_response(row: FolderRow) -> Result { } fn validate_folder_name(name: &str) -> Result<(), AppError> { - if name.trim().is_empty() { - return Err(AppError::Validation("folder name must not be blank".into())); - } - if name.chars().count() > 255 { - return Err(AppError::Validation("folder name is too long".into())); - } - Ok(()) + super::validate_name(name, "folder") } fn map_database_error(error: sqlx::Error) -> AppError { diff --git a/core/src/drive/mod.rs b/core/src/drive/mod.rs index 2b816a6..e3eeacf 100644 --- a/core/src/drive/mod.rs +++ b/core/src/drive/mod.rs @@ -3,3 +3,48 @@ pub mod files; pub mod folders; pub mod upload_body; + +pub(crate) fn validate_name(name: &str, kind: &'static str) -> Result<(), crate::error::AppError> { + if name.trim().is_empty() { + return Err(crate::error::AppError::Validation(format!( + "{kind} name must not be blank" + ))); + } + if name.chars().count() > 255 { + return Err(crate::error::AppError::Validation(format!( + "{kind} name is too long" + ))); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use rstest::rstest; + + #[rstest] + #[case("", Some("must not be blank"))] + #[case(" \n", Some("must not be blank"))] + #[case(&"a".repeat(255), None)] + #[case(&"a".repeat(256), Some("is too long"))] + #[case(&"あ".repeat(255), None)] + #[case(&"あ".repeat(256), Some("is too long"))] + fn validate_name_enforces_character_limits( + #[case] name: &str, + #[case] expected: Option<&str>, + #[values("file", "folder")] kind: &'static str, + ) { + // Given: a file or folder name at a validation boundary. + // When + let result = super::validate_name(name, kind); + // Then + match expected { + Some(message) => assert!(matches!( + result, + Err(crate::error::AppError::Validation(actual)) + if actual == format!("{kind} name {message}") + )), + None => assert!(result.is_ok()), + } + } +} From 9a3a5b82cce6777dfb79384015772fd36ee01d93 Mon Sep 17 00:00:00 2001 From: turtton Date: Sat, 12 Sep 2026 23:35:44 +0900 Subject: [PATCH 02/13] =?UTF-8?q?fix(billing):=20upsert=5Frule=20=E3=82=92?= =?UTF-8?q?=E5=8D=98=E4=B8=80=20INSERT=20...=20ON=20CONFLICT=20=E3=81=AB?= =?UTF-8?q?=E5=8E=9F=E5=AD=90=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ultraworked with [Sisyphus](https://github.com/code-yeongyu/ohmy-openagent) Co-authored-by: Sisyphus --- core/src/billing/rules.rs | 89 +++++++++++++++++++++++++-------------- 1 file changed, 58 insertions(+), 31 deletions(-) diff --git a/core/src/billing/rules.rs b/core/src/billing/rules.rs index 4af4f4e..c444aba 100644 --- a/core/src/billing/rules.rs +++ b/core/src/billing/rules.rs @@ -48,9 +48,8 @@ pub async fn list_rules(pool: &sqlx::PgPool) -> Result, AppErro /// Insert or update the rule for (`scope`, `key`). `scope = None` targets /// the global layer. /// -/// Manual two-step upsert: the unique constraint is an expression index -/// over `COALESCE(owner_type,''), COALESCE(owner_id,''), key`, which plain -/// `ON CONFLICT` cannot target in text queries. +/// Upserts a rule using `billing_rules_scope_key_uq` as an expression-index +/// arbiter; the double parentheses enable PostgreSQL expression inference. pub async fn upsert_rule( pool: &sqlx::PgPool, scope: Option<&Owner>, @@ -73,37 +72,11 @@ pub async fn upsert_rule( None => (None, None), }; - let mut update = sqlx::query_as::<_, BillingRule>( - "UPDATE billing_rules SET value = $3, enabled = $4 \ - WHERE key = $5 \ - AND ((owner_type IS NULL AND $1::text IS NULL) OR owner_type = $1) \ - AND ((owner_id IS NULL AND $2::text IS NULL) OR owner_id = $2) \ - RETURNING id, owner_type, owner_id, key, value, enabled, created_at", - ); - match scope { - Some(owner) => { - update = update.bind(&owner.owner_type).bind(&owner.owner_id); - } - None => { - update = update - .bind(Option::::None) - .bind(Option::::None); - } - } - let updated = update - .bind(&value) - .bind(enabled) - .bind(key) - .fetch_optional(pool) - .await - .map_err(|err| AppError::Internal(format!("update billing rule failed: {err}")))?; - if let Some(rule) = updated { - return Ok(rule); - } - sqlx::query_as::<_, BillingRule>( "INSERT INTO billing_rules (owner_type, owner_id, key, value, enabled) \ VALUES ($1, $2, $3, $4, $5) \ + ON CONFLICT ((COALESCE(owner_type, '')), (COALESCE(owner_id, '')), key) \ + DO UPDATE SET value = EXCLUDED.value, enabled = EXCLUDED.enabled \ RETURNING id, owner_type, owner_id, key, value, enabled, created_at", ) .bind(owner_type) @@ -256,6 +229,60 @@ mod tests { cleanup(&pool, &owner).await; } + #[tokio::test] + #[ignore = "requires the compose postgres stack"] + async fn pg_concurrent_upserts_do_not_conflict() { + let pool = test_pool().await; + let owner = Owner::new("test-billing-rules", Uuid::now_v7().to_string()); + cleanup(&pool, &owner).await; + + let mut tasks = tokio::task::JoinSet::new(); + for value in 0..8 { + let pool = pool.clone(); + let owner = owner.clone(); + tasks.spawn(async move { + upsert_rule( + &pool, + Some(&owner), + "max_file_bytes", + serde_json::json!(value), + true, + ) + .await + }); + } + + let mut results = Vec::new(); + while let Some(result) = tasks.join_next().await { + results.push(result.expect("upsert task panicked")); + } + + assert!(results.iter().all(Result::is_ok)); + let rules: Vec = results + .into_iter() + .map(|result| result.expect("concurrent upsert failed")) + .collect(); + let first_id = rules[0].id; + assert!(rules.iter().all(|rule| rule.id == first_id)); + assert!( + rules + .iter() + .all(|rule| (0..8).any(|value| rule.value == serde_json::json!(value))) + ); + let stored_value: serde_json::Value = sqlx::query_scalar( + "SELECT value FROM billing_rules WHERE owner_type = $1 AND owner_id = $2 AND key = $3", + ) + .bind(&owner.owner_type) + .bind(&owner.owner_id) + .bind("max_file_bytes") + .fetch_one(&pool) + .await + .expect("stored concurrent upsert value"); + assert!((0..8).any(|value| stored_value == serde_json::json!(value))); + + cleanup(&pool, &owner).await; + } + async fn test_pool() -> sqlx::PgPool { let url = std::env::var("BOOSKIFF_DATABASE_URL") .unwrap_or_else(|_| "postgres://booskiff:booskiff@127.0.0.1:5432/booskiff".to_owned()); From 97f119d9ed66e7ac32a0d0cc79d3ded31b9bf202 Mon Sep 17 00:00:00 2001 From: turtton Date: Sat, 12 Sep 2026 23:39:01 +0900 Subject: [PATCH 03/13] =?UTF-8?q?fix(drive):=20=E3=83=9C=E3=83=87=E3=82=A3?= =?UTF-8?q?=E3=82=B5=E3=82=A4=E3=82=BA=E5=BC=B7=E5=88=B6=E3=82=92=20Counti?= =?UTF-8?q?ngBody=20=E3=81=AB=E4=B8=80=E6=9C=AC=E5=8C=96=E3=81=97=E8=B6=85?= =?UTF-8?q?=E9=81=8E=E5=BF=9C=E7=AD=94=E3=82=92=E5=9B=BA=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/drive/files.rs | 106 +++++++++++++++++++++++++++++++++++----- core/src/main.rs | 2 +- 2 files changed, 95 insertions(+), 13 deletions(-) diff --git a/core/src/drive/files.rs b/core/src/drive/files.rs index 1a935e3..46b130b 100644 --- a/core/src/drive/files.rs +++ b/core/src/drive/files.rs @@ -7,7 +7,7 @@ use std::time::Duration; use aws_smithy_types::body::SdkBody; use axum::body::Body; -use axum::extract::{DefaultBodyLimit, Path, Query, State}; +use axum::extract::{Path, Query, State}; use axum::http::{HeaderMap, StatusCode, header}; use axum::routing::{get, post}; use axum::{Json, Router}; @@ -19,7 +19,6 @@ use uuid::Uuid; use crate::auth::extractor::AccountContext; use crate::billing::usage::{add_used_bytes, subtract_used_bytes, used_bytes}; -use crate::config::Config; use crate::error::AppError; use crate::model::{OBJECT_KIND_ORIGINAL, Owner}; use crate::state::AppState; @@ -29,7 +28,6 @@ use super::upload_body::CountingBody; const DEFAULT_LIST_LIMIT: i64 = 50; const MAX_LIST_LIMIT: i64 = 200; -const BODY_LIMIT_SLACK: i64 = 1024 * 1024; #[derive(Debug, Serialize, utoipa::ToSchema)] pub struct FileResponse { @@ -108,15 +106,10 @@ impl FileRow { } } -pub fn upload_router(config: &Config) -> Router { - let body_limit_i64 = config - .plan_premium_max_file_bytes - .saturating_add(BODY_LIMIT_SLACK); - let body_limit = usize::try_from(body_limit_i64).unwrap_or(usize::MAX); - Router::new().route( - "/v1/files", - post(upload_file).route_layer(DefaultBodyLimit::max(body_limit)), - ) +/// CountingBody in upload_file is the sole body-size enforcement path. +/// DefaultBodyLimit would be a no-op because upload_file extracts Body directly. +pub fn upload_router() -> Router { + Router::new().route("/v1/files", post(upload_file)) } pub fn files_router() -> Router { @@ -489,6 +482,95 @@ async fn persist_uploaded_file( #[cfg(test)] mod tests { use super::*; + use crate::config::Config; + use axum::response::IntoResponse; + + #[tokio::test] + async fn upload_returns_413_when_declared_length_exceeds_limit() { + // Given: a closed pool makes any database access fail. + let config = Config::default(); + let pool = sqlx::postgres::PgPoolOptions::new() + .connect_lazy("postgres://unused:unused@127.0.0.1:1/unused") + .unwrap(); + pool.close().await; + let storage = Storage::build(&config).await.unwrap(); + let state = test_state(config, pool, storage).await; + let ctx = AccountContext { + owner: Owner::new("test-files", Uuid::now_v7().to_string()), + limits: crate::model::Limits { + storage_quota_bytes: 1024, + max_file_bytes: 4, + rate_limit_rpm: 100, + }, + }; + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_LENGTH, "5".parse().unwrap()); + + // When + let error = upload_file( + ctx, + State(state), + Query(UploadQuery { + name: Some("large.txt".into()), + mime: None, + folder_id: None, + }), + headers, + Body::empty(), + ) + .await + .unwrap_err(); + + // Then + assert!(matches!(error, AppError::PayloadTooLarge(_))); + assert_eq!( + error.into_response().status(), + StatusCode::PAYLOAD_TOO_LARGE + ); + } + + #[tokio::test] + #[ignore = "requires compose postgres and MinIO"] + async fn upload_returns_413_when_actual_body_exceeds_limit() { + // Given: the declared length fits exactly, but the body exceeds the cap. + let config = Config::default(); + let pool = test_pool().await; + let storage = Storage::build(&config).await.unwrap(); + storage.ensure_bucket().await.unwrap(); + let state = test_state(config, pool, storage).await; + let ctx = AccountContext { + owner: Owner::new("test-files", Uuid::now_v7().to_string()), + limits: crate::model::Limits { + storage_quota_bytes: 1024, + max_file_bytes: 4, + rate_limit_rpm: 100, + }, + }; + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_LENGTH, "4".parse().unwrap()); + + // When + let error = upload_file( + ctx, + State(state), + Query(UploadQuery { + name: Some("large.txt".into()), + mime: None, + folder_id: None, + }), + headers, + Body::from("12345"), + ) + .await + .unwrap_err(); + + // Then + assert!(matches!(error, AppError::PayloadTooLarge(_))); + assert_eq!( + error.into_response().status(), + StatusCode::PAYLOAD_TOO_LARGE + ); + } #[test] fn foreign_file_row_maps_to_not_found() { diff --git a/core/src/main.rs b/core/src/main.rs index ae245bf..c900b99 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -93,7 +93,7 @@ async fn run() { } let jwks_cache = JwksCache::new(config.jwt_trusted_issuers.clone()); - let upload_routes = drive::files::upload_router(&config); + let upload_routes = drive::files::upload_router(); let files_routes = drive::files::files_router(); let state = AppState { pool, From 837c6d57df9fbdbc81b0e51c3b0c6da21ae0d9f3 Mon Sep 17 00:00:00 2001 From: turtton Date: Sat, 12 Sep 2026 23:44:48 +0900 Subject: [PATCH 04/13] =?UTF-8?q?fix(drive):=20=E3=83=95=E3=82=A9=E3=83=AB?= =?UTF-8?q?=E3=83=80=E5=89=8A=E9=99=A4=E7=AB=B6=E5=90=88=E6=99=82=E3=81=AE?= =?UTF-8?q?=20FK=20=E9=81=95=E5=8F=8D=E3=82=92=20404=20=E3=81=AB=E3=83=9E?= =?UTF-8?q?=E3=83=83=E3=83=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- core/src/drive/files.rs | 51 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) diff --git a/core/src/drive/files.rs b/core/src/drive/files.rs index 46b130b..68e5766 100644 --- a/core/src/drive/files.rs +++ b/core/src/drive/files.rs @@ -450,7 +450,7 @@ async fn persist_uploaded_file( .bind(upload.size_bytes) .fetch_one(&mut *tx) .await - .map_err(|err| AppError::Internal(format!("insert file: {err}")))?; + .map_err(|err| map_insert_file_error(err, upload.folder_id))?; sqlx::query( "INSERT INTO file_objects \ (file_id, object_kind, storage_key, size_bytes, mime_type) VALUES ($1, $2, $3, $4, $5)", @@ -479,6 +479,17 @@ async fn persist_uploaded_file( Ok(response) } +fn map_insert_file_error(error: sqlx::Error, folder_id: Option) -> AppError { + match (&error, folder_id) { + (sqlx::Error::Database(database_error), Some(folder_id)) + if database_error.code().as_deref() == Some("23503") => + { + AppError::NotFound(format!("folder {folder_id}")) + } + _ => AppError::Internal(format!("insert file: {error}")), + } +} + #[cfg(test)] mod tests { use super::*; @@ -572,6 +583,44 @@ mod tests { ); } + #[tokio::test] + #[ignore = "requires the compose postgres stack"] + async fn pg_upload_into_vanished_folder_is_not_found() { + // Given: the folder ID does not exist in PostgreSQL. + let config = Config::default(); + let pool = test_pool().await; + let storage = Storage::build(&config).await.unwrap(); + let state = test_state(config, pool, storage).await; + let folder_id = Uuid::now_v7(); + let ctx = AccountContext { + owner: Owner::new("test-drive-files", Uuid::now_v7().to_string()), + limits: crate::model::Limits { + storage_quota_bytes: 1024, + max_file_bytes: 1024, + rate_limit_rpm: 100, + }, + }; + + // When + let error = persist_uploaded_file( + &state, + &ctx, + UploadedFile { + id: Uuid::now_v7(), + folder_id: Some(folder_id), + name: "vanished-folder.txt".into(), + mime: "text/plain".into(), + key: "test-drive-files/vanished-folder.txt".into(), + size_bytes: 1, + }, + ) + .await + .unwrap_err(); + + // Then + assert!(matches!(error, AppError::NotFound(_))); + } + #[test] fn foreign_file_row_maps_to_not_found() { // Given From 50caa052e60d32fa13b61a6dd46abc554ee12665 Mon Sep 17 00:00:00 2001 From: turtton Date: Sat, 12 Sep 2026 23:53:45 +0900 Subject: [PATCH 05/13] =?UTF-8?q?feat(billing):=20=E8=AA=B2=E9=87=91?= =?UTF-8?q?=E8=A7=A3=E6=B1=BA=E3=81=AB=20TTL=20=E3=82=AD=E3=83=A3=E3=83=83?= =?UTF-8?q?=E3=82=B7=E3=83=A5=E3=81=A8=E7=AE=A1=E7=90=86=E8=80=85=E6=9B=B8?= =?UTF-8?q?=E3=81=8D=E8=BE=BC=E3=81=BF=E5=8D=B3=E6=99=82=E7=84=A1=E5=8A=B9?= =?UTF-8?q?=E5=8C=96=E3=82=92=E5=B0=8E=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/admin/billing_cache_tests.rs | 106 ++++++++++++ core/src/admin/handlers.rs | 18 +- core/src/auth/admin_auth.rs | 6 + core/src/auth/extractor.rs | 19 +- core/src/billing/cache.rs | 239 ++++++++++++++++++++++++++ core/src/billing/mod.rs | 1 + core/src/billing/resolve.rs | 122 +++++++++++++ core/src/config.rs | 7 + core/src/drive/files.rs | 3 + core/src/main.rs | 3 + core/src/state.rs | 1 + 11 files changed, 518 insertions(+), 7 deletions(-) create mode 100644 core/src/admin/billing_cache_tests.rs create mode 100644 core/src/billing/cache.rs diff --git a/core/src/admin/billing_cache_tests.rs b/core/src/admin/billing_cache_tests.rs new file mode 100644 index 0000000..465823c --- /dev/null +++ b/core/src/admin/billing_cache_tests.rs @@ -0,0 +1,106 @@ +use super::*; +use crate::billing::cache::{BillingCache, OwnerEntry}; +use crate::config::Config; +use axum::body::Body; +use axum::http::Request; +use http_body_util::BodyExt; +use std::sync::Arc; +use tower::ServiceExt; + +#[tokio::test] +#[ignore = "requires compose postgres"] +async fn pg_admin_billing_writes_invalidate_cache() { + let config = Config::default(); + let pool = sqlx::PgPool::connect(&config.database_url).await.unwrap(); + let mut guard = pool.begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(0x0062_6969_i64) + .execute(&mut *guard) + .await + .unwrap(); + let token = uuid::Uuid::now_v7().to_string(); + sqlx::query("INSERT INTO admin_tokens (name, token_hash) VALUES ($1, $2)") + .bind(&token) + .bind(hash_admin_token(&token)) + .execute(&pool) + .await + .unwrap(); + let owner = Owner::new("test-admin-cache", uuid::Uuid::now_v7().to_string()); + let cache = Arc::new(BillingCache::new(config.billing_cache_ttl_secs)); + let state = AppState { + s3: crate::storage::Storage::build(&config).await.unwrap(), + jwks_cache: crate::auth::jwks::JwksCache::new(Vec::new()), + pool: pool.clone(), + config, + rate_limiters: Arc::new(crate::state::RateLimiters::default()), + billing_cache: Arc::clone(&cache), + }; + let router = admin_router().with_state(state); + let plan_uri = format!("/owners/{}/{}/plan", owner.owner_type, owner.owner_id); + let owner_rule = serde_json::json!({"owner_type": owner.owner_type, "owner_id": owner.owner_id, + "key": "max_file_bytes", "value": 71, "enabled": true}); + let mut rule_id = String::new(); + for (method, uri, body, all) in [ + ( + "PUT", + plan_uri.clone(), + serde_json::json!({"plan":"premium"}), + false, + ), + ("DELETE", plan_uri, serde_json::Value::Null, false), + ("POST", "/billing/rules".into(), owner_rule, false), + ("DELETE", String::new(), serde_json::Value::Null, true), + ( + "POST", + "/billing/rules".into(), + serde_json::json!({"key":"max_file_bytes", "value":72, "enabled":true}), + true, + ), + ("DELETE", String::new(), serde_json::Value::Null, true), + ] { + cache.store_global(vec![]); + for key in [owner.key(), "other".into()] { + cache.store_owner( + &key, + OwnerEntry { + assignment: None, + rules: vec![], + }, + ); + } + let uri = if uri.is_empty() { + format!("/billing/rules/{rule_id}") + } else { + uri + }; + let request = Request::builder() + .method(method) + .uri(uri) + .header("x-admin-token", &token) + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(); + let response = router.clone().oneshot(request).await.unwrap(); + assert!( + response.status().is_success(), + "{method}: {}", + response.status() + ); + assert!( + cache.get_owner(&owner.key()).is_none(), + "{method} must invalidate owner" + ); + assert_eq!(cache.get_owner("other").is_none(), all); + assert_eq!(cache.get_global().is_none(), all); + if method == "POST" { + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + let item: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + rule_id = item["id"].as_str().unwrap().to_owned(); + } + } + sqlx::query("DELETE FROM admin_tokens WHERE name = $1") + .bind(&token) + .execute(&pool) + .await + .unwrap(); +} diff --git a/core/src/admin/handlers.rs b/core/src/admin/handlers.rs index 1055cc4..714bbb4 100644 --- a/core/src/admin/handlers.rs +++ b/core/src/admin/handlers.rs @@ -293,6 +293,10 @@ async fn create_billing_rule( request.enabled, ) .await?; + match scope { + Some(owner) => state.billing_cache.invalidate_owner(&owner.key()), + None => state.billing_cache.invalidate_all(), + } Ok(Json(billing_rule_item(rule)?)) } @@ -314,6 +318,8 @@ async fn delete_billing_rule( Path(id): Path, ) -> Result { if rules::delete_rule(&state.pool, id).await? { + // Deletion returns no scope; conservatively invalidate every layer. + state.billing_cache.invalidate_all(); Ok(StatusCode::NO_CONTENT) } else { Err(AppError::NotFound(format!("billing rule {id} not found"))) @@ -343,7 +349,9 @@ async fn set_plan( Json(request): Json, ) -> Result { let plan = parse_plan(&request.plan)?; - assignments::set_plan(&state.pool, &Owner::new(owner_type, owner_id), plan).await?; + let owner = Owner::new(owner_type, owner_id); + assignments::set_plan(&state.pool, &owner, plan).await?; + state.billing_cache.invalidate_owner(&owner.key()); Ok(StatusCode::NO_CONTENT) } @@ -395,6 +403,7 @@ async fn delete_plan( ) -> Result { let owner = Owner::new(owner_type, owner_id); if assignments::delete_plan(&state.pool, &owner).await? { + state.billing_cache.invalidate_owner(&owner.key()); Ok(StatusCode::NO_CONTENT) } else { Err(AppError::NotFound(format!( @@ -456,6 +465,10 @@ pub fn admin_router() -> Router { .route("/owners/{owner_type}/{owner_id}/usage", get(get_usage)) } +#[cfg(test)] +#[path = "billing_cache_tests.rs"] +mod billing_cache_tests; + #[cfg(test)] mod tests { use std::collections::HashSet; @@ -603,6 +616,9 @@ mod tests { pool: pool.clone(), s3: Storage::build(&config).await.unwrap(), jwks_cache: JwksCache::new(Vec::new()), + billing_cache: std::sync::Arc::new(crate::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), config, rate_limiters: std::sync::Arc::new(RateLimiters::default()), }; diff --git a/core/src/auth/admin_auth.rs b/core/src/auth/admin_auth.rs index 1835133..a28bab7 100644 --- a/core/src/auth/admin_auth.rs +++ b/core/src/auth/admin_auth.rs @@ -140,6 +140,9 @@ mod tests { .expect("lazy pool never connects"), s3: Storage::build(&config).await.unwrap(), jwks_cache: JwksCache::new(Vec::new()), + billing_cache: Arc::new(crate::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), config, rate_limiters: Arc::new(RateLimiters::default()), } @@ -204,6 +207,9 @@ mod tests { s3: Storage::build(&config).await.unwrap(), pool, jwks_cache: JwksCache::new(Vec::new()), + billing_cache: Arc::new(crate::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), config, rate_limiters: Arc::new(RateLimiters::default()), }; diff --git a/core/src/auth/extractor.rs b/core/src/auth/extractor.rs index d52762f..999aa80 100644 --- a/core/src/auth/extractor.rs +++ b/core/src/auth/extractor.rs @@ -5,7 +5,7 @@ use axum::http::request::Parts; use axum::response::{IntoResponse, Response}; use crate::auth::jwt::{owner_from_claims, verify_token}; -use crate::billing::resolve::effective_limits; +use crate::billing::resolve::effective_limits_cached; use crate::error::AppError; use crate::model::{Limits, Owner}; use crate::state::AppState; @@ -30,9 +30,10 @@ impl FromRequestParts for AccountContext { .map_err(|err| err.into_response())?; let owner = owner_from_claims(&claims, &state.config.jwt_owner_type_claim) .map_err(|err| err.into_response())?; - let limits = effective_limits(&state.pool, &state.config, &owner) - .await - .map_err(|err| err.into_response())?; + let limits = + effective_limits_cached(&state.billing_cache, &state.pool, &state.config, &owner) + .await + .map_err(|err| err.into_response())?; if !state .rate_limiters .check(&owner.key(), limits.rate_limit_rpm) @@ -122,8 +123,8 @@ mod tests { } /// Compose postgres (compose.yml), required by the DB-backed tests: - /// the billing `effective_limits` resolution queries `billing_rules` - /// on every request, so the 200/429 paths need a live database. + /// billing cache misses query `billing_rules`, so the 200/429 paths + /// need a live database. const PUBLISH_DB_URL: &str = "postgres://booskiff:booskiff@127.0.0.1:5432/booskiff"; /// Router state for the DB-free paths (header parsing and token @@ -144,6 +145,9 @@ mod tests { .expect("lazy pool never connects"), s3: Storage::build(&config).await.unwrap(), jwks_cache: JwksCache::new(config.jwt_trusted_issuers.clone()), + billing_cache: Arc::new(crate::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), config, rate_limiters: Arc::new(RateLimiters::default()), } @@ -168,6 +172,9 @@ mod tests { pool: sqlx::PgPool::connect(PUBLISH_DB_URL).await.unwrap(), s3: Storage::build(&config).await.unwrap(), jwks_cache: JwksCache::new(config.jwt_trusted_issuers.clone()), + billing_cache: Arc::new(crate::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), config, rate_limiters: Arc::new(RateLimiters::default()), } diff --git a/core/src/billing/cache.rs b/core/src/billing/cache.rs new file mode 100644 index 0000000..edd8974 --- /dev/null +++ b/core/src/billing/cache.rs @@ -0,0 +1,239 @@ +//! Process-local billing layers; zero TTL bypasses storage entirely. + +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +type Rules = Vec<(String, serde_json::Value)>; + +pub struct BillingCache { + ttl: Duration, + global: Mutex>>, + owners: Mutex>>, + generation: Mutex>, +} + +struct Cached { + value: T, + fetched_at: Instant, +} + +#[derive(Clone)] +pub struct OwnerEntry { + pub assignment: Option, + pub rules: Rules, +} + +/// At 10k owners, inserting an untracked key clears the map. Reloading +/// billing layers is cheap compared with unbounded process-local growth. +const MAX_CACHED_OWNERS: usize = 10_000; + +impl BillingCache { + pub fn new(ttl_secs: u64) -> Self { + Self { + ttl: Duration::from_secs(ttl_secs), + global: Mutex::new(None), + owners: Mutex::new(HashMap::new()), + generation: Mutex::new(std::sync::Arc::new(())), + } + } + + pub fn get_global(&self) -> Option { + let global = self.global.lock().unwrap_or_else(|err| err.into_inner()); + global + .as_ref() + .filter(|entry| entry.fetched_at.elapsed() < self.ttl) + .map(|entry| entry.value.clone()) + } + + pub fn get_owner(&self, key: &str) -> Option { + let owners = self.owners.lock().unwrap_or_else(|err| err.into_inner()); + owners + .get(key) + .filter(|entry| entry.fetched_at.elapsed() < self.ttl) + .map(|entry| entry.value.clone()) + } + + pub fn store_global(&self, value: Rules) { + self.store_global_if_current(value, &self.generation()); + } + + pub(crate) fn generation(&self) -> std::sync::Arc<()> { + self.generation + .lock() + .unwrap_or_else(|err| err.into_inner()) + .clone() + } + + pub(crate) fn store_global_if_current(&self, value: Rules, generation: &std::sync::Arc<()>) { + let current = self + .generation + .lock() + .unwrap_or_else(|err| err.into_inner()); + if !std::sync::Arc::ptr_eq(¤t, generation) { + return; + } + if self.ttl.is_zero() { + return; + } + *self.global.lock().unwrap_or_else(|err| err.into_inner()) = Some(Cached { + value, + fetched_at: Instant::now(), + }); + } + + pub fn store_owner(&self, key: &str, value: OwnerEntry) { + self.store_owner_if_current(key, value, &self.generation()); + } + + pub(crate) fn store_owner_if_current( + &self, + key: &str, + value: OwnerEntry, + generation: &std::sync::Arc<()>, + ) { + let current = self + .generation + .lock() + .unwrap_or_else(|err| err.into_inner()); + if !std::sync::Arc::ptr_eq(¤t, generation) { + return; + } + if self.ttl.is_zero() { + return; + } + let mut owners = self.owners.lock().unwrap_or_else(|err| err.into_inner()); + if owners.len() >= MAX_CACHED_OWNERS && !owners.contains_key(key) { + owners.clear(); + } + owners.insert( + key.to_owned(), + Cached { + value, + fetched_at: Instant::now(), + }, + ); + } + + pub fn invalidate_owner(&self, key: &str) { + let mut generation = self + .generation + .lock() + .unwrap_or_else(|err| err.into_inner()); + *generation = std::sync::Arc::new(()); + self.owners + .lock() + .unwrap_or_else(|err| err.into_inner()) + .remove(key); + } + + pub fn invalidate_all(&self) { + let mut generation = self + .generation + .lock() + .unwrap_or_else(|err| err.into_inner()); + *generation = std::sync::Arc::new(()); + self.global + .lock() + .unwrap_or_else(|err| err.into_inner()) + .take(); + self.owners + .lock() + .unwrap_or_else(|err| err.into_inner()) + .clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry() -> OwnerEntry { + OwnerEntry { + assignment: Some(crate::model::Plan::Premium), + rules: vec![], + } + } + + #[test] + fn fresh_entries_expire() { + let cache = BillingCache::new(60); + cache.store_global(vec![]); + cache.store_owner("a", entry()); + assert!(cache.get_global().is_some()); + assert!(cache.get_owner("a").is_some()); + cache.global.lock().unwrap().as_mut().unwrap().fetched_at -= Duration::from_secs(61); + cache + .owners + .lock() + .unwrap() + .get_mut("a") + .unwrap() + .fetched_at -= Duration::from_secs(61); + assert!(cache.get_global().is_none()); + assert!(cache.get_owner("a").is_none()); + } + + #[test] + fn invalidate_owner_preserves_other_layers() { + let cache = BillingCache::new(60); + cache.store_global(vec![]); + cache.store_owner("a", entry()); + cache.store_owner("b", entry()); + cache.invalidate_owner("a"); + assert!(cache.get_owner("a").is_none()); + assert!(cache.get_owner("b").is_some()); + assert!(cache.get_global().is_some()); + } + + #[test] + fn invalidate_all_clears_both_layers() { + let cache = BillingCache::new(60); + cache.store_global(vec![]); + cache.store_owner("a", entry()); + cache.invalidate_all(); + assert!(cache.get_global().is_none()); + assert!(cache.get_owner("a").is_none()); + } + + #[test] + fn zero_ttl_disables_caching() { + let cache = BillingCache::new(0); + cache.store_global(vec![]); + cache.store_owner("a", entry()); + assert!(cache.get_global().is_none()); + assert!(cache.get_owner("a").is_none()); + assert!(cache.owners.lock().unwrap().is_empty()); + } + + #[test] + fn invalidation_rejects_in_flight_fills() { + let cache = BillingCache::new(60); + let generation = cache.generation(); + cache.invalidate_owner("a"); + cache.store_owner_if_current("a", entry(), &generation); + assert!(cache.get_owner("a").is_none()); + let generation = cache.generation(); + cache.invalidate_all(); + cache.store_global_if_current(vec![], &generation); + cache.store_owner_if_current("a", entry(), &generation); + assert!(cache.get_global().is_none()); + assert!(cache.get_owner("a").is_none()); + } + + #[test] + fn overflow_clears_owners_but_not_global() { + let cache = BillingCache::new(60); + cache.store_global(vec![]); + for i in 0..MAX_CACHED_OWNERS { + cache.store_owner(&i.to_string(), entry()); + } + cache.store_owner("0", entry()); + assert_eq!(cache.owners.lock().unwrap().len(), MAX_CACHED_OWNERS); + cache.store_owner("new", entry()); + assert_eq!(cache.owners.lock().unwrap().len(), 1); + assert!(cache.get_owner("0").is_none()); + assert!(cache.get_owner("new").is_some()); + assert!(cache.get_global().is_some()); + } +} diff --git a/core/src/billing/mod.rs b/core/src/billing/mod.rs index 226bbe6..58e3039 100644 --- a/core/src/billing/mod.rs +++ b/core/src/billing/mod.rs @@ -2,6 +2,7 @@ //! TODO(Wave5). pub mod assignments; +pub mod cache; pub mod plans; pub mod provider; pub mod resolve; diff --git a/core/src/billing/resolve.rs b/core/src/billing/resolve.rs index ace9d26..f70d73d 100644 --- a/core/src/billing/resolve.rs +++ b/core/src/billing/resolve.rs @@ -42,6 +42,43 @@ pub async fn effective_limits( )) } +pub async fn effective_limits_cached( + cache: &crate::billing::cache::BillingCache, + pool: &sqlx::PgPool, + config: &Config, + owner: &Owner, +) -> Result { + let generation = cache.generation(); + let global = match cache.get_global() { + Some(rules) => rules, + None => { + let rules = load_rules(pool, GLOBAL_RULES_SQL, None).await?; + cache.store_global_if_current(rules.clone(), &generation); + rules + } + }; + let key = owner.key(); + let entry = match cache.get_owner(&key) { + Some(entry) => entry, + None => { + let assignment = match config.premium_mode { + PremiumMode::Everyone => None, + PremiumMode::Mirror => crate::billing::assignments::get_plan(pool, owner).await?, + }; + let rules = load_rules(pool, OWNER_RULES_SQL, Some(owner)).await?; + let entry = crate::billing::cache::OwnerEntry { assignment, rules }; + cache.store_owner_if_current(&key, entry.clone(), &generation); + entry + } + }; + Ok(resolve_from_parts( + config, + entry.assignment, + global, + entry.rules, + )) +} + /// Pure layer composition mirroring [`effective_limits`] without the DB so /// the resolution-order contract is unit-testable: plan base limits → /// global rules → owner rules (later layers win per key). @@ -360,6 +397,91 @@ mod tests { cleanup(&pool, &owner).await; } + #[tokio::test] + #[ignore] + async fn pg_cached_limits_stale_until_invalidated() { + use crate::billing::cache::BillingCache; + let pool = test_pool().await; + let _guard = global_rules_guard(&pool).await; + let owner = Owner::new("test-billing-cache", uuid::Uuid::now_v7().to_string()); + let other = Owner::new("test-billing-cache", uuid::Uuid::now_v7().to_string()); + cleanup(&pool, &owner).await; + let config = mirror_config(); + let cache = BillingCache::new(60); + for target in [&owner, &other] { + sqlx::query("INSERT INTO billing_rules (owner_type, owner_id, key, value) VALUES ($1, $2, 'max_file_bytes', '71')") + .bind(&target.owner_type).bind(&target.owner_id).execute(&pool).await.unwrap(); + assert_eq!( + effective_limits_cached(&cache, &pool, &config, target) + .await + .unwrap() + .max_file_bytes, + 71 + ); + sqlx::query( + "UPDATE billing_rules SET value = '82' WHERE owner_type = $1 AND owner_id = $2", + ) + .bind(&target.owner_type) + .bind(&target.owner_id) + .execute(&pool) + .await + .unwrap(); + crate::billing::assignments::set_plan(&pool, target, Plan::Premium) + .await + .unwrap(); + } + sqlx::query("INSERT INTO billing_rules (key, value) VALUES ('storage_quota_bytes', '93')") + .execute(&pool) + .await + .unwrap(); + let stale = effective_limits_cached(&cache, &pool, &config, &owner) + .await + .unwrap(); + assert_eq!(stale.max_file_bytes, 71); + assert_eq!(stale.rate_limit_rpm, config.plan_default_rate_limit_rpm); + assert_eq!( + stale.storage_quota_bytes, + config.plan_default_storage_quota_bytes + ); + cache.invalidate_owner(&owner.key()); + let fresh = effective_limits_cached(&cache, &pool, &config, &owner) + .await + .unwrap(); + assert_eq!(fresh.max_file_bytes, 82); + assert_eq!(fresh.rate_limit_rpm, config.plan_premium_rate_limit_rpm); + assert_eq!( + fresh.storage_quota_bytes, + config.plan_premium_storage_quota_bytes + ); + assert_eq!( + effective_limits_cached(&cache, &pool, &config, &other) + .await + .unwrap() + .max_file_bytes, + 71 + ); + cache.invalidate_all(); + let fresh = effective_limits_cached(&cache, &pool, &config, &other) + .await + .unwrap(); + assert_eq!(fresh.max_file_bytes, 82); + assert_eq!(fresh.storage_quota_bytes, 93); + // A closed pool proves a fully warm hit issues no SQL at all. + let closed = sqlx::postgres::PgPoolOptions::new() + .connect_lazy(&config.database_url) + .unwrap(); + closed.close().await; + assert_eq!( + effective_limits_cached(&cache, &closed, &config, &other) + .await + .unwrap() + .max_file_bytes, + 82 + ); + cleanup(&pool, &owner).await; + cleanup(&pool, &other).await; + } + async fn test_pool() -> sqlx::PgPool { let url = std::env::var("BOOSKIFF_DATABASE_URL") .unwrap_or_else(|_| "postgres://booskiff:booskiff@127.0.0.1:5432/booskiff".to_owned()); diff --git a/core/src/config.rs b/core/src/config.rs index aa60059..7816a11 100644 --- a/core/src/config.rs +++ b/core/src/config.rs @@ -53,6 +53,7 @@ pub struct Config { pub jwt_owner_type_claim: String, pub listen_addr: String, pub premium_mode: PremiumMode, + pub billing_cache_ttl_secs: u64, pub payment_provider: PaymentProviderCfg, pub plan_default_storage_quota_bytes: i64, pub plan_default_max_file_bytes: i64, @@ -82,6 +83,7 @@ impl Default for Config { jwt_owner_type_claim: "owner_type".into(), listen_addr: "0.0.0.0:3000".into(), premium_mode: PremiumMode::Everyone, + billing_cache_ttl_secs: 60, payment_provider: PaymentProviderCfg::Disabled, plan_default_storage_quota_bytes: 1024 * 1024 * 1024, plan_default_max_file_bytes: 100 * 1024 * 1024, @@ -127,6 +129,11 @@ impl Config { "BOOSKIFF_JWT_OWNER_TYPE_CLAIM", ), listen_addr: override_str(d.listen_addr, "BOOSKIFF_LISTEN_ADDR"), + billing_cache_ttl_secs: override_parse( + d.billing_cache_ttl_secs, + "BOOSKIFF_BILLING_CACHE_TTL_SECS", + "billing_cache_ttl_secs", + )?, premium_mode: parse_premium_mode( "BOOSKIFF_PREMIUM_MODE", std::env::var("BOOSKIFF_PREMIUM_MODE") diff --git a/core/src/drive/files.rs b/core/src/drive/files.rs index 68e5766..1b01851 100644 --- a/core/src/drive/files.rs +++ b/core/src/drive/files.rs @@ -720,6 +720,9 @@ mod tests { async fn test_state(config: Config, pool: sqlx::PgPool, s3: Storage) -> AppState { AppState { + billing_cache: std::sync::Arc::new(crate::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), jwks_cache: crate::auth::jwks::JwksCache::new(config.jwt_trusted_issuers.clone()), pool, s3, diff --git a/core/src/main.rs b/core/src/main.rs index c900b99..ba99552 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -98,6 +98,9 @@ async fn run() { let state = AppState { pool, s3: storage, + billing_cache: Arc::new(core::billing::cache::BillingCache::new( + config.billing_cache_ttl_secs, + )), config, jwks_cache, rate_limiters: Arc::new(RateLimiters::default()), diff --git a/core/src/state.rs b/core/src/state.rs index ce1e2ea..bca36ac 100644 --- a/core/src/state.rs +++ b/core/src/state.rs @@ -75,6 +75,7 @@ pub struct AppState { pub config: Config, pub jwks_cache: JwksCache, pub rate_limiters: Arc, + pub billing_cache: Arc, } #[cfg(test)] From 349ad6e24eb0e9a8968f484c0006c4c19e723517 Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:00:13 +0900 Subject: [PATCH 06/13] =?UTF-8?q?docs:=20/public=20=E7=84=A1=E8=AA=8D?= =?UTF-8?q?=E8=A8=BC=E3=83=BBimmutable=20=E3=82=AD=E3=83=A3=E3=83=83?= =?UTF-8?q?=E3=82=B7=E3=83=A5=E9=9D=9E=E5=A4=B1=E5=8A=B9=E3=83=BB=E3=83=97?= =?UTF-8?q?=E3=83=AD=E3=82=BB=E3=82=B9=E5=86=85=20state=20=E9=9D=9E?= =?UTF-8?q?=E5=85=B1=E6=9C=89=E3=82=92=E6=98=8E=E8=A8=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 ++ docs/operations.md | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+) create mode 100644 docs/operations.md diff --git a/README.md b/README.md index 516dcc2..0f34e5a 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,4 @@ # Booskiff ShuttlePub 群のファイル保管サービス (Misskey Drive 的 UX)。core は Rust (Axum) + PostgreSQL + S3 互換ストレージ。 + +運用仕様メモは [docs/operations.md](docs/operations.md) を参照。 diff --git a/docs/operations.md b/docs/operations.md new file mode 100644 index 0000000..0d4d469 --- /dev/null +++ b/docs/operations.md @@ -0,0 +1,33 @@ +# 運用仕様メモ + +drive-foundation レビューで確定した運用上の仕様ポイントをまとめる。実装の根拠は core/src 配下の各モジュール (`public.rs`, `state.rs`, `auth/rate_limit.rs`, `billing/cache.rs`, `drive/files.rs`, `drive/upload_body.rs`, `config.rs`)。 + +## `/public/{key}` は無認証の公開エンドポイント + +- 認証なしで誰でもダウンロードできる。帯域計量 (レスポンス転送量の計測・課金) は初期スコープ外。 +- DoS 緩和として per-IP レート制限を付与している。既定 300 req/min で、環境変数 `BOOSKIFF_PUBLIC_RATE_LIMIT_RPM` で変更できる。 +- 制限キーはクライアント IP (`ConnectInfo`)。非 TCP リスナ等で ConnectInfo が取得できない環境では制限をスキップする (fail-open)。 + +## 公開ファイルのキャッシュと失効 + +- 公開レスポンスには `Cache-Control: public, max-age=31536000, immutable` を返す。 +- unpublish / 削除しても、ブラウザや CDN 等の外部キャッシュにコンテンツが残り得る。外部キャッシュの即時失効は保証しない。 +- 緩和の仕組み: `publish_file` は publish のたびに public_key を乱数で再生成する。unpublish で public_key は NULL になるため旧キーの URL は 404 になり、再 publish すると別キーの URL に変わる。 + +## レート制限 state と課金キャッシュはプロセス内メモリ + +- owner 用 `RateLimiters`、public 用 `PublicRateLimiter`、課金解決キャッシュ `BillingCache` はすべてプロセス内メモリに保持し、複数レプリカ間で共有されない。レプリカごとに実効制限 (req/min やキャッシュ鮮度) が発散し得る。 +- rpm が変わると対象キーの limiter を再生成するため、消費済み分はリセットされ burst が復活する。既存トークンバケットの残量を新 quota と調停するコストを避ける、意図的な設計上の制約。 +- `BillingCache` の TTL は既定 60 秒。環境変数 `BOOSKIFF_BILLING_CACHE_TTL_SECS` で変更でき、0 を指定するとキャッシュ自体を無効化 (保存をスキップ) する。管理者 API による書き込み (課金ルール・割当・プランの変更) 時は、対象 owner のエントリまたは全体を即時無効化する。 + +## ボディサイズ強制 + +- アップロードのボディサイズ強制は `CountingBody` (`drive/upload_body.rs`) が唯一の経路。`upload_file` が axum の `Body` を直接抽出するため `DefaultBodyLimit` レイヤは no-op になり、撤去済み。 +- 実測バイト数 (counted) が `max_file_bytes` を超えた場合は 413 `payload_too_large`。宣言 Content-Length を超えた場合は 400 `validation`。 + +## 環境変数 + +| 変数 | 既定値 | 意味 | +| --- | --- | --- | +| `BOOSKIFF_PUBLIC_RATE_LIMIT_RPM` | 300 | `/public/{key}` の per-IP レート制限 (req/min) | +| `BOOSKIFF_BILLING_CACHE_TTL_SECS` | 60 | 課金解決キャッシュの TTL (秒)。0 で無効 | From c0dfd7f255ffa903f2334db2d12d2b3f8493330a Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:02:15 +0900 Subject: [PATCH 07/13] =?UTF-8?q?feat(public):=20/public/{key}=20=E3=81=AB?= =?UTF-8?q?=20per-IP=20=E3=83=AC=E3=83=BC=E3=83=88=E5=88=B6=E9=99=90=20(?= =?UTF-8?q?=E6=97=A2=E5=AE=9A=20300=20rpm)=20=E3=82=92=E4=BB=98=E4=B8=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/admin/billing_cache_tests.rs | 1 + core/src/admin/handlers.rs | 3 ++ core/src/auth/admin_auth.rs | 2 + core/src/auth/extractor.rs | 2 + core/src/auth/rate_limit.rs | 54 +++++++++++++++++++- core/src/config.rs | 7 +++ core/src/drive/files.rs | 3 ++ core/src/main.rs | 10 +++- core/src/public.rs | 72 ++++++++++++++++++++++++++- core/src/state.rs | 3 +- 10 files changed, 153 insertions(+), 4 deletions(-) diff --git a/core/src/admin/billing_cache_tests.rs b/core/src/admin/billing_cache_tests.rs index 465823c..5041c3f 100644 --- a/core/src/admin/billing_cache_tests.rs +++ b/core/src/admin/billing_cache_tests.rs @@ -33,6 +33,7 @@ async fn pg_admin_billing_writes_invalidate_cache() { pool: pool.clone(), config, rate_limiters: Arc::new(crate::state::RateLimiters::default()), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(300)), billing_cache: Arc::clone(&cache), }; let router = admin_router().with_state(state); diff --git a/core/src/admin/handlers.rs b/core/src/admin/handlers.rs index 714bbb4..0ec7a19 100644 --- a/core/src/admin/handlers.rs +++ b/core/src/admin/handlers.rs @@ -621,6 +621,9 @@ mod tests { )), config, rate_limiters: std::sync::Arc::new(RateLimiters::default()), + public_rate_limiter: std::sync::Arc::new( + crate::auth::rate_limit::PublicRateLimiter::new(300), + ), }; let router = axum::Router::new() .nest("/v1/admin", admin_router()) diff --git a/core/src/auth/admin_auth.rs b/core/src/auth/admin_auth.rs index a28bab7..5d332ad 100644 --- a/core/src/auth/admin_auth.rs +++ b/core/src/auth/admin_auth.rs @@ -145,6 +145,7 @@ mod tests { )), config, rate_limiters: Arc::new(RateLimiters::default()), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(300)), } } @@ -212,6 +213,7 @@ mod tests { )), config, rate_limiters: Arc::new(RateLimiters::default()), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(300)), }; let router = admin_router(state.clone()); diff --git a/core/src/auth/extractor.rs b/core/src/auth/extractor.rs index 999aa80..b995582 100644 --- a/core/src/auth/extractor.rs +++ b/core/src/auth/extractor.rs @@ -150,6 +150,7 @@ mod tests { )), config, rate_limiters: Arc::new(RateLimiters::default()), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(300)), } } @@ -177,6 +178,7 @@ mod tests { )), config, rate_limiters: Arc::new(RateLimiters::default()), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(300)), } } diff --git a/core/src/auth/rate_limit.rs b/core/src/auth/rate_limit.rs index 34fad13..2133b3a 100644 --- a/core/src/auth/rate_limit.rs +++ b/core/src/auth/rate_limit.rs @@ -1,7 +1,37 @@ -//! Per-owner rate limiting backed by governor. +//! Per-owner and public per-IP rate limiting backed by governor. use governor::{DefaultDirectRateLimiter, Quota, RateLimiter}; +use std::net::IpAddr; use std::num::NonZeroU32; +use std::sync::atomic::{AtomicUsize, Ordering}; + +/// Process-local public budgets; replicas do not share their counters. +/// Every 10,000 checks evict fully replenished entries so a spray of distinct +/// IPs does not accumulate stale DashMap entries indefinitely. Active entries +/// remain until replenished; this is periodic reclamation, not a hard size cap. +pub struct PublicRateLimiter { + limiter: governor::DefaultKeyedRateLimiter, + checks_since_sweep: AtomicUsize, +} + +impl PublicRateLimiter { + /// `rpm` of 0 is clamped to 1, matching the owner limiter. + pub fn new(rpm: u32) -> Self { + let quota = Quota::per_minute(NonZeroU32::new(rpm).unwrap_or(NonZeroU32::MIN)); + Self { + limiter: RateLimiter::keyed(quota), + checks_since_sweep: AtomicUsize::new(0), + } + } + + pub fn check(&self, ip: IpAddr) -> bool { + // This standalone sweep counter does not publish other shared state. + if self.checks_since_sweep.fetch_add(1, Ordering::Relaxed) % 10_000 == 9_999 { + self.limiter.retain_recent(); + } + self.limiter.check_key(&ip).is_ok() + } +} /// A fixed-window-ish limiter allowing `rpm` requests per minute with full /// burst capacity available immediately. @@ -30,6 +60,28 @@ impl OwnerRateLimiter { mod tests { use super::*; + #[test] + fn public_limiter_allows_burst_then_denies_when_budget_exhausted() { + // Given a two-request budget for one IP. + let limiter = PublicRateLimiter::new(2); + let ip = std::net::Ipv4Addr::LOCALHOST.into(); + // When a burst exceeds that budget. + let admitted = [limiter.check(ip), limiter.check(ip), limiter.check(ip)]; + // Then only the first two requests are admitted. + assert_eq!(admitted, [true, true, false]); + } + + #[test] + fn public_limiter_clamps_quota_when_rpm_is_zero() { + // Given a zero configured quota. + let limiter = PublicRateLimiter::new(0); + let ip = std::net::Ipv6Addr::LOCALHOST.into(); + // When two requests arrive. + let admitted = [limiter.check(ip), limiter.check(ip)]; + // Then the effective quota is one. + assert_eq!(admitted, [true, false]); + } + #[test] fn limiter_allows_exactly_burst_then_denies() { let limiter = OwnerRateLimiter::new(2); diff --git a/core/src/config.rs b/core/src/config.rs index 7816a11..66bb96a 100644 --- a/core/src/config.rs +++ b/core/src/config.rs @@ -63,6 +63,7 @@ pub struct Config { pub plan_premium_rate_limit_rpm: u32, pub presigned_get_ttl_secs: u64, pub public_base_url: String, + pub public_rate_limit_rpm: u32, /// Raw admin token seeded into `admin_tokens` (name `bootstrap`) on /// startup when no active token with that hash exists yet. pub admin_bootstrap_token: Option, @@ -93,6 +94,7 @@ impl Default for Config { plan_premium_rate_limit_rpm: 300, presigned_get_ttl_secs: 900, public_base_url: "http://localhost:3000".into(), + public_rate_limit_rpm: 300, admin_bootstrap_token: None, } } @@ -182,6 +184,11 @@ impl Config { "presigned_get_ttl_secs", )?, public_base_url: override_str(d.public_base_url, "BOOSKIFF_PUBLIC_BASE_URL"), + public_rate_limit_rpm: override_parse( + d.public_rate_limit_rpm, + "BOOSKIFF_PUBLIC_RATE_LIMIT_RPM", + "public_rate_limit_rpm", + )?, admin_bootstrap_token: override_opt_str( d.admin_bootstrap_token, "BOOSKIFF_ADMIN_BOOTSTRAP_TOKEN", diff --git a/core/src/drive/files.rs b/core/src/drive/files.rs index 1b01851..894668a 100644 --- a/core/src/drive/files.rs +++ b/core/src/drive/files.rs @@ -728,6 +728,9 @@ mod tests { s3, config, rate_limiters: std::sync::Arc::new(crate::state::RateLimiters::default()), + public_rate_limiter: std::sync::Arc::new( + crate::auth::rate_limit::PublicRateLimiter::new(300), + ), } } diff --git a/core/src/main.rs b/core/src/main.rs index ba99552..2ec127f 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -101,6 +101,9 @@ async fn run() { billing_cache: Arc::new(core::billing::cache::BillingCache::new( config.billing_cache_ttl_secs, )), + public_rate_limiter: Arc::new(core::auth::rate_limit::PublicRateLimiter::new( + config.public_rate_limit_rpm, + )), config, jwks_cache, rate_limiters: Arc::new(RateLimiters::default()), @@ -141,7 +144,12 @@ async fn run() { std::process::exit(1); }); tracing::info!("listening on http://{listen_addr}"); - if let Err(err) = axum::serve(listener, app).await { + if let Err(err) = axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + { eprintln!("server error: {err}"); std::process::exit(1); } diff --git a/core/src/public.rs b/core/src/public.rs index 925c4c1..add4f12 100644 --- a/core/src/public.rs +++ b/core/src/public.rs @@ -2,11 +2,12 @@ use axum::Router; use axum::body::Body; -use axum::extract::{Path, State}; +use axum::extract::{ConnectInfo, Path, State}; use axum::http::header::{CACHE_CONTROL, CONTENT_LENGTH, CONTENT_TYPE}; use axum::http::{HeaderMap, HeaderValue, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::routing::get; +use std::net::SocketAddr; use crate::error::AppError; use crate::model::OBJECT_KIND_ORIGINAL; @@ -42,7 +43,17 @@ pub fn public_router() -> Router { async fn get_public_file( State(state): State, Path(key): Path, + connect_info: Result, axum::extract::rejection::ExtensionRejection>, ) -> Result { + let admitted = match connect_info { + Ok(ConnectInfo(addr)) => state.public_rate_limiter.check(addr.ip()), + // Deliberately bypass limiting for oneshot tests or non-TCP listeners + // without ConnectInfo. The production TCP server always injects it. + Err(_) => true, + }; + if !admitted { + return Err(AppError::RateLimited); + } let file = sqlx::query_as::<_, PublicFile>( "SELECT id FROM files WHERE public_key = $1 AND is_public = TRUE", ) @@ -82,6 +93,65 @@ async fn get_public_file( #[cfg(test)] mod tests { use super::IMMUTABLE_CACHE_CONTROL; + use super::*; + use axum::extract::ConnectInfo; + use axum::http::Request; + use std::net::SocketAddr; + use std::sync::Arc; + use tower::ServiceExt; + + #[tokio::test] + async fn public_router_limits_each_ip_and_bypasses_missing_connect_info() { + // Given a closed lazy pool and a two-request public budget. + let config = crate::config::Config::default(); + let pool = sqlx::PgPool::connect_lazy("postgres://127.0.0.1:9/never").unwrap(); + pool.close().await; + let state = AppState { + pool, + s3: crate::storage::Storage::build(&config).await.unwrap(), + jwks_cache: crate::auth::jwks::JwksCache::new(Vec::new()), + rate_limiters: Arc::new(crate::state::RateLimiters::default()), + billing_cache: Arc::new(crate::billing::cache::BillingCache::new(60)), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(2)), + config, + }; + let app = public_router().with_state(state); + let first_ip = SocketAddr::from(([192, 0, 2, 1], 1234)); + let same_ip_new_port = SocketAddr::from(([192, 0, 2, 1], 5678)); + let other_ip = SocketAddr::from(([192, 0, 2, 2], 1234)); + + // When requests exhaust one IP, change IP, then omit connection info. + for (addr, limited) in [ + (Some(first_ip), false), + (Some(same_ip_new_port), false), + (Some(first_ip), true), + (Some(other_ip), false), + (None, false), + (None, false), + (None, false), + ] { + let mut request = Request::builder() + .uri("/public/test-key") + .body(Body::empty()) + .unwrap(); + if let Some(addr) = addr { + request.extensions_mut().insert(ConnectInfo(addr)); + } + let response = app.clone().oneshot(request).await.unwrap(); + + // Then only the exhausted IP receives the standard 429 envelope. + if limited { + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + let bytes = axum::body::to_bytes(response.into_body(), 4096) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["error"]["code"], "rate_limited"); + } else { + assert_ne!(response.status(), StatusCode::TOO_MANY_REQUESTS); + } + } + } #[test] fn immutable_cache_control_is_public_and_one_year() { diff --git a/core/src/state.rs b/core/src/state.rs index bca36ac..f19d7fd 100644 --- a/core/src/state.rs +++ b/core/src/state.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; use crate::auth::jwks::JwksCache; -use crate::auth::rate_limit::OwnerRateLimiter; +use crate::auth::rate_limit::{OwnerRateLimiter, PublicRateLimiter}; use crate::config::Config; use crate::storage::Storage; @@ -75,6 +75,7 @@ pub struct AppState { pub config: Config, pub jwks_cache: JwksCache, pub rate_limiters: Arc, + pub public_rate_limiter: Arc, pub billing_cache: Arc, } From a91758ae52a2c14796301e5815dac72d5166568e Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:17:31 +0900 Subject: [PATCH 08/13] =?UTF-8?q?test(drive):=20=E3=83=95=E3=82=A9?= =?UTF-8?q?=E3=83=AB=E3=83=80=E5=89=8A=E9=99=A4=E3=81=A8=E3=82=A2=E3=83=83?= =?UTF-8?q?=E3=83=97=E3=83=AD=E3=83=BC=E3=83=89=E3=81=AE=E5=90=8C=E6=99=82?= =?UTF-8?q?=E7=AB=B6=E5=90=88=E3=81=A7=20500=20=E3=82=92=E8=BF=94=E3=81=95?= =?UTF-8?q?=E3=81=AA=E3=81=84=E3=81=93=E3=81=A8=E3=82=92=E5=9B=BA=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- core/src/drive/files.rs | 72 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/core/src/drive/files.rs b/core/src/drive/files.rs index 894668a..1c587ea 100644 --- a/core/src/drive/files.rs +++ b/core/src/drive/files.rs @@ -386,6 +386,12 @@ fn parse_content_length(headers: &HeaderMap) -> Result { Ok(declared) } +/// Checks folder ownership only; it does not create folders. +/// +/// The originally reviewed concurrent-folder-creation race therefore does not +/// exist in this code. The live race is folder deletion after this check but +/// before the file INSERT; the resulting FK violation is mapped to `NotFound` +/// by `map_insert_file_error` and covered by both vanished-folder tests below. async fn ensure_owned_folder( pool: &sqlx::PgPool, owner: &Owner, @@ -621,6 +627,72 @@ mod tests { assert!(matches!(error, AppError::NotFound(_))); } + #[tokio::test] + #[ignore = "requires the compose postgres stack"] + async fn pg_concurrent_folder_delete_and_upload_never_500() { + // Given: an owner and a real folder in PostgreSQL. + let config = Config::default(); + let pool = test_pool().await; + let storage = Storage::build(&config).await.unwrap(); + let owner = Owner::new("test-drive-files-race", Uuid::now_v7().to_string()); + let ctx = AccountContext { + owner: owner.clone(), + limits: crate::model::Limits { + storage_quota_bytes: 1024, + max_file_bytes: 1024, + rate_limit_rpm: 100, + }, + }; + let state = test_state(config, pool.clone(), storage).await; + + for round in 0..8 { + let folder_id: Uuid = sqlx::query_scalar( + "INSERT INTO folders (owner_type, owner_id, name) VALUES ($1, $2, $3) RETURNING id", + ) + .bind(&owner.owner_type) + .bind(&owner.owner_id) + .bind(format!("race-{round}")) + .fetch_one(&pool) + .await + .unwrap(); + + // When: file INSERT and folder DELETE run concurrently. + let (upload_result, delete_result) = tokio::join!( + persist_uploaded_file( + &state, + &ctx, + UploadedFile { + id: Uuid::now_v7(), + folder_id: Some(folder_id), + name: format!("race-{round}.txt"), + mime: "text/plain".into(), + key: format!("test-drive-files/race-{round}.txt"), + size_bytes: 1, + }, + ), + sqlx::query("DELETE FROM folders WHERE id = $1") + .bind(folder_id) + .execute(&pool), + ); + + delete_result.unwrap(); + + // Then: deletion may win (404) or insertion may win (success), but + // the FK race must never escape as an internal error (500). + match upload_result { + Ok(_) | Err(AppError::NotFound(_)) => {} + Err(error) => panic!("concurrent upload returned unexpected error: {error:?}"), + } + + sqlx::query("DELETE FROM files WHERE owner_type = $1 AND owner_id = $2") + .bind(&owner.owner_type) + .bind(&owner.owner_id) + .execute(&pool) + .await + .unwrap(); + } + } + #[test] fn foreign_file_row_maps_to_not_found() { // Given From 60781875e485086e75aec3481003cee1f279e18a Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:18:10 +0900 Subject: [PATCH 09/13] =?UTF-8?q?fix(admin):=20=E7=A9=BA=E3=81=AE=20owner?= =?UTF-8?q?=20scope=20=E3=82=92=E6=8B=92=E5=90=A6=E3=81=97=E3=81=A6=20glob?= =?UTF-8?q?al=20=E3=83=AB=E3=83=BC=E3=83=AB=E8=A1=9D=E7=AA=81=E3=82=92?= =?UTF-8?q?=E9=98=B2=E3=81=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/admin/billing_cache_tests.rs | 63 +++++++++++++++++++++++++-- core/src/admin/handlers.rs | 22 ++++++++-- 2 files changed, 77 insertions(+), 8 deletions(-) diff --git a/core/src/admin/billing_cache_tests.rs b/core/src/admin/billing_cache_tests.rs index 5041c3f..8c108c9 100644 --- a/core/src/admin/billing_cache_tests.rs +++ b/core/src/admin/billing_cache_tests.rs @@ -60,9 +60,10 @@ async fn pg_admin_billing_writes_invalidate_cache() { ("DELETE", String::new(), serde_json::Value::Null, true), ] { cache.store_global(vec![]); - for key in [owner.key(), "other".into()] { + let other = Owner::new("other", "owner"); + for cached_owner in [&owner, &other] { cache.store_owner( - &key, + cached_owner, OwnerEntry { assignment: None, rules: vec![], @@ -88,10 +89,10 @@ async fn pg_admin_billing_writes_invalidate_cache() { response.status() ); assert!( - cache.get_owner(&owner.key()).is_none(), + cache.get_owner(&owner).is_none(), "{method} must invalidate owner" ); - assert_eq!(cache.get_owner("other").is_none(), all); + assert_eq!(cache.get_owner(&other).is_none(), all); assert_eq!(cache.get_global().is_none(), all); if method == "POST" { let bytes = response.into_body().collect().await.unwrap().to_bytes(); @@ -99,6 +100,60 @@ async fn pg_admin_billing_writes_invalidate_cache() { rule_id = item["id"].as_str().unwrap().to_owned(); } } + + sqlx::query( + "INSERT INTO billing_rules (owner_type, owner_id, key, value, enabled) \ + VALUES (NULL, NULL, $1, $2, true) \ + ON CONFLICT ((COALESCE(owner_type, '')), (COALESCE(owner_id, '')), key) \ + DO UPDATE SET value = EXCLUDED.value, enabled = EXCLUDED.enabled", + ) + .bind("max_file_bytes") + .bind(serde_json::json!(72)) + .execute(&pool) + .await + .unwrap(); + cache.store_global(vec![("max_file_bytes".to_owned(), serde_json::json!(72))]); + let request = Request::builder() + .method("POST") + .uri("/billing/rules") + .header("x-admin-token", &token) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ + "owner_type": "", + "owner_id": owner.owner_id, + "key": "max_file_bytes", + "value": 999, + "enabled": true + }) + .to_string(), + )) + .unwrap(); + let response = router.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["error"]["code"], "validation"); + assert!( + body["error"]["message"] + .as_str() + .unwrap() + .contains("owner_type must not be blank") + ); + + let global_value: serde_json::Value = sqlx::query_scalar( + "SELECT value FROM billing_rules WHERE owner_type IS NULL AND owner_id IS NULL AND key = $1", + ) + .bind("max_file_bytes") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(global_value, serde_json::json!(72)); + assert_eq!( + cache.get_global(), + Some(vec![("max_file_bytes".to_owned(), serde_json::json!(72))]) + ); + sqlx::query("DELETE FROM admin_tokens WHERE name = $1") .bind(&token) .execute(&pool) diff --git a/core/src/admin/handlers.rs b/core/src/admin/handlers.rs index 0ec7a19..1bb5ed8 100644 --- a/core/src/admin/handlers.rs +++ b/core/src/admin/handlers.rs @@ -131,7 +131,19 @@ fn owner_scope( ) -> Result, AppError> { match (owner_type, owner_id) { (None, None) => Ok(None), - (Some(owner_type), Some(owner_id)) => Ok(Some(Owner::new(owner_type, owner_id))), + (Some(owner_type), Some(owner_id)) => { + if owner_type.trim().is_empty() { + return Err(AppError::Validation( + "owner_type must not be blank".to_owned(), + )); + } + if owner_id.trim().is_empty() { + return Err(AppError::Validation( + "owner_id must not be blank".to_owned(), + )); + } + Ok(Some(Owner::new(owner_type, owner_id))) + } (Some(_), None) | (None, Some(_)) => Err(AppError::Validation( "owner_type and owner_id must be given together; omit both for a global rule" .to_owned(), @@ -294,7 +306,7 @@ async fn create_billing_rule( ) .await?; match scope { - Some(owner) => state.billing_cache.invalidate_owner(&owner.key()), + Some(owner) => state.billing_cache.invalidate_owner(&owner), None => state.billing_cache.invalidate_all(), } Ok(Json(billing_rule_item(rule)?)) @@ -351,7 +363,7 @@ async fn set_plan( let plan = parse_plan(&request.plan)?; let owner = Owner::new(owner_type, owner_id); assignments::set_plan(&state.pool, &owner, plan).await?; - state.billing_cache.invalidate_owner(&owner.key()); + state.billing_cache.invalidate_owner(&owner); Ok(StatusCode::NO_CONTENT) } @@ -403,7 +415,7 @@ async fn delete_plan( ) -> Result { let owner = Owner::new(owner_type, owner_id); if assignments::delete_plan(&state.pool, &owner).await? { - state.billing_cache.invalidate_owner(&owner.key()); + state.billing_cache.invalidate_owner(&owner); Ok(StatusCode::NO_CONTENT) } else { Err(AppError::NotFound(format!( @@ -459,6 +471,8 @@ pub fn admin_router() -> Router { ) .route("/billing/rules/{id}", delete(delete_billing_rule)) .route( + // Axum path parameters require non-empty segments, so these plan + // routes cannot receive an empty owner component. "/owners/{owner_type}/{owner_id}/plan", put(set_plan).get(get_plan).delete(delete_plan), ) From 6bceebd105802848e2d445e0000d2afc43f7ca70 Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:19:20 +0900 Subject: [PATCH 10/13] =?UTF-8?q?fix(billing):=20=E8=AA=B2=E9=87=91?= =?UTF-8?q?=E3=82=AD=E3=83=A3=E3=83=83=E3=82=B7=E3=83=A5=E3=81=AE=20owner?= =?UTF-8?q?=20=E3=82=AD=E3=83=BC=E3=82=92=E6=A7=8B=E9=80=A0=E5=8C=96?= =?UTF-8?q?=E3=81=97=E3=81=A6=E3=83=86=E3=83=8A=E3=83=B3=E3=83=88=E9=96=93?= =?UTF-8?q?=E8=A1=9D=E7=AA=81=E3=82=92=E9=98=B2=E3=81=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/billing/cache.rs | 114 +++++++++++++++++++++++++----------- core/src/billing/resolve.rs | 7 +-- 2 files changed, 84 insertions(+), 37 deletions(-) diff --git a/core/src/billing/cache.rs b/core/src/billing/cache.rs index edd8974..151b2d5 100644 --- a/core/src/billing/cache.rs +++ b/core/src/billing/cache.rs @@ -4,12 +4,14 @@ use std::collections::HashMap; use std::sync::Mutex; use std::time::{Duration, Instant}; +use crate::model::Owner; + type Rules = Vec<(String, serde_json::Value)>; pub struct BillingCache { ttl: Duration, global: Mutex>>, - owners: Mutex>>, + owners: Mutex>>, generation: Mutex>, } @@ -46,10 +48,10 @@ impl BillingCache { .map(|entry| entry.value.clone()) } - pub fn get_owner(&self, key: &str) -> Option { + pub fn get_owner(&self, owner: &Owner) -> Option { let owners = self.owners.lock().unwrap_or_else(|err| err.into_inner()); owners - .get(key) + .get(&(owner.owner_type.clone(), owner.owner_id.clone())) .filter(|entry| entry.fetched_at.elapsed() < self.ttl) .map(|entry| entry.value.clone()) } @@ -82,13 +84,13 @@ impl BillingCache { }); } - pub fn store_owner(&self, key: &str, value: OwnerEntry) { - self.store_owner_if_current(key, value, &self.generation()); + pub fn store_owner(&self, owner: &Owner, value: OwnerEntry) { + self.store_owner_if_current(owner, value, &self.generation()); } pub(crate) fn store_owner_if_current( &self, - key: &str, + owner: &Owner, value: OwnerEntry, generation: &std::sync::Arc<()>, ) { @@ -103,11 +105,12 @@ impl BillingCache { return; } let mut owners = self.owners.lock().unwrap_or_else(|err| err.into_inner()); - if owners.len() >= MAX_CACHED_OWNERS && !owners.contains_key(key) { + let key = (owner.owner_type.clone(), owner.owner_id.clone()); + if owners.len() >= MAX_CACHED_OWNERS && !owners.contains_key(&key) { owners.clear(); } owners.insert( - key.to_owned(), + key, Cached { value, fetched_at: Instant::now(), @@ -115,7 +118,7 @@ impl BillingCache { ); } - pub fn invalidate_owner(&self, key: &str) { + pub fn invalidate_owner(&self, owner: &Owner) { let mut generation = self .generation .lock() @@ -124,7 +127,7 @@ impl BillingCache { self.owners .lock() .unwrap_or_else(|err| err.into_inner()) - .remove(key); + .remove(&(owner.owner_type.clone(), owner.owner_id.clone())); } pub fn invalidate_all(&self) { @@ -155,85 +158,130 @@ mod tests { } } + fn owner(owner_type: &str, owner_id: &str) -> Owner { + Owner::new(owner_type, owner_id) + } + #[test] fn fresh_entries_expire() { let cache = BillingCache::new(60); + let owner = owner("account", "a"); cache.store_global(vec![]); - cache.store_owner("a", entry()); + cache.store_owner(&owner, entry()); assert!(cache.get_global().is_some()); - assert!(cache.get_owner("a").is_some()); + assert!(cache.get_owner(&owner).is_some()); cache.global.lock().unwrap().as_mut().unwrap().fetched_at -= Duration::from_secs(61); cache .owners .lock() .unwrap() - .get_mut("a") + .get_mut(&(owner.owner_type.clone(), owner.owner_id.clone())) .unwrap() .fetched_at -= Duration::from_secs(61); assert!(cache.get_global().is_none()); - assert!(cache.get_owner("a").is_none()); + assert!(cache.get_owner(&owner).is_none()); } #[test] fn invalidate_owner_preserves_other_layers() { let cache = BillingCache::new(60); + let first = owner("account", "a"); + let second = owner("account", "b"); cache.store_global(vec![]); - cache.store_owner("a", entry()); - cache.store_owner("b", entry()); - cache.invalidate_owner("a"); - assert!(cache.get_owner("a").is_none()); - assert!(cache.get_owner("b").is_some()); + cache.store_owner(&first, entry()); + cache.store_owner(&second, entry()); + cache.invalidate_owner(&first); + assert!(cache.get_owner(&first).is_none()); + assert!(cache.get_owner(&second).is_some()); assert!(cache.get_global().is_some()); } + #[test] + fn owner_keys_are_structured_and_collision_free() { + let cache = BillingCache::new(60); + let first = owner("a:b", "c"); + let second = owner("a", "b:c"); + let first_entry = OwnerEntry { + assignment: Some(crate::model::Plan::Default), + rules: vec![], + }; + let second_entry = entry(); + + cache.store_owner(&first, first_entry.clone()); + cache.store_owner(&second, second_entry.clone()); + + assert_eq!( + cache.get_owner(&first).unwrap().assignment, + first_entry.assignment + ); + assert_eq!( + cache.get_owner(&second).unwrap().assignment, + second_entry.assignment + ); + + cache.invalidate_owner(&first); + assert!(cache.get_owner(&first).is_none()); + assert_eq!( + cache.get_owner(&second).unwrap().assignment, + second_entry.assignment + ); + } + #[test] fn invalidate_all_clears_both_layers() { let cache = BillingCache::new(60); + let owner = owner("account", "a"); cache.store_global(vec![]); - cache.store_owner("a", entry()); + cache.store_owner(&owner, entry()); cache.invalidate_all(); assert!(cache.get_global().is_none()); - assert!(cache.get_owner("a").is_none()); + assert!(cache.get_owner(&owner).is_none()); } #[test] fn zero_ttl_disables_caching() { let cache = BillingCache::new(0); + let owner = owner("account", "a"); cache.store_global(vec![]); - cache.store_owner("a", entry()); + cache.store_owner(&owner, entry()); assert!(cache.get_global().is_none()); - assert!(cache.get_owner("a").is_none()); + assert!(cache.get_owner(&owner).is_none()); assert!(cache.owners.lock().unwrap().is_empty()); } #[test] fn invalidation_rejects_in_flight_fills() { let cache = BillingCache::new(60); + let owner = owner("account", "a"); let generation = cache.generation(); - cache.invalidate_owner("a"); - cache.store_owner_if_current("a", entry(), &generation); - assert!(cache.get_owner("a").is_none()); + cache.invalidate_owner(&owner); + cache.store_owner_if_current(&owner, entry(), &generation); + assert!(cache.get_owner(&owner).is_none()); let generation = cache.generation(); cache.invalidate_all(); cache.store_global_if_current(vec![], &generation); - cache.store_owner_if_current("a", entry(), &generation); + cache.store_owner_if_current(&owner, entry(), &generation); assert!(cache.get_global().is_none()); - assert!(cache.get_owner("a").is_none()); + assert!(cache.get_owner(&owner).is_none()); } #[test] fn overflow_clears_owners_but_not_global() { let cache = BillingCache::new(60); + let owner_for_zero = |value: String| Owner::new("owner", value); cache.store_global(vec![]); for i in 0..MAX_CACHED_OWNERS { - cache.store_owner(&i.to_string(), entry()); + let owner = owner_for_zero(i.to_string()); + cache.store_owner(&owner, entry()); } - cache.store_owner("0", entry()); + let zero = owner_for_zero("0".to_owned()); + cache.store_owner(&zero, entry()); assert_eq!(cache.owners.lock().unwrap().len(), MAX_CACHED_OWNERS); - cache.store_owner("new", entry()); + let new_owner = owner_for_zero("new".to_owned()); + cache.store_owner(&new_owner, entry()); assert_eq!(cache.owners.lock().unwrap().len(), 1); - assert!(cache.get_owner("0").is_none()); - assert!(cache.get_owner("new").is_some()); + assert!(cache.get_owner(&zero).is_none()); + assert!(cache.get_owner(&new_owner).is_some()); assert!(cache.get_global().is_some()); } } diff --git a/core/src/billing/resolve.rs b/core/src/billing/resolve.rs index f70d73d..805d1b7 100644 --- a/core/src/billing/resolve.rs +++ b/core/src/billing/resolve.rs @@ -57,8 +57,7 @@ pub async fn effective_limits_cached( rules } }; - let key = owner.key(); - let entry = match cache.get_owner(&key) { + let entry = match cache.get_owner(owner) { Some(entry) => entry, None => { let assignment = match config.premium_mode { @@ -67,7 +66,7 @@ pub async fn effective_limits_cached( }; let rules = load_rules(pool, OWNER_RULES_SQL, Some(owner)).await?; let entry = crate::billing::cache::OwnerEntry { assignment, rules }; - cache.store_owner_if_current(&key, entry.clone(), &generation); + cache.store_owner_if_current(owner, entry.clone(), &generation); entry } }; @@ -443,7 +442,7 @@ mod tests { stale.storage_quota_bytes, config.plan_default_storage_quota_bytes ); - cache.invalidate_owner(&owner.key()); + cache.invalidate_owner(&owner); let fresh = effective_limits_cached(&cache, &pool, &config, &owner) .await .unwrap(); From 6f3e62c451af058457c70894cdeba96958f58ad5 Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:20:04 +0900 Subject: [PATCH 11/13] =?UTF-8?q?feat(public):=20=E4=BF=A1=E9=A0=BC?= =?UTF-8?q?=E3=83=97=E3=83=AD=E3=82=AD=E3=82=B7=E8=83=8C=E5=BE=8C=E3=81=A7?= =?UTF-8?q?=20X-Forwarded-For=20=E3=81=AE=E3=82=AF=E3=83=A9=E3=82=A4?= =?UTF-8?q?=E3=82=A2=E3=83=B3=E3=83=88=20IP=20=E3=82=92=E5=88=B6=E9=99=90?= =?UTF-8?q?=E3=82=AD=E3=83=BC=E3=81=AB=E4=BD=BF=E3=81=88=E3=82=8B=E3=82=88?= =?UTF-8?q?=E3=81=86=E3=81=AB=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/config.rs | 7 +++ core/src/public.rs | 37 ++++++++++-- core/src/public/proxy_tests.rs | 102 ++++++++++++++++++++++++++++++++ deploy/self-hosting/compose.yml | 4 ++ docs/operations.md | 6 +- 5 files changed, 149 insertions(+), 7 deletions(-) create mode 100644 core/src/public/proxy_tests.rs diff --git a/core/src/config.rs b/core/src/config.rs index 66bb96a..a386006 100644 --- a/core/src/config.rs +++ b/core/src/config.rs @@ -64,6 +64,7 @@ pub struct Config { pub presigned_get_ttl_secs: u64, pub public_base_url: String, pub public_rate_limit_rpm: u32, + pub trust_proxy_headers: bool, /// Raw admin token seeded into `admin_tokens` (name `bootstrap`) on /// startup when no active token with that hash exists yet. pub admin_bootstrap_token: Option, @@ -95,6 +96,7 @@ impl Default for Config { presigned_get_ttl_secs: 900, public_base_url: "http://localhost:3000".into(), public_rate_limit_rpm: 300, + trust_proxy_headers: false, admin_bootstrap_token: None, } } @@ -193,6 +195,11 @@ impl Config { d.admin_bootstrap_token, "BOOSKIFF_ADMIN_BOOTSTRAP_TOKEN", ), + trust_proxy_headers: override_parse( + d.trust_proxy_headers, + "BOOSKIFF_TRUST_PROXY_HEADERS", + "trust_proxy_headers", + )?, }) } } diff --git a/core/src/public.rs b/core/src/public.rs index add4f12..16ac918 100644 --- a/core/src/public.rs +++ b/core/src/public.rs @@ -7,7 +7,7 @@ use axum::http::header::{CACHE_CONTROL, CONTENT_LENGTH, CONTENT_TYPE}; use axum::http::{HeaderMap, HeaderValue, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::routing::get; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use crate::error::AppError; use crate::model::OBJECT_KIND_ORIGINAL; @@ -15,6 +15,9 @@ use crate::state::AppState; const IMMUTABLE_CACHE_CONTROL: &str = "public, max-age=31536000, immutable"; +#[cfg(test)] +mod proxy_tests; + #[derive(Debug, sqlx::FromRow)] struct PublicFile { id: uuid::Uuid, @@ -30,6 +33,23 @@ pub fn public_router() -> Router { Router::new().route("/public/{key}", get(get_public_file)) } +fn resolve_client_ip( + trust_proxy_headers: bool, + forwarded_for: Option<&HeaderValue>, + peer: Option, +) -> Option { + // This opt-in gate is the trust boundary: enable only behind a trusted proxy + // that sanitizes X-Forwarded-For, never on a directly exposed listener. + if trust_proxy_headers { + forwarded_for + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(',').find_map(|ip| ip.trim().parse().ok())) + .or(peer) + } else { + peer + } +} + #[utoipa::path( get, path = "/public/{key}", @@ -44,12 +64,17 @@ async fn get_public_file( State(state): State, Path(key): Path, connect_info: Result, axum::extract::rejection::ExtensionRejection>, + headers: HeaderMap, ) -> Result { - let admitted = match connect_info { - Ok(ConnectInfo(addr)) => state.public_rate_limiter.check(addr.ip()), - // Deliberately bypass limiting for oneshot tests or non-TCP listeners - // without ConnectInfo. The production TCP server always injects it. - Err(_) => true, + let peer = connect_info.ok().map(|ConnectInfo(addr)| addr.ip()); + let admitted = match resolve_client_ip( + state.config.trust_proxy_headers, + headers.get("x-forwarded-for"), + peer, + ) { + Some(ip) => state.public_rate_limiter.check(ip), + // Preserve fail-open when neither trusted headers nor ConnectInfo yield an IP. + None => true, }; if !admitted { return Err(AppError::RateLimited); diff --git a/core/src/public/proxy_tests.rs b/core/src/public/proxy_tests.rs new file mode 100644 index 0000000..fc38597 --- /dev/null +++ b/core/src/public/proxy_tests.rs @@ -0,0 +1,102 @@ +use super::*; +use crate::config::Config; +use axum::http::Request; +use rstest::rstest; +use std::net::IpAddr; +use std::sync::Arc; +use tower::ServiceExt; + +#[rstest] +#[case(true, Some("198.51.100.1, 203.0.113.1"), Some("198.51.100.1"))] +#[case(true, Some("garbage, 2001:db8::1, 203.0.113.1"), Some("2001:db8::1"))] +#[case(true, Some("garbage, unknown"), Some("192.0.2.1"))] +#[case(true, Some(""), Some("192.0.2.1"))] +#[case(true, None, Some("192.0.2.1"))] +#[case(false, Some("198.51.100.1"), Some("192.0.2.1"))] +fn client_ip_resolves_when_proxy_trust_is_configured( + #[case] trusted: bool, + #[case] forwarded: Option<&str>, + #[case] expected: Option<&str>, +) { + // Given a peer and an optional forwarded chain. + let peer = Some("192.0.2.1".parse::().unwrap()); + let header = forwarded.map(|value| HeaderValue::from_str(value).unwrap()); + // When resolving the client at the trust boundary. + let resolved = resolve_client_ip(trusted, header.as_ref(), peer); + // Then the first valid forwarded IP is used only when trusted. + assert_eq!(resolved, expected.map(|value| value.parse().unwrap())); +} + +#[rstest] +#[case(false)] +#[case(true)] +fn client_ip_is_absent_when_no_sources_exist(#[case] trusted: bool) { + // Given no header or peer; when resolving; then preserve fail-open. + assert_eq!(resolve_client_ip(trusted, None, None), None); +} + +#[test] +fn client_ip_falls_back_when_header_is_not_text() { + // Given an opaque header and a valid peer. + let header = HeaderValue::from_bytes(b"\xff").unwrap(); + let peer = Some("192.0.2.1".parse().unwrap()); + // When resolving; then use the peer rather than the invalid header. + assert_eq!(resolve_client_ip(true, Some(&header), peer), peer); +} + +#[test] +fn proxy_headers_are_untrusted_when_config_is_default() { + // Given default configuration; when inspecting trust; then it is opt-in. + assert!(!Config::default().trust_proxy_headers); +} + +#[rstest] +#[case(true)] +#[case(false)] +#[tokio::test] +async fn public_router_separates_forwarded_budgets_only_when_trusted(#[case] trusted: bool) { + // Given two forwarded clients behind the same peer with a two-request budget. + let config = Config { + trust_proxy_headers: trusted, + ..Config::default() + }; + let pool = sqlx::PgPool::connect_lazy("postgres://127.0.0.1:9/never").unwrap(); + pool.close().await; + let state = AppState { + pool, + s3: crate::storage::Storage::build(&config).await.unwrap(), + jwks_cache: crate::auth::jwks::JwksCache::new(Vec::new()), + rate_limiters: Arc::new(crate::state::RateLimiters::default()), + billing_cache: Arc::new(crate::billing::cache::BillingCache::new(60)), + public_rate_limiter: Arc::new(crate::auth::rate_limit::PublicRateLimiter::new(2)), + config, + }; + let app = public_router().with_state(state); + let peer = SocketAddr::from(([192, 0, 2, 1], 1234)); + // When both clients exhaust their budgets through that peer. + for (forwarded, limited) in [ + ("198.51.100.1", false), + ("198.51.100.1", false), + ("198.51.100.1", true), + ("198.51.100.2", !trusted), + ("198.51.100.2", !trusted), + ("198.51.100.2", true), + ] { + let request = Request::builder() + .uri("/public/test-key") + .header("x-forwarded-for", forwarded) + .extension(ConnectInfo(peer)) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(request).await.unwrap(); + // Then trusted clients are independent; spoofing cannot bypass peer limiting. + assert_eq!( + response.status(), + if limited { + StatusCode::TOO_MANY_REQUESTS + } else { + StatusCode::INTERNAL_SERVER_ERROR + } + ); + } +} diff --git a/deploy/self-hosting/compose.yml b/deploy/self-hosting/compose.yml index e96259c..986d3a8 100644 --- a/deploy/self-hosting/compose.yml +++ b/deploy/self-hosting/compose.yml @@ -33,6 +33,7 @@ services: BOOSKIFF_JWT_AUDIENCE: ${BOOSKIFF_JWT_AUDIENCE:-} BOOSKIFF_JWT_OWNER_TYPE_CLAIM: ${BOOSKIFF_JWT_OWNER_TYPE_CLAIM:-owner_type} BOOSKIFF_PREMIUM_MODE: ${BOOSKIFF_PREMIUM_MODE:-everyone} + BOOSKIFF_BILLING_CACHE_TTL_SECS: ${BOOSKIFF_BILLING_CACHE_TTL_SECS:-60} BOOSKIFF_PAYMENT_PROVIDER: ${BOOSKIFF_PAYMENT_PROVIDER:-disabled} BOOSKIFF_PLAN_DEFAULT_STORAGE_QUOTA_BYTES: ${BOOSKIFF_PLAN_DEFAULT_STORAGE_QUOTA_BYTES:-1073741824} BOOSKIFF_PLAN_DEFAULT_MAX_FILE_BYTES: ${BOOSKIFF_PLAN_DEFAULT_MAX_FILE_BYTES:-104857600} @@ -42,6 +43,9 @@ services: BOOSKIFF_PLAN_PREMIUM_RATE_LIMIT_RPM: ${BOOSKIFF_PLAN_PREMIUM_RATE_LIMIT_RPM:-300} BOOSKIFF_PRESIGNED_GET_TTL_SECS: ${BOOSKIFF_PRESIGNED_GET_TTL_SECS:-900} BOOSKIFF_PUBLIC_BASE_URL: ${BOOSKIFF_PUBLIC_BASE_URL:-http://127.0.0.1:18080} + BOOSKIFF_PUBLIC_RATE_LIMIT_RPM: ${BOOSKIFF_PUBLIC_RATE_LIMIT_RPM:-300} + # Only Caddy is published; disable this if core is exposed directly. + BOOSKIFF_TRUST_PROXY_HEADERS: ${BOOSKIFF_TRUST_PROXY_HEADERS:-true} BOOSKIFF_ADMIN_BOOTSTRAP_TOKEN: ${BOOSKIFF_ADMIN_BOOTSTRAP_TOKEN:-} RUST_LOG: ${RUST_LOG:-info} depends_on: diff --git a/docs/operations.md b/docs/operations.md index 0d4d469..21b083c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -6,7 +6,10 @@ drive-foundation レビューで確定した運用上の仕様ポイントをま - 認証なしで誰でもダウンロードできる。帯域計量 (レスポンス転送量の計測・課金) は初期スコープ外。 - DoS 緩和として per-IP レート制限を付与している。既定 300 req/min で、環境変数 `BOOSKIFF_PUBLIC_RATE_LIMIT_RPM` で変更できる。 -- 制限キーはクライアント IP (`ConnectInfo`)。非 TCP リスナ等で ConnectInfo が取得できない環境では制限をスキップする (fail-open)。 +- 既定の制限キーは接続元の peer IP (`ConnectInfo`)。`BOOSKIFF_TRUST_PROXY_HEADERS` は既定 `false` で、X-Forwarded-For は無視する。 +- リバースプロキシ配下では peer IP がプロキシの IP になるため、`BOOSKIFF_TRUST_PROXY_HEADERS=true` で X-Forwarded-For のカンマ区切りリスト中、最左の有効な IP を制限キーに使う。ヘッダーがない場合や有効な IP を解析できない場合は peer IP に戻る。どちらからも IP を取得できない環境では従来どおり制限をスキップする (fail-open)。 +- この設定は X-Forwarded-For を適切に設定・浄化する信頼できるプロキシ (同梱 Caddy 等) の背後でのみ有効化すること。core を直接公開した状態で有効化すると、ヘッダーの偽装により制限を回避できる。core への直接アクセスを許可しないこと。 +- セルフホスト用 Compose は Caddy だけがホストにポートを公開し、core は内部ネットワークのみのため、この Compose に限り `BOOSKIFF_TRUST_PROXY_HEADERS` の既定値を `true` にしている。core を直接公開する構成へ変更する場合は `false` にする。 ## 公開ファイルのキャッシュと失効 @@ -30,4 +33,5 @@ drive-foundation レビューで確定した運用上の仕様ポイントをま | 変数 | 既定値 | 意味 | | --- | --- | --- | | `BOOSKIFF_PUBLIC_RATE_LIMIT_RPM` | 300 | `/public/{key}` の per-IP レート制限 (req/min) | +| `BOOSKIFF_TRUST_PROXY_HEADERS` | false (セルフホスト Compose は true) | 信頼プロキシの X-Forwarded-For を制限キーに利用。直接公開時は有効化禁止 | | `BOOSKIFF_BILLING_CACHE_TTL_SECS` | 60 | 課金解決キャッシュの TTL (秒)。0 で無効 | From 362a09db808e064bcbe42f13c432da09ad4847a9 Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:22:17 +0900 Subject: [PATCH 12/13] =?UTF-8?q?test(public):=20=E6=97=A2=E5=AE=9A?= =?UTF-8?q?=E5=80=A4=20300=20rpm=20=E3=81=A7=20per-IP=20=E5=88=B6=E9=99=90?= =?UTF-8?q?=E3=81=8C=E5=8A=B9=E3=81=8F=E3=81=93=E3=81=A8=E3=82=92=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/src/auth/rate_limit.rs | 12 ++++++++++++ core/src/config.rs | 6 ++++++ 2 files changed, 18 insertions(+) diff --git a/core/src/auth/rate_limit.rs b/core/src/auth/rate_limit.rs index 2133b3a..bae4baf 100644 --- a/core/src/auth/rate_limit.rs +++ b/core/src/auth/rate_limit.rs @@ -60,6 +60,18 @@ impl OwnerRateLimiter { mod tests { use super::*; + #[test] + fn public_limiter_allows_300_then_denies_when_default_budget_is_exhausted() { + // Given the documented 300 rpm budget for one IP. + let limiter = PublicRateLimiter::new(300); + let ip = std::net::Ipv4Addr::LOCALHOST.into(); + // When 301 requests arrive in a single burst. + let admitted: [bool; 301] = std::array::from_fn(|_| limiter.check(ip)); + // Then the first 300 succeed and the 301st is denied. + assert!(admitted[..300].iter().all(|allowed| *allowed)); + assert!(!admitted[300]); + } + #[test] fn public_limiter_allows_burst_then_denies_when_budget_exhausted() { // Given a two-request budget for one IP. diff --git a/core/src/config.rs b/core/src/config.rs index a386006..2356c0f 100644 --- a/core/src/config.rs +++ b/core/src/config.rs @@ -293,6 +293,12 @@ mod tests { use super::*; use rstest::rstest; + #[test] + fn public_rate_limit_defaults_to_300_when_config_is_default() { + // Given no overrides; when constructing defaults; then pin the documented quota. + assert_eq!(Config::default().public_rate_limit_rpm, 300); + } + #[rstest] #[case( "https://id.example.com", From 1372b30e4260d294479f58479c9cfe83b1bfaa99 Mon Sep 17 00:00:00 2001 From: turtton Date: Sun, 13 Sep 2026 00:38:49 +0900 Subject: [PATCH 13/13] =?UTF-8?q?fix(ci):=20MinIO=20=E3=82=A4=E3=83=A1?= =?UTF-8?q?=E3=83=BC=E3=82=B8=E3=82=92=20quay.io=20=E3=81=AB=E7=A7=BB?= =?UTF-8?q?=E8=A1=8C=E3=81=97=E3=81=A6=20e2e=20=E3=81=AE=20pull=20?= =?UTF-8?q?=E5=A4=B1=E6=95=97=E3=82=92=E8=A7=A3=E6=B6=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- compose.yml | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/compose.yml b/compose.yml index 5369d7b..76af589 100644 --- a/compose.yml +++ b/compose.yml @@ -17,7 +17,9 @@ services: retries: 30 minio: - image: minio/minio:latest + # quay.io: MinIO no longer publishes to Docker Hub (pull access denied), + # which broke the CI e2e job. + image: quay.io/minio/minio:latest command: server /data --console-address ":9001" environment: MINIO_ROOT_USER: booskiff @@ -36,7 +38,7 @@ services: # Create the dev bucket once MinIO is healthy; always exits 0 so the # compose project reports healthy even when the bucket already exists. mc-init: - image: minio/mc + image: quay.io/minio/mc:latest depends_on: minio: condition: service_healthy