diff --git a/.env.example b/.env.example index b2156c1..afe4348 100644 --- a/.env.example +++ b/.env.example @@ -14,6 +14,8 @@ CRANK_UI_IMAGE=crank/ui:dev CRANK_STORAGE_ROOT=/var/lib/crank/storage CRANK_PUBLISH_BIND=127.0.0.1 CRANK_ADMIN_BIND=0.0.0.0:3001 +CRANK_ADMIN_RATE_LIMIT_RPS=30 +CRANK_ADMIN_RATE_LIMIT_BURST=60 CRANK_MCP_BIND=0.0.0.0:3002 CRANK_MCP_REFRESH_MS=5000 CRANK_MCP_RATE_LIMIT_RPS=60 diff --git a/apps/admin-api/src/app.rs b/apps/admin-api/src/app.rs index fd82646..c668b63 100644 --- a/apps/admin-api/src/app.rs +++ b/apps/admin-api/src/app.rs @@ -5,6 +5,7 @@ use axum::{ use crate::{ auth::{require_session, require_workspace_session}, + rate_limit::apply_api_rate_limit, request_context::apply_request_context, routes::{ access::{ @@ -206,6 +207,10 @@ pub fn build_app(state: AppState) -> Router { .merge(protected_auth_router), ) .nest("/api/admin", admin_router) + .layer(middleware::from_fn_with_state( + state.clone(), + apply_api_rate_limit, + )) .layer(middleware::from_fn(apply_request_context)) .with_state(state) } @@ -2481,6 +2486,9 @@ mod tests { test_auth_settings(), test_secret_crypto(), ), + api_rate_limiter: crank_runtime::RequestRateLimiter::new( + crank_runtime::RequestRateLimitConfig::new(10_000, 10_000).unwrap(), + ), }) } @@ -2570,6 +2578,78 @@ mod tests { client } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn rejects_rapid_login_requests_with_429() { + let registry = test_registry().await; + let storage_root = test_storage_root("login_rate_limit"); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let app = build_app(AppState { + service: AdminService::new( + registry, + storage_root, + test_auth_settings(), + test_secret_crypto(), + ), + api_rate_limiter: crank_runtime::RequestRateLimiter::new( + crank_runtime::RequestRateLimitConfig::new(1, 1).unwrap(), + ), + }); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); + let handle = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); + }); + + let client = reqwest::Client::new(); + let root_url = format!("http://{address}"); + let first_response = client + .post(format!("{root_url}/api/auth/login")) + .header("x-request-id", "req_admin_rate_01") + .json(&json!({ + "email": TEST_AUTH_EMAIL, + "password": TEST_AUTH_PASSWORD, + })) + .send() + .await + .unwrap(); + assert_eq!(first_response.status(), reqwest::StatusCode::OK); + + let second_response = client + .post(format!("{root_url}/api/auth/login")) + .header("x-request-id", "req_admin_rate_02") + .json(&json!({ + "email": TEST_AUTH_EMAIL, + "password": TEST_AUTH_PASSWORD, + })) + .send() + .await + .unwrap(); + + assert_eq!( + second_response.status(), + reqwest::StatusCode::TOO_MANY_REQUESTS + ); + assert_eq!( + second_response.headers()["x-request-id"].to_str().unwrap(), + "req_admin_rate_02" + ); + let payload = second_response.json::().await.unwrap(); + assert_eq!(payload["error"]["code"], "rate_limited"); + let retry_after_ms = payload["error"]["context"]["retry_after_ms"] + .as_u64() + .unwrap(); + assert!((1..=1000).contains(&retry_after_ms)); + + let _ = shutdown_tx.send(()); + let _ = handle.await; + } + async fn assert_success_json(response: reqwest::Response) -> Value { let status = response.status(); let body = response.text().await.unwrap(); diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index 3c19c14..0551b24 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -41,6 +41,11 @@ pub enum ApiError { context: Option, }, #[error("{message}")] + RateLimited { + message: String, + context: Option, + }, + #[error("{message}")] Internal { message: String, context: Option, @@ -76,6 +81,13 @@ impl ApiError { } } + pub(crate) fn rate_limited_with_context(message: impl Into, context: Value) -> Self { + Self::RateLimited { + message: message.into(), + context: Some(context), + } + } + pub(crate) fn validation_with_context(message: impl Into, context: Value) -> Self { Self::Validation { message: message.into(), @@ -104,6 +116,7 @@ impl ApiError { Self::Validation { .. } => StatusCode::BAD_REQUEST, Self::NotFound { .. } => StatusCode::NOT_FOUND, Self::Conflict { .. } => StatusCode::CONFLICT, + Self::RateLimited { .. } => StatusCode::TOO_MANY_REQUESTS, Self::Internal { .. } => StatusCode::INTERNAL_SERVER_ERROR, } } @@ -115,6 +128,7 @@ impl ApiError { Self::Validation { .. } => "validation_error", Self::NotFound { .. } => "not_found", Self::Conflict { .. } => "conflict", + Self::RateLimited { .. } => "rate_limited", Self::Internal { .. } => "internal_error", } } @@ -130,7 +144,8 @@ impl IntoResponse for ApiError { | Self::Forbidden { message, .. } | Self::Validation { message, .. } | Self::NotFound { message, .. } - | Self::Conflict { message, .. } => { + | Self::Conflict { message, .. } + | Self::RateLimited { message, .. } => { warn!(error_code = self.code(), error_message = %message) } } @@ -159,6 +174,7 @@ impl ApiError { | Self::Validation { context, .. } | Self::NotFound { context, .. } | Self::Conflict { context, .. } + | Self::RateLimited { context, .. } | Self::Internal { context, .. } => context.clone(), } } diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index ff1b17b..d4ea324 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -1,6 +1,7 @@ mod app; mod auth; mod error; +mod rate_limit; mod request_context; mod routes; mod service; @@ -20,7 +21,9 @@ use crate::{ service::AdminService, state::AppState, }; -use crank_runtime::{RuntimeExecutor, RuntimeLimits, SecretCrypto}; +use crank_runtime::{ + RequestRateLimitConfig, RequestRateLimiter, RuntimeExecutor, RuntimeLimits, SecretCrypto, +}; #[tokio::main] async fn main() -> Result<(), Box> { @@ -58,6 +61,7 @@ async fn main() -> Result<(), Box> { }, }; let runtime_limits = RuntimeLimits::from_env()?; + let api_rate_limit = admin_api_rate_limit_config_from_env()?; let secret_crypto = SecretCrypto::new(&env::var("CRANK_MASTER_KEY")?)?; let runtime = RuntimeExecutor::with_limits(runtime_limits); let service = AdminService::new_with_runtime( @@ -71,7 +75,10 @@ async fn main() -> Result<(), Box> { if env_flag("CRANK_DEMO_SEED") { service.seed_demo_assets().await?; } - let state = AppState { service }; + let state = AppState { + service, + api_rate_limiter: RequestRateLimiter::new(api_rate_limit), + }; let app = build_app(state); let listener = TcpListener::bind(socket_addr).await?; @@ -80,6 +87,8 @@ async fn main() -> Result<(), Box> { runtime_max_concurrent_window = runtime_limits.max_concurrent_window, runtime_max_concurrent_sessions = runtime_limits.max_concurrent_sessions, runtime_max_concurrent_jobs = runtime_limits.max_concurrent_jobs, + admin_rate_limit_rps = api_rate_limit.requests_per_second, + admin_rate_limit_burst = api_rate_limit.burst, max_connections = pool_config.max_connections, min_connections = pool_config.min_connections, acquire_timeout_ms = pool_config.acquire_timeout_ms, @@ -105,6 +114,20 @@ fn env_flag(name: &str) -> bool { ) } +fn admin_api_rate_limit_config_from_env() +-> Result> { + let requests_per_second = env::var("CRANK_ADMIN_RATE_LIMIT_RPS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(30); + let burst = env::var("CRANK_ADMIN_RATE_LIMIT_BURST") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(60); + + Ok(RequestRateLimitConfig::new(requests_per_second, burst)?) +} + fn database_options_from_env() -> Result> { if let Ok(database_url) = env::var("CRANK_DATABASE_URL") { return Ok(database_url.parse::()?); diff --git a/apps/admin-api/src/rate_limit.rs b/apps/admin-api/src/rate_limit.rs new file mode 100644 index 0000000..33a4180 --- /dev/null +++ b/apps/admin-api/src/rate_limit.rs @@ -0,0 +1,106 @@ +use axum::{ + extract::{Request, State}, + http::header::{COOKIE, HeaderMap}, + middleware::Next, + response::Response, +}; +use crank_runtime::RateLimitRejection; + +use crate::{auth::SESSION_COOKIE_NAME, error::ApiError, state::AppState}; + +pub async fn apply_api_rate_limit( + State(state): State, + request: Request, + next: Next, +) -> Result { + let key = rate_limit_key(request.headers(), request.uri().path()); + if let Err(rejection) = state.api_rate_limiter.check(&key) { + return Err(ApiError::rate_limited_with_context( + "request rate limit exceeded", + rejection_context(rejection), + )); + } + + Ok(next.run(request).await) +} + +fn rejection_context(rejection: RateLimitRejection) -> serde_json::Value { + serde_json::json!({ + "retry_after_ms": rejection.retry_after_ms, + }) +} + +fn rate_limit_key(headers: &HeaderMap, path: &str) -> String { + if let Some(session_id) = session_id_from_headers(headers) { + return format!("session:{session_id}"); + } + + if let Some(forwarded_for) = header_value(headers, "x-forwarded-for") { + let ip = forwarded_for + .split(',') + .next() + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or("unknown"); + return format!("ip:{ip}"); + } + + if let Some(real_ip) = header_value(headers, "x-real-ip") { + return format!("ip:{real_ip}"); + } + + format!("anonymous:{path}") +} + +fn session_id_from_headers(headers: &HeaderMap) -> Option { + let cookies = headers.get(COOKIE)?.to_str().ok()?; + for part in cookies.split(';') { + let (name, value) = part.trim().split_once('=')?; + if name != SESSION_COOKIE_NAME { + continue; + } + let (session_id, _) = value.split_once('.')?; + if !session_id.is_empty() { + return Some(session_id.to_owned()); + } + } + + None +} + +fn header_value<'a>(headers: &'a HeaderMap, name: &'static str) -> Option<&'a str> { + headers.get(name)?.to_str().ok().map(str::trim) +} + +#[cfg(test)] +mod tests { + use axum::http::{HeaderMap, HeaderValue, header::COOKIE}; + + use super::rate_limit_key; + + #[test] + fn keys_by_session_cookie_first() { + let mut headers = HeaderMap::new(); + headers.insert( + COOKIE, + HeaderValue::from_static("theme=dark; crank_session=sess_123.secret_456"), + ); + headers.insert("x-forwarded-for", HeaderValue::from_static("10.0.0.5")); + + assert_eq!( + rate_limit_key(&headers, "/api/auth/login"), + "session:sess_123" + ); + } + + #[test] + fn falls_back_to_forwarded_ip() { + let mut headers = HeaderMap::new(); + headers.insert( + "x-forwarded-for", + HeaderValue::from_static("10.0.0.5, 10.0.0.6"), + ); + + assert_eq!(rate_limit_key(&headers, "/api/auth/login"), "ip:10.0.0.5"); + } +} diff --git a/apps/admin-api/src/state.rs b/apps/admin-api/src/state.rs index 6c618ca..99b5165 100644 --- a/apps/admin-api/src/state.rs +++ b/apps/admin-api/src/state.rs @@ -1,6 +1,8 @@ use crate::service::AdminService; +use crank_runtime::RequestRateLimiter; #[derive(Clone)] pub struct AppState { pub service: AdminService, + pub api_rate_limiter: RequestRateLimiter, } diff --git a/docs/runtime-config.md b/docs/runtime-config.md index 866a22b..7b3068c 100644 --- a/docs/runtime-config.md +++ b/docs/runtime-config.md @@ -71,6 +71,8 @@ var/crank/ - `CRANK_STORAGE_ROOT` - `CRANK_PUBLISH_BIND` - `CRANK_ADMIN_BIND` +- `CRANK_ADMIN_RATE_LIMIT_RPS` +- `CRANK_ADMIN_RATE_LIMIT_BURST` - `CRANK_MCP_BIND` - `CRANK_MCP_REFRESH_MS` - `CRANK_MCP_RATE_LIMIT_RPS` @@ -90,6 +92,8 @@ var/crank/ Стартовое значение для refresh published tools: +- `CRANK_ADMIN_RATE_LIMIT_RPS=30` +- `CRANK_ADMIN_RATE_LIMIT_BURST=60` - `CRANK_MCP_REFRESH_MS=5000` - `CRANK_MCP_RATE_LIMIT_RPS=60` - `CRANK_MCP_RATE_LIMIT_BURST=120` @@ -207,6 +211,11 @@ runtime env-переменной. Это значит, что: - `CRANK_MCP_RATE_LIMIT_RPS=60` - `CRANK_MCP_RATE_LIMIT_BURST=120` +Для admin-api ingress throttling используются явные defaults: + +- `CRANK_ADMIN_RATE_LIMIT_RPS=30` +- `CRANK_ADMIN_RATE_LIMIT_BURST=60` + `CRANK_DATABASE_URL` допускается только как backward-compatible fallback для локальных тестов и переходного периода, но не как основная deployment-модель.