diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index 2cfba2c..172acb4 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -16,6 +16,7 @@ use tracing::{debug, error, info, warn}; use tranquil_pds::api::error::{ApiError, DbResultExt}; use tranquil_pds::auth::{Auth, AuthAny, NotTakendown, Permissive, VerifyScope}; use tranquil_pds::delegation::DelegationActionType; +use tranquil_pds::rate_limit::{BlobUploadLimit, RateLimited}; use tranquil_pds::state::AppState; use tranquil_pds::types::{CidLink, Did, Nsid}; use tranquil_pds::util::get_header_str; @@ -46,6 +47,7 @@ fn detect_mime_type(data: &[u8], client_hint: &str) -> String { pub async fn upload_blob( State(state): State, + _rate_limit: RateLimited, headers: axum::http::HeaderMap, auth: AuthAny, body: Body, diff --git a/crates/tranquil-api/src/repo/record/batch.rs b/crates/tranquil-api/src/repo/record/batch.rs index e823a41..569fd3d 100644 --- a/crates/tranquil-api/src/repo/record/batch.rs +++ b/crates/tranquil-api/src/repo/record/batch.rs @@ -12,6 +12,7 @@ use tranquil_pds::auth::{ Active, Auth, WriteOpKind, require_not_migrated, require_verified_or_delegated, verify_batch_write_scopes, }; +use tranquil_pds::rate_limit::check_repo_write_rate_limits; use tranquil_pds::repo::TrackingBlockStore; use tranquil_pds::repo_ops::{ CommitResult, FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids, @@ -374,6 +375,14 @@ pub async fn apply_writes( } let did = principal_did.into_did(); + let points = input.writes.iter().fold(0u32, |total, write| { + total.saturating_add(match write { + WriteOp::Create { .. } => 3, + WriteOp::Update { .. } => 2, + WriteOp::Delete { .. } => 1, + }) + }); + check_repo_write_rate_limits(&state, did.as_str(), points).await?; require_not_migrated(&state, &did).await?; require_verified_or_delegated(&state, batch_proof.user()).await?; diff --git a/crates/tranquil-api/src/repo/record/delete.rs b/crates/tranquil-api/src/repo/record/delete.rs index 3134bff..2a5991d 100644 --- a/crates/tranquil-api/src/repo/record/delete.rs +++ b/crates/tranquil-api/src/repo/record/delete.rs @@ -7,6 +7,7 @@ use std::str::FromStr; use tranquil_pds::api::error::ApiError; use tranquil_pds::auth::{Active, Auth, VerifyScope}; use tranquil_pds::cid_types::RecordCid; +use tranquil_pds::rate_limit::check_repo_write_rate_limits; use tranquil_pds::repo_ops::{ FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write, with_repair_retry, }; @@ -40,6 +41,7 @@ pub async fn delete_record( let scope_proof = auth.verify_repo_delete(&input.collection)?; let repo_auth = prepare_repo_write(&state, &scope_proof, &input.repo).await?; let did = repo_auth.did; + check_repo_write_rate_limits(&state, did.as_str(), 1).await?; let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; diff --git a/crates/tranquil-api/src/repo/record/write.rs b/crates/tranquil-api/src/repo/record/write.rs index e5966f7..ebc3625 100644 --- a/crates/tranquil-api/src/repo/record/write.rs +++ b/crates/tranquil-api/src/repo/record/write.rs @@ -13,6 +13,7 @@ use tranquil_pds::auth::{ Active, Auth, AuthSource, RepoScopeAction, ScopeVerified, VerifyScope, require_not_migrated, require_verified_or_delegated, }; +use tranquil_pds::rate_limit::check_repo_write_rate_limits; use tranquil_pds::repo_ops::{ FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids, finalize_repo_write, with_repair_retry, @@ -127,6 +128,7 @@ pub async fn create_record( let scope_proof = auth.verify_repo_create(&input.collection)?; let repo_auth = prepare_repo_write(&state, &scope_proof, &input.repo).await?; let did = repo_auth.did; + check_repo_write_rate_limits(&state, did.as_str(), 3).await?; let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; @@ -328,6 +330,7 @@ pub async fn put_record( let upsert_proof = auth.verify_repo_upsert(&input.collection)?; let repo_auth = prepare_repo_write(&state, &upsert_proof, &input.repo).await?; let did = repo_auth.did; + check_repo_write_rate_limits(&state, did.as_str(), 2).await?; let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; diff --git a/crates/tranquil-api/src/server/account_status.rs b/crates/tranquil-api/src/server/account_status.rs index de83896..72a51c9 100644 --- a/crates/tranquil-api/src/server/account_status.rs +++ b/crates/tranquil-api/src/server/account_status.rs @@ -17,6 +17,9 @@ use tranquil_pds::auth::{Auth, NotTakendown, Permissive, require_legacy_session_ use tranquil_pds::cache::Cache; use tranquil_pds::oauth::scopes::{AccountAction, AccountAttr}; use tranquil_pds::plc::PlcClient; +use tranquil_pds::rate_limit::{ + AccountRequestDailyLimit, DeleteAccountLimit, RateLimited, check_user_rate_limit, +}; use tranquil_pds::state::AppState; use tranquil_pds::types::{PlainPassword, Tid}; use uuid::Uuid; @@ -421,12 +424,8 @@ pub async fn activate_account( "[MIGRATION] activateAccount: Sequencing account event (active=true) for did={}", did ); - if let Err(e) = tranquil_pds::repo_ops::sequence_account_event( - &state, - &did, - final_status, - ) - .await + if let Err(e) = + tranquil_pds::repo_ops::sequence_account_event(&state, &did, final_status).await { warn!( "[MIGRATION] activateAccount: Failed to sequence account activation event: {}", @@ -574,12 +573,8 @@ pub async fn deactivate_account( { warn!("failed to sync deactivation to repo backend: {e:?}"); } - if let Err(e) = tranquil_pds::repo_ops::sequence_account_event( - &state, - &did, - final_status, - ) - .await + if let Err(e) = + tranquil_pds::repo_ops::sequence_account_event(&state, &did, final_status).await { warn!("Failed to sequence account deactivated event: {}", e); } @@ -597,6 +592,8 @@ pub async fn request_account_delete( State(state): State, auth: Auth, ) -> Result, ApiError> { + let _rate_limit = + check_user_rate_limit::(&state, auth.did.as_str()).await?; let session_mfa = require_legacy_session_mfa(&state, &auth).await?; let user_id = state @@ -640,6 +637,7 @@ pub struct DeleteAccountInput { pub async fn delete_account( State(state): State, + _rate_limit: RateLimited, Json(input): Json, ) -> Result, ApiError> { let did = &input.did; diff --git a/crates/tranquil-api/src/server/email.rs b/crates/tranquil-api/src/server/email.rs index 401ae06..ed2a8cd 100644 --- a/crates/tranquil-api/src/server/email.rs +++ b/crates/tranquil-api/src/server/email.rs @@ -17,7 +17,9 @@ use tranquil_pds::api::{ }; use tranquil_pds::auth::{Auth, NotTakendown}; use tranquil_pds::oauth::scopes::{AccountAction, AccountAttr}; -use tranquil_pds::rate_limit::{EmailUpdateLimit, RateLimited, VerificationCheckLimit}; +use tranquil_pds::rate_limit::{ + EmailUpdateLimit, RateLimited, VerificationCheckLimit, check_user_rate_limit, +}; use tranquil_pds::state::AppState; use tranquil_pds::types::{AtIdentifier, Did}; @@ -55,11 +57,11 @@ pub struct RequestEmailUpdateInput { pub async fn request_email_update( State(state): State, - _rate_limit: RateLimited, auth: Auth, input: Option>, ) -> Result, ApiError> { auth.check_account_scope(AccountAttr::Email, AccountAction::Manage)?; + let _rate_limit = check_user_rate_limit::(&state, auth.did.as_str()).await?; let user = state .repos diff --git a/crates/tranquil-api/src/server/passkey_account.rs b/crates/tranquil-api/src/server/passkey_account.rs index df217e1..43982de 100644 --- a/crates/tranquil-api/src/server/passkey_account.rs +++ b/crates/tranquil-api/src/server/passkey_account.rs @@ -12,7 +12,9 @@ use tranquil_pds::api::{OptionsResponse, SuccessResponse}; use tranquil_pds::auth::NormalizedLoginIdentifier; use tranquil_pds::auth::{ServiceTokenVerifier, generate_app_password, is_service_token}; -use tranquil_pds::rate_limit::{AccountCreationLimit, PasswordResetLimit, RateLimited}; +use tranquil_pds::rate_limit::{ + AccountCreationLimit, PasswordResetDailyLimit, PasswordResetLimit, RateLimited, +}; use tranquil_pds::state::AppState; use tranquil_pds::types::{Did, Handle, Jti, Nsid, PlainPassword}; use tranquil_pds::validation::validate_password; @@ -663,6 +665,7 @@ pub struct RequestPasskeyRecoveryInput { pub async fn request_passkey_recovery( State(state): State, _rate_limit: RateLimited, + _daily_rate_limit: RateLimited, Json(input): Json, ) -> Result, ApiError> { let hostname_for_handles = tranquil_config::get().server.hostname_without_port(); diff --git a/crates/tranquil-api/src/server/password.rs b/crates/tranquil-api/src/server/password.rs index db8f327..3190856 100644 --- a/crates/tranquil-api/src/server/password.rs +++ b/crates/tranquil-api/src/server/password.rs @@ -8,7 +8,9 @@ use tranquil_pds::auth::{ Active, Auth, NormalizedLoginIdentifier, require_legacy_session_mfa, require_reauth_window, require_reauth_window_if_available, }; -use tranquil_pds::rate_limit::{PasswordResetLimit, RateLimited, ResetPasswordLimit}; +use tranquil_pds::rate_limit::{ + PasswordResetDailyLimit, PasswordResetLimit, RateLimited, ResetPasswordLimit, +}; use tranquil_pds::state::AppState; use tranquil_pds::types::{Handle, PlainPassword}; use tranquil_pds::validation::validate_password; @@ -22,6 +24,7 @@ pub struct RequestPasswordResetInput { pub async fn request_password_reset( State(state): State, _rate_limit: RateLimited, + _daily_rate_limit: RateLimited, Json(input): Json, ) -> Result, ApiError> { let identifier = input.email.trim(); diff --git a/crates/tranquil-api/src/server/session.rs b/crates/tranquil-api/src/server/session.rs index 413359a..b491e1c 100644 --- a/crates/tranquil-api/src/server/session.rs +++ b/crates/tranquil-api/src/server/session.rs @@ -16,7 +16,7 @@ use tranquil_pds::auth::{ require_reauth_window, }; use tranquil_pds::rate_limit::{ - LoginLimit, RateLimited, RefreshSessionLimit, TotpVerifyLimit, + ClientIp, RateLimited, RefreshSessionLimit, TotpVerifyLimit, check_login_rate_limits, check_user_rate_limit_with_message, }; use tranquil_pds::state::AppState; @@ -84,10 +84,9 @@ pub struct CreateSessionOutput { pub async fn create_session( State(state): State, - rate_limit: RateLimited, + client_ip: ClientIp, Json(input): Json, ) -> Result { - let client_ip = rate_limit.client_ip(); info!( "create_session called with identifier: {}", input.identifier @@ -95,6 +94,7 @@ pub async fn create_session( let hostname_for_handles = tranquil_config::get().server.hostname_without_port(); let normalized_identifier = NormalizedLoginIdentifier::normalize(&input.identifier, hostname_for_handles); + check_login_rate_limits(&state, normalized_identifier.as_str(), client_ip.as_str()).await?; info!( "Normalized identifier: {} -> {}", input.identifier, normalized_identifier @@ -349,7 +349,7 @@ pub async fn create_session( state.repos.infra.as_ref(), row.id, hostname, - client_ip, + client_ip.as_str(), row.preferred_comms_channel, ) .await diff --git a/crates/tranquil-cache/src/lib.rs b/crates/tranquil-cache/src/lib.rs index 2551cf8..e3001bc 100644 --- a/crates/tranquil-cache/src/lib.rs +++ b/crates/tranquil-cache/src/lib.rs @@ -1,4 +1,4 @@ -pub use tranquil_infra::{Cache, CacheError, DistributedRateLimiter}; +pub use tranquil_infra::{Cache, CacheError, DistributedRateLimiter, RateLimitStatus}; use async_trait::async_trait; use std::sync::Arc; @@ -114,6 +114,55 @@ return c", } } + async fn consume_rate_limit( + &self, + key: &str, + limit: u32, + window_ms: u64, + points: u32, + ) -> RateLimitStatus { + let mut conn = self.conn.clone(); + let full_key = format!("rl:{}", key); + let window_ms = window_ms.max(1); + let result: Result<(i64, i64, i64), _> = redis::Script::new( + r"local c = tonumber(redis.call('GET', KEYS[1]) or '0') +local cost = tonumber(ARGV[1]) +local limit = tonumber(ARGV[2]) +if c + cost > limit then + local ttl = redis.call('PTTL', KEYS[1]) + return {0, c, ttl} +end +local next = redis.call('INCRBY', KEYS[1], cost) +local ttl = redis.call('PTTL', KEYS[1]) +if ttl < 0 then + redis.call('PEXPIRE', KEYS[1], ARGV[3]) + ttl = tonumber(ARGV[3]) +end +return {1, next, ttl}", + ) + .key(&full_key) + .arg(points) + .arg(limit) + .arg(window_ms) + .invoke_async(&mut conn) + .await; + match result { + Ok((allowed, count, ttl)) => { + let count = u64::try_from(count).unwrap_or(u64::MAX); + let retry_after_ms = u64::try_from(ttl).unwrap_or(window_ms); + if allowed == 1 { + RateLimitStatus::allowed(limit, count, retry_after_ms) + } else { + RateLimitStatus::rejected(limit, count, retry_after_ms) + } + } + Err(e) => { + tracing::warn!(error = %e, "redis rate limit script failed"); + RateLimitStatus::backend_error(window_ms) + } + } + } + async fn peek_rate_limit_count(&self, key: &str, _window_ms: u64) -> u64 { let mut conn = self.conn.clone(); let full_key = format!("rl:{}", key); diff --git a/crates/tranquil-config/src/lib.rs b/crates/tranquil-config/src/lib.rs index 976ecbe..2be83ca 100644 --- a/crates/tranquil-config/src/lib.rs +++ b/crates/tranquil-config/src/lib.rs @@ -261,6 +261,26 @@ impl TranquilConfig { // -- tls -------------------------------------------------------------- self.server.tls.validate(&mut errors); + if self + .server + .rate_limit_bypass_key + .as_deref() + .is_some_and(|key| key.trim().is_empty()) + { + errors.push( + "server.rate_limit_bypass_key (PDS_RATE_LIMIT_BYPASS_KEY) must not be empty" + .to_string(), + ); + } + if let Some(ips) = &self.server.rate_limit_bypass_ips { + for ip in ips { + if ip.parse::().is_err() { + errors.push(format!( + "server.rate_limit_bypass_ips (PDS_RATE_LIMIT_BYPASS_IPS) contains invalid IP address: {ip}" + )); + } + } + } // -- cache ------------------------------------------------------------ self.cache.validate(&mut errors); @@ -464,6 +484,14 @@ pub struct ServerConfig { #[config(env = "DISABLE_RATE_LIMITING", default = false)] pub disable_rate_limiting: bool, + /// Client IP addresses that bypass all rate limits. + #[config(env = "PDS_RATE_LIMIT_BYPASS_IPS", parse_env = split_comma_list)] + pub rate_limit_bypass_ips: Option>, + + /// Optional value accepted in the x-ratelimit-bypass request header. + #[config(env = "PDS_RATE_LIMIT_BYPASS_KEY")] + pub rate_limit_bypass_key: Option, + /// Skip the verified-comms-channel gate for login and record writes. /// Please keep this off unless you're an invite-only PDS! #[config(env = "DISABLE_ACCOUNT_VERIFICATION_GATE", default = false)] diff --git a/crates/tranquil-infra/src/lib.rs b/crates/tranquil-infra/src/lib.rs index 9928884..3fb9f04 100644 --- a/crates/tranquil-infra/src/lib.rs +++ b/crates/tranquil-infra/src/lib.rs @@ -60,7 +60,70 @@ pub trait Cache: Send + Sync { #[async_trait] pub trait DistributedRateLimiter: Send + Sync { async fn check_rate_limit(&self, key: &str, limit: u32, window_ms: u64) -> bool; + + async fn consume_rate_limit( + &self, + key: &str, + limit: u32, + window_ms: u64, + points: u32, + ) -> RateLimitStatus { + if points == 0 { + let count = self.peek_rate_limit_count(key, window_ms).await; + return RateLimitStatus::allowed(limit, count, window_ms); + } + for _ in 0..points { + if !self.check_rate_limit(key, limit, window_ms).await { + let count = self.peek_rate_limit_count(key, window_ms).await; + return RateLimitStatus::rejected(limit, count, window_ms); + } + } + let count = self.peek_rate_limit_count(key, window_ms).await; + RateLimitStatus::allowed(limit, count, window_ms) + } + async fn peek_rate_limit_count(&self, _key: &str, _window_ms: u64) -> u64 { 0 } } + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RateLimitStatus { + pub allowed: bool, + pub backend_error: bool, + pub remaining: u32, + pub retry_after_ms: u64, +} + +impl RateLimitStatus { + pub fn allowed(limit: u32, count: u64, retry_after_ms: u64) -> Self { + Self { + allowed: true, + backend_error: false, + remaining: u64::from(limit) + .saturating_sub(count) + .min(u64::from(u32::MAX)) as u32, + retry_after_ms, + } + } + + pub fn rejected(limit: u32, count: u64, retry_after_ms: u64) -> Self { + Self { + allowed: false, + backend_error: false, + remaining: u64::from(limit) + .saturating_sub(count) + .min(u64::from(u32::MAX)) as u32, + retry_after_ms, + } + } + + pub fn backend_error(retry_after_ms: u64) -> Self { + Self { + allowed: false, + backend_error: true, + remaining: 0, + retry_after_ms, + } + } +} diff --git a/crates/tranquil-pds/src/api/error.rs b/crates/tranquil-pds/src/api/error.rs index 7163f92..ba22642 100644 --- a/crates/tranquil-pds/src/api/error.rs +++ b/crates/tranquil-pds/src/api/error.rs @@ -55,6 +55,11 @@ pub enum ApiError { InsufficientScope(Option), InvitesDisabled, RateLimitExceeded(Option), + RateLimitExceededWithStatus { + message: Option, + kind: crate::state::RateLimitKind, + status: crate::cache::RateLimitStatus, + }, PayloadTooLarge(String), TotpAlreadyEnabled, TotpNotEnabled, @@ -179,7 +184,9 @@ impl ApiError { | Self::MfaVerificationRequired | Self::MfaVerificationRequiredWithMethods { .. } | Self::AuthorizationError(_) => StatusCode::FORBIDDEN, - Self::RateLimitExceeded(_) => StatusCode::TOO_MANY_REQUESTS, + Self::RateLimitExceeded(_) | Self::RateLimitExceededWithStatus { .. } => { + StatusCode::TOO_MANY_REQUESTS + } Self::PayloadTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, Self::AccountNotFound | Self::RecordNotFound @@ -319,7 +326,9 @@ impl ApiError { Self::InviteCodeRequired => Cow::Borrowed("InviteCodeRequired"), Self::RepoNotReady => Cow::Borrowed("RepoNotReady"), Self::MfaVerificationRequired => Cow::Borrowed("MfaVerificationRequired"), - Self::RateLimitExceeded(_) => Cow::Borrowed("RateLimitExceeded"), + Self::RateLimitExceeded(_) | Self::RateLimitExceededWithStatus { .. } => { + Cow::Borrowed("RateLimitExceeded") + } Self::PayloadTooLarge(_) => Cow::Borrowed("PayloadTooLarge"), Self::DeviceNotFound => Cow::Borrowed("DeviceNotFound"), Self::NoEmail => Cow::Borrowed("NoEmail"), @@ -395,6 +404,9 @@ impl ApiError { Self::RateLimitExceeded(msg) => msg .clone() .unwrap_or_else(|| "Rate limit exceeded".into()), + Self::RateLimitExceededWithStatus { message, .. } => message + .clone() + .unwrap_or_else(|| "Rate limit exceeded".into()), Self::ServiceUnavailable(msg) => msg .clone() .unwrap_or_else(|| "Service temporarily unavailable".into()), @@ -626,6 +638,13 @@ impl IntoResponse for ApiError { ), ); } + Self::RateLimitExceededWithStatus { kind, status, .. } => { + return crate::rate_limit::with_rate_limit_headers( + response, + kind.params(), + *status, + ); + } _ => {} } response @@ -843,7 +862,15 @@ impl From for ApiError { impl From for ApiError { fn from(e: crate::rate_limit::UserRateLimitError) -> Self { - Self::RateLimitExceeded(e.message) + let status = e.status(); + if status.backend_error { + return Self::InternalError(None); + } + Self::RateLimitExceededWithStatus { + message: e.message, + kind: e.kind, + status, + } } } diff --git a/crates/tranquil-pds/src/api/proxy_client.rs b/crates/tranquil-pds/src/api/proxy_client.rs index a429dbe..b6464ac 100644 --- a/crates/tranquil-pds/src/api/proxy_client.rs +++ b/crates/tranquil-pds/src/api/proxy_client.rs @@ -79,7 +79,8 @@ pub async fn is_ssrf_safe(url: &str) -> Result<(), SsrfError> { let port = parsed .port() .unwrap_or(if scheme == "https" { 443 } else { 80 }); - let socket_addrs: Vec = match tokio::net::lookup_host((host, port)).await { + let socket_addrs: Vec = match tokio::net::lookup_host((host, port)).await + { Ok(addrs) => addrs.collect(), Err(_) => return Err(SsrfError::DnsResolutionFailed(host.to_string())), }; diff --git a/crates/tranquil-pds/src/auth/extractor.rs b/crates/tranquil-pds/src/auth/extractor.rs index 9d3a0d6..3a2c06b 100644 --- a/crates/tranquil-pds/src/auth/extractor.rs +++ b/crates/tranquil-pds/src/auth/extractor.rs @@ -619,10 +619,20 @@ mod tests { #[test] fn scoped_sessions_require_permission_checks() { - assert!(auth_with(Some("repo:app.bsky.feed.post?action=create"), AuthSource::Session) - .needs_scope_check()); - assert!(!auth_with(Some(crate::auth::TokenScope::Access.as_str()), AuthSource::Session) - .needs_scope_check()); + assert!( + auth_with( + Some("repo:app.bsky.feed.post?action=create"), + AuthSource::Session + ) + .needs_scope_check() + ); + assert!( + !auth_with( + Some(crate::auth::TokenScope::Access.as_str()), + AuthSource::Session + ) + .needs_scope_check() + ); assert!(auth_with(None, AuthSource::OAuth).needs_scope_check()); } } diff --git a/crates/tranquil-pds/src/auth/scope_check.rs b/crates/tranquil-pds/src/auth/scope_check.rs index e8273bb..dc9b509 100644 --- a/crates/tranquil-pds/src/auth/scope_check.rs +++ b/crates/tranquil-pds/src/auth/scope_check.rs @@ -93,3 +93,43 @@ pub fn check_identity_scope( .assert_identity(attr) .map_err(|e| ApiError::InsufficientScope(Some(e.to_string()))) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn app_bsky_rpc_wildcard_allows_configured_appview_audience() { + let lxm: Nsid = "app.bsky.feed.getTimeline".parse().expect("valid NSID"); + + assert!( + check_rpc_scope( + &AuthSource::OAuth, + Some("rpc:app.bsky.*?aud=*"), + "did:web:api.bsky.app", + &lxm, + ) + .is_ok() + ); + assert!( + check_rpc_scope( + &AuthSource::OAuth, + Some("rpc:app.bsky.*?aud=%2A"), + "did:web:api.bsky.app", + &lxm, + ) + .is_ok() + ); + + let other_lxm: Nsid = "com.example.getData".parse().expect("valid NSID"); + assert!( + check_rpc_scope( + &AuthSource::OAuth, + Some("rpc:app.bsky.*?aud=*"), + "did:web:api.bsky.app", + &other_lxm, + ) + .is_err() + ); + } +} diff --git a/crates/tranquil-pds/src/cache/mod.rs b/crates/tranquil-pds/src/cache/mod.rs index 52ddb33..5dcda29 100644 --- a/crates/tranquil-pds/src/cache/mod.rs +++ b/crates/tranquil-pds/src/cache/mod.rs @@ -1,4 +1,6 @@ -pub use tranquil_cache::{Cache, CacheError, DistributedRateLimiter, NoOpCache, create_cache}; +pub use tranquil_cache::{ + Cache, CacheError, DistributedRateLimiter, NoOpCache, RateLimitStatus, create_cache, +}; #[cfg(feature = "valkey")] pub use tranquil_cache::{RedisRateLimiter, ValkeyCache}; diff --git a/crates/tranquil-pds/src/delegation/scopes.rs b/crates/tranquil-pds/src/delegation/scopes.rs index 382e4ef..8503011 100644 --- a/crates/tranquil-pds/src/delegation/scopes.rs +++ b/crates/tranquil-pds/src/delegation/scopes.rs @@ -83,11 +83,24 @@ fn scope_coverage(granted: &[ParsedScope], scope: &str, has_owner_access: bool) }; } + if has_owner_access + && matches!( + &requested, + ParsedScope::Rpc(rpc) + if rpc.lxms.len() == 1 + && rpc.lxms.contains("app.bsky.*") + && rpc.aud.as_deref() == Some("*") + ) + { + return GrantCoverage::Full; + } + match coverage(granted, &requested) { Coverage::Full => GrantCoverage::Full, Coverage::Narrowed(ParsedScope::Repo(repo)) => { GrantCoverage::Narrowed(repo.to_scope_string()) } + Coverage::Narrowed(ParsedScope::Rpc(rpc)) => GrantCoverage::Narrowed(rpc.to_string()), Coverage::Narrowed(_) => GrantCoverage::Full, Coverage::Withheld => GrantCoverage::Withheld, } @@ -156,6 +169,39 @@ mod tests { ); } + #[test] + fn test_intersect_owner_grant_covers_app_bsky_rpc_wildcard() { + assert_eq!( + intersect_scopes("rpc:app.bsky.*?aud=*", OWNER_FULL_SCOPES), + "rpc:app.bsky.*?aud=*" + ); + assert_eq!( + intersect_scopes("rpc:app.bsky.*?aud=%2A", OWNER_FULL_SCOPES), + "rpc:app.bsky.*?aud=%2A" + ); + } + + #[test] + fn test_intersect_partial_rpc_grant_is_limited_to_granted_methods() { + assert_eq!( + intersect_scopes( + "rpc:app.bsky.*?aud=*", + "rpc:app.bsky.feed.getTimeline?aud=*" + ), + "rpc:app.bsky.feed.getTimeline?aud=*" + ); + } + + #[test] + fn test_intersect_non_owner_does_not_gain_app_bsky_or_other_rpc_scopes() { + let granted = "repo:* blob:*/* account:*?action=manage"; + assert_eq!(intersect_scopes("rpc:app.bsky.*?aud=*", granted), ""); + assert_eq!( + intersect_scopes("rpc:com.example.*?aud=*", OWNER_FULL_SCOPES), + "" + ); + } + #[test] fn test_intersect_partial_grant_does_not_gain_broad_transition_scopes() { let result = intersect_scopes( diff --git a/crates/tranquil-pds/src/lib.rs b/crates/tranquil-pds/src/lib.rs index 8bf5e31..ce4efef 100644 --- a/crates/tranquil-pds/src/lib.rs +++ b/crates/tranquil-pds/src/lib.rs @@ -99,6 +99,10 @@ pub fn app_with_routes(state: AppState, external: ExternalRoutes) -> Router { .nest("/.well-known", well_known_router) .route("/metrics", get(metrics::metrics_handler)) .merge(external.extra) + .layer(middleware::from_fn_with_state( + state.clone(), + rate_limit::global_xrpc_rate_limit, + )) .layer(DefaultBodyLimit::max(GENERAL_BODY_LIMIT)) .layer(axum::middleware::map_response(rewrite_extractor_errors)) .layer(middleware::from_fn(metrics::metrics_middleware)) @@ -122,6 +126,11 @@ pub fn app_with_routes(state: AppState, external: ExternalRoutes) -> Router { util::HEADER_DPOP_NONCE, util::HEADER_ATPROTO_REPO_REV, util::HEADER_ATPROTO_CONTENT_LABELERS, + http::HeaderName::from_static("ratelimit-limit"), + http::HeaderName::from_static("ratelimit-remaining"), + http::HeaderName::from_static("ratelimit-reset"), + http::HeaderName::from_static("ratelimit-policy"), + http::header::RETRY_AFTER, ]), ) .with_state(state); diff --git a/crates/tranquil-pds/src/rate_limit/extractor.rs b/crates/tranquil-pds/src/rate_limit/extractor.rs index 68a5db7..d30043a 100644 --- a/crates/tranquil-pds/src/rate_limit/extractor.rs +++ b/crates/tranquil-pds/src/rate_limit/extractor.rs @@ -7,8 +7,9 @@ use axum::{ }; use crate::api::error::ApiError; +use crate::cache::RateLimitStatus; use crate::oauth::OAuthError; -use crate::state::{AppState, RateLimitKind}; +use crate::state::{AppState, RateLimitKind, RateLimitParams}; use crate::util::client_ip_from_parts; pub trait RateLimitPolicy: Send + Sync + 'static { @@ -35,6 +36,31 @@ impl RateLimitPolicy for ResetPasswordLimit { const KIND: RateLimitKind = RateLimitKind::ResetPassword; } +pub struct PasswordResetDailyLimit; +impl RateLimitPolicy for PasswordResetDailyLimit { + const KIND: RateLimitKind = RateLimitKind::PasswordResetDaily; +} + +pub struct DeleteAccountLimit; +impl RateLimitPolicy for DeleteAccountLimit { + const KIND: RateLimitKind = RateLimitKind::DeleteAccount; +} + +pub struct GetRepoLimit; +impl RateLimitPolicy for GetRepoLimit { + const KIND: RateLimitKind = RateLimitKind::GetRepo; +} + +pub struct BlobUploadLimit; +impl RateLimitPolicy for BlobUploadLimit { + const KIND: RateLimitKind = RateLimitKind::BlobUpload; +} + +pub struct AccountRequestDailyLimit; +impl RateLimitPolicy for AccountRequestDailyLimit { + const KIND: RateLimitKind = RateLimitKind::AccountRequestDaily; +} + pub struct RefreshSessionLimit; impl RateLimitPolicy for RefreshSessionLimit { const KIND: RateLimitKind = RateLimitKind::RefreshSession; @@ -116,34 +142,69 @@ impl RateLimitPolicy for HandleVerificationLimit { } pub trait RateLimitRejection: IntoResponse + Send + 'static { - fn new() -> Self; + fn new(params: RateLimitParams, status: RateLimitStatus) -> Self; + fn backend_error() -> Self; } -pub struct ApiRateLimitRejection; +pub struct ApiRateLimitRejection { + params: RateLimitParams, + status: RateLimitStatus, +} impl RateLimitRejection for ApiRateLimitRejection { - fn new() -> Self { - Self + fn new(params: RateLimitParams, status: RateLimitStatus) -> Self { + Self { params, status } + } + + fn backend_error() -> Self { + Self { + params: RateLimitKind::Global.params(), + status: RateLimitStatus::backend_error(0), + } } } impl IntoResponse for ApiRateLimitRejection { fn into_response(self) -> Response { - ApiError::RateLimitExceeded(None).into_response() + if self.status.backend_error { + return ApiError::InternalError(None).into_response(); + } + with_rate_limit_headers( + ApiError::RateLimitExceeded(None).into_response(), + self.params, + self.status, + ) } } -pub struct OAuthRateLimitRejection; +pub struct OAuthRateLimitRejection { + params: RateLimitParams, + status: RateLimitStatus, +} impl RateLimitRejection for OAuthRateLimitRejection { - fn new() -> Self { - Self + fn new(params: RateLimitParams, status: RateLimitStatus) -> Self { + Self { params, status } + } + + fn backend_error() -> Self { + Self { + params: RateLimitKind::Global.params(), + status: RateLimitStatus::backend_error(0), + } } } impl IntoResponse for OAuthRateLimitRejection { fn into_response(self) -> Response { - OAuthError::RateLimited.into_response() + if self.status.backend_error { + return OAuthError::ServerError("An internal error occurred".into()).into_response(); + } + with_rate_limit_headers( + OAuthError::RateLimited.into_response(), + self.params, + self.status, + ) } } @@ -175,13 +236,18 @@ impl FromRequestParts ) -> Result { let client_ip = client_ip_from_parts(parts); - if !state.check_rate_limit(P::KIND, &client_ip).await { + let status = state.consume_rate_limit(P::KIND, &client_ip, 1).await; + if status.backend_error { + tracing::error!(kind = ?P::KIND, "Rate limit backend failed"); + return Err(R::backend_error()); + } + if !status.allowed { tracing::warn!( ip = %client_ip, kind = ?P::KIND, "Rate limit exceeded" ); - return Err(R::new()); + return Err(R::new(P::KIND.params(), status)); } Ok(RateLimitedInner { @@ -194,26 +260,62 @@ impl FromRequestParts pub type RateLimited

= RateLimitedInner; pub type OAuthRateLimited

= RateLimitedInner; +pub struct ClientIp(String); + +impl ClientIp { + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Display for ClientIp { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +impl FromRequestParts for ClientIp { + type Rejection = std::convert::Infallible; + + async fn from_request_parts( + parts: &mut Parts, + _state: &AppState, + ) -> Result { + Ok(Self(client_ip_from_parts(parts))) + } +} + #[derive(Debug)] pub struct UserRateLimitError { pub kind: RateLimitKind, pub message: Option, + status: RateLimitStatus, } impl UserRateLimitError { - pub fn new(kind: RateLimitKind) -> Self { + pub fn new(kind: RateLimitKind, status: RateLimitStatus) -> Self { Self { kind, message: None, + status, } } - pub fn with_message(kind: RateLimitKind, message: impl Into) -> Self { + pub fn with_message( + kind: RateLimitKind, + status: RateLimitStatus, + message: impl Into, + ) -> Self { Self { kind, message: Some(message.into()), + status, } } + + pub fn status(&self) -> RateLimitStatus { + self.status + } } impl std::fmt::Display for UserRateLimitError { @@ -229,7 +331,11 @@ impl std::error::Error for UserRateLimitError {} impl IntoResponse for UserRateLimitError { fn into_response(self) -> Response { - ApiError::RateLimitExceeded(self.message).into_response() + with_rate_limit_headers( + ApiError::RateLimitExceeded(self.message).into_response(), + self.kind.params(), + self.status, + ) } } @@ -249,13 +355,18 @@ pub async fn check_user_rate_limit( state: &AppState, user_key: &str, ) -> Result, UserRateLimitError> { - if !state.check_rate_limit(P::KIND, user_key).await { + let status = state.consume_rate_limit(P::KIND, user_key, 1).await; + if status.backend_error { + tracing::error!(kind = ?P::KIND, "Rate limit backend failed"); + return Err(UserRateLimitError::new(P::KIND, status)); + } + if !status.allowed { tracing::warn!( key = %user_key, kind = ?P::KIND, "User rate limit exceeded" ); - return Err(UserRateLimitError::new(P::KIND)); + return Err(UserRateLimitError::new(P::KIND, status)); } Ok(UserRateLimitProof::new()) } @@ -265,13 +376,171 @@ pub async fn check_user_rate_limit_with_message( user_key: &str, error_message: impl Into, ) -> Result, UserRateLimitError> { - if !state.check_rate_limit(P::KIND, user_key).await { + let status = state.consume_rate_limit(P::KIND, user_key, 1).await; + if status.backend_error { + tracing::error!(kind = ?P::KIND, "Rate limit backend failed"); + return Err(UserRateLimitError::new(P::KIND, status)); + } + if !status.allowed { tracing::warn!( key = %user_key, kind = ?P::KIND, "User rate limit exceeded" ); - return Err(UserRateLimitError::with_message(P::KIND, error_message)); + return Err(UserRateLimitError::with_message( + P::KIND, + status, + error_message, + )); } Ok(UserRateLimitProof::new()) } + +pub async fn check_repo_write_rate_limits( + state: &AppState, + did: &str, + points: u32, +) -> Result<(), UserRateLimitError> { + let hourly = state + .consume_rate_limit(RateLimitKind::RepoWriteHourly, did, points) + .await; + let daily = state + .consume_rate_limit(RateLimitKind::RepoWriteDaily, did, points) + .await; + if !hourly.allowed && !hourly.backend_error { + return Err(UserRateLimitError::new( + RateLimitKind::RepoWriteHourly, + hourly, + )); + } + if !daily.allowed && !daily.backend_error { + return Err(UserRateLimitError::new( + RateLimitKind::RepoWriteDaily, + daily, + )); + } + Ok(()) +} + +pub async fn check_login_rate_limits( + state: &AppState, + identifier: &str, + client_ip: &str, +) -> Result<(), UserRateLimitError> { + let key = format!("{identifier}-{client_ip}"); + let short = state + .consume_rate_limit(RateLimitKind::Login, &key, 1) + .await; + let daily = state + .consume_rate_limit(RateLimitKind::LoginDaily, &key, 1) + .await; + if short.backend_error { + return Err(UserRateLimitError::new(RateLimitKind::Login, short)); + } + if daily.backend_error { + return Err(UserRateLimitError::new(RateLimitKind::LoginDaily, daily)); + } + if !short.allowed { + return Err(UserRateLimitError::new(RateLimitKind::Login, short)); + } + if !daily.allowed { + return Err(UserRateLimitError::new(RateLimitKind::LoginDaily, daily)); + } + Ok(()) +} + +pub fn with_rate_limit_headers( + mut response: Response, + params: RateLimitParams, + status: RateLimitStatus, +) -> Response { + use axum::http::{HeaderName, HeaderValue}; + + let window_secs = params.window_ms.div_ceil(1_000); + let retry_after_secs = status.retry_after_ms.div_ceil(1_000); + let reset = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + .saturating_add(retry_after_secs); + let headers = response.headers_mut(); + for (name, value) in [ + ("ratelimit-limit", params.limit.to_string()), + ("ratelimit-remaining", status.remaining.to_string()), + ("ratelimit-reset", reset.to_string()), + ( + "ratelimit-policy", + format!("{};w={}", params.limit, window_secs), + ), + ] { + if let (Ok(name), Ok(value)) = ( + HeaderName::from_bytes(name.as_bytes()), + HeaderValue::from_str(&value), + ) { + headers.insert(name, value); + } + } + if !status.allowed + && let Ok(value) = HeaderValue::from_str(&retry_after_secs.to_string()) + { + headers.insert(http::header::RETRY_AFTER, value); + } + response +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::StatusCode; + + #[test] + fn rejected_response_has_reference_rate_limit_headers() { + let response = with_rate_limit_headers( + StatusCode::TOO_MANY_REQUESTS.into_response(), + RateLimitParams { + limit: 30, + window_ms: 300_000, + }, + RateLimitStatus { + allowed: false, + backend_error: false, + remaining: 0, + retry_after_ms: 12_001, + }, + ); + + assert_eq!(response.headers()["ratelimit-limit"], "30"); + assert_eq!(response.headers()["ratelimit-remaining"], "0"); + assert_eq!(response.headers()["ratelimit-policy"], "30;w=300"); + assert_eq!(response.headers()["retry-after"], "13"); + assert!(response.headers().contains_key("ratelimit-reset")); + } + + #[test] + fn allowed_response_omits_retry_after() { + let response = with_rate_limit_headers( + StatusCode::OK.into_response(), + RateLimitParams { + limit: 30, + window_ms: 300_000, + }, + RateLimitStatus { + allowed: true, + backend_error: false, + remaining: 29, + retry_after_ms: 300_000, + }, + ); + + assert!(!response.headers().contains_key("retry-after")); + } + + #[test] + fn endpoint_limiter_backend_failure_fails_closed() { + let api_response = ApiRateLimitRejection::backend_error().into_response(); + let oauth_response = OAuthRateLimitRejection::backend_error().into_response(); + + assert_eq!(api_response.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(oauth_response.status(), StatusCode::INTERNAL_SERVER_ERROR); + } +} diff --git a/crates/tranquil-pds/src/rate_limit/mod.rs b/crates/tranquil-pds/src/rate_limit/mod.rs index f14d191..dce9b29 100644 --- a/crates/tranquil-pds/src/rate_limit/mod.rs +++ b/crates/tranquil-pds/src/rate_limit/mod.rs @@ -3,12 +3,18 @@ mod extractor; pub use extractor::*; use crate::state::RateLimitKind; +use axum::{ + extract::{Request, State}, + middleware::Next, + response::{IntoResponse, Response}, +}; use governor::{ RateLimiter, clock::DefaultClock, state::{InMemoryState, NotKeyed, keyed::DefaultKeyedStateStore}, }; use std::sync::Arc; +use subtle::ConstantTimeEq; pub type KeyedRateLimiter = RateLimiter, DefaultClock>; pub type GlobalRateLimiter = RateLimiter; @@ -115,6 +121,67 @@ impl RateLimiters { } } +pub async fn global_xrpc_rate_limit( + State(state): State, + request: Request, + next: Next, +) -> Response { + let (parts, body) = request.into_parts(); + let path = parts.uri.path(); + if !path.starts_with("/xrpc/") || path == "/xrpc/com.atproto.sync.getRepo" { + return next.run(Request::from_parts(parts, body)).await; + } + + let client_ip = crate::util::client_ip_from_parts(&parts); + let config = &tranquil_config::get().server; + let parsed_client_ip = client_ip.parse::().ok(); + let ip_bypassed = config.rate_limit_bypass_ips.as_ref().is_some_and(|ips| { + parsed_client_ip.is_some_and(|client_ip| { + ips.iter() + .filter_map(|ip| ip.parse::().ok()) + .any(|ip| ip == client_ip) + }) + }); + let header_bypassed = config + .rate_limit_bypass_key + .as_ref() + .is_some_and(|expected| { + parts + .headers + .get("x-ratelimit-bypass") + .and_then(|value| value.to_str().ok()) + .is_some_and(|actual| { + actual.len() == expected.len() + && bool::from(actual.as_bytes().ct_eq(expected.as_bytes())) + }) + }); + if ip_bypassed || header_bypassed { + return next.run(Request::from_parts(parts, body)).await; + } + + let status = state + .consume_rate_limit(RateLimitKind::Global, &client_ip, 1) + .await; + if status.backend_error { + tracing::error!("Global rate limit backend failed; allowing request"); + return next.run(Request::from_parts(parts, body)).await; + } + if status.allowed { + return extractor::with_rate_limit_headers( + next.run(Request::from_parts(parts, body)).await, + RateLimitKind::Global.params(), + status, + ); + } + + tracing::warn!(ip = %client_ip, path, "Global XRPC rate limit exceeded"); + extractor::with_rate_limit_headers( + crate::api::error::ApiError::RateLimitExceeded(None).into_response(), + RateLimitKind::Global.params(), + status, + ) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index c20362a..7ffd3d2 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -1,5 +1,5 @@ use crate::auth::webauthn::WebAuthnConfig; -use crate::cache::{Cache, DistributedRateLimiter, create_cache}; +use crate::cache::{Cache, DistributedRateLimiter, RateLimitStatus, create_cache}; use crate::circuit_breaker::CircuitBreakers; use crate::config::AuthConfig; use crate::did::DidResolver; @@ -71,7 +71,7 @@ pub struct AppState { pub repo_export_semaphore: Arc, } -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RateLimitParams { pub limit: u32, pub window_ms: u64, @@ -90,10 +90,19 @@ impl RateLimitParams { #[derive(Debug, Clone, Copy)] pub enum RateLimitKind { + Global, Login, + LoginDaily, AccountCreation, PasswordReset, + PasswordResetDaily, ResetPassword, + DeleteAccount, + AccountRequestDaily, + RepoWriteHourly, + RepoWriteDaily, + GetRepo, + BlobUpload, RefreshSession, OAuthToken, OAuthAuthorize, @@ -115,10 +124,19 @@ pub enum RateLimitKind { impl RateLimitKind { const fn key_prefix(&self) -> &'static str { match self { + Self::Global => "global_ip", Self::Login => "login", + Self::LoginDaily => "login_daily", Self::AccountCreation => "account_creation", Self::PasswordReset => "password_reset", + Self::PasswordResetDaily => "password_reset_daily", Self::ResetPassword => "reset_password", + Self::DeleteAccount => "delete_account", + Self::AccountRequestDaily => "account_request_daily", + Self::RepoWriteHourly => "repo_write_hourly", + Self::RepoWriteDaily => "repo_write_daily", + Self::GetRepo => "get_repo", + Self::BlobUpload => "blob_upload", Self::RefreshSession => "refresh_session", Self::OAuthToken => "oauth_token", Self::OAuthAuthorize => "oauth_authorize", @@ -140,21 +158,57 @@ impl RateLimitKind { pub const fn params(&self) -> RateLimitParams { match self { + Self::Global => RateLimitParams { + limit: 3_000, + window_ms: 5 * 60_000, + }, Self::Login => RateLimitParams { - limit: 10, - window_ms: 60_000, + limit: 30, + window_ms: 5 * 60_000, + }, + Self::LoginDaily => RateLimitParams { + limit: 300, + window_ms: 24 * 60 * 60_000, }, Self::AccountCreation => RateLimitParams { - limit: 10, - window_ms: 3_600_000, + limit: 100, + window_ms: 5 * 60_000, }, Self::PasswordReset => RateLimitParams { - limit: 5, - window_ms: 3_600_000, + limit: 15, + window_ms: 60 * 60_000, + }, + Self::PasswordResetDaily => RateLimitParams { + limit: 50, + window_ms: 24 * 60 * 60_000, }, Self::ResetPassword => RateLimitParams { - limit: 10, - window_ms: 60_000, + limit: 50, + window_ms: 5 * 60_000, + }, + Self::DeleteAccount => RateLimitParams { + limit: 50, + window_ms: 5 * 60_000, + }, + Self::AccountRequestDaily => RateLimitParams { + limit: 15, + window_ms: 24 * 60 * 60_000, + }, + Self::RepoWriteHourly => RateLimitParams { + limit: 5_000, + window_ms: 60 * 60_000, + }, + Self::RepoWriteDaily => RateLimitParams { + limit: 35_000, + window_ms: 24 * 60 * 60_000, + }, + Self::GetRepo => RateLimitParams { + limit: 6_000, + window_ms: 5 * 60_000, + }, + Self::BlobUpload => RateLimitParams { + limit: 1_000, + window_ms: 24 * 60 * 60_000, }, Self::RefreshSession => RateLimitParams { limit: 60, @@ -181,8 +235,8 @@ impl RateLimitKind { window_ms: 60_000, }, Self::EmailUpdate => RateLimitParams { - limit: 5, - window_ms: 3_600_000, + limit: 15, + window_ms: 24 * 60 * 60_000, }, Self::TotpVerify => RateLimitParams { limit: 5, @@ -224,6 +278,120 @@ impl RateLimitKind { } } +#[cfg(test)] +mod rate_limit_policy_tests { + use super::{RateLimitKind, RateLimitParams}; + + #[test] + fn reference_method_limits_match_atproto_pds() { + assert_eq!( + RateLimitKind::Global.params(), + RateLimitParams { + limit: 3_000, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::Login.params(), + RateLimitParams { + limit: 30, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::LoginDaily.params(), + RateLimitParams { + limit: 300, + window_ms: 24 * 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::AccountCreation.params(), + RateLimitParams { + limit: 100, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::PasswordReset.params(), + RateLimitParams { + limit: 15, + window_ms: 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::PasswordResetDaily.params(), + RateLimitParams { + limit: 50, + window_ms: 24 * 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::ResetPassword.params(), + RateLimitParams { + limit: 50, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::DeleteAccount.params(), + RateLimitParams { + limit: 50, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::AccountRequestDaily.params(), + RateLimitParams { + limit: 15, + window_ms: 24 * 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::HandleUpdate.params(), + RateLimitParams { + limit: 10, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::HandleUpdateDaily.params(), + RateLimitParams { + limit: 50, + window_ms: 24 * 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::RepoWriteHourly.params(), + RateLimitParams { + limit: 5_000, + window_ms: 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::RepoWriteDaily.params(), + RateLimitParams { + limit: 35_000, + window_ms: 24 * 60 * 60_000, + } + ); + assert_eq!( + RateLimitKind::GetRepo.params(), + RateLimitParams { + limit: 6_000, + window_ms: 5 * 60_000, + } + ); + assert_eq!( + RateLimitKind::BlobUpload.params(), + RateLimitParams { + limit: 1_000, + window_ms: 24 * 60 * 60_000, + } + ); + } +} + impl AppState { pub fn plc_client(&self) -> PlcClient { PlcClient::with_cache(None, Some(self.cache.clone())) @@ -470,53 +638,30 @@ impl AppState { } pub async fn check_rate_limit(&self, kind: RateLimitKind, client_ip: &str) -> bool { + self.consume_rate_limit(kind, client_ip, 1).await.allowed + } + + pub async fn consume_rate_limit( + &self, + kind: RateLimitKind, + key_part: &str, + points: u32, + ) -> RateLimitStatus { if RATE_LIMITING_DISABLED.load(Ordering::Relaxed) { - return true; + return RateLimitStatus::allowed(u32::MAX, 0, 0); } let limiter_name = kind.key_prefix(); - - let limiter = match kind { - RateLimitKind::Login => &self.rate_limiters.login, - RateLimitKind::AccountCreation => &self.rate_limiters.account_creation, - RateLimitKind::PasswordReset => &self.rate_limiters.password_reset, - RateLimitKind::ResetPassword => &self.rate_limiters.reset_password, - RateLimitKind::RefreshSession => &self.rate_limiters.refresh_session, - RateLimitKind::OAuthToken => &self.rate_limiters.oauth_token, - RateLimitKind::OAuthAuthorize => &self.rate_limiters.oauth_authorize, - RateLimitKind::OAuthPar => &self.rate_limiters.oauth_par, - RateLimitKind::OAuthIntrospect => &self.rate_limiters.oauth_introspect, - RateLimitKind::AppPassword => &self.rate_limiters.app_password, - RateLimitKind::EmailUpdate => &self.rate_limiters.email_update, - RateLimitKind::TotpVerify => &self.rate_limiters.totp_verify, - RateLimitKind::HandleUpdate => &self.rate_limiters.handle_update, - RateLimitKind::HandleUpdateDaily => &self.rate_limiters.handle_update_daily, - RateLimitKind::VerificationCheck => &self.rate_limiters.verification_check, - RateLimitKind::SsoInitiate => &self.rate_limiters.sso_initiate, - RateLimitKind::SsoCallback => &self.rate_limiters.sso_callback, - RateLimitKind::SsoUnlink => &self.rate_limiters.sso_unlink, - RateLimitKind::OAuthRegisterComplete => &self.rate_limiters.oauth_register_complete, - RateLimitKind::HandleVerification => &self.rate_limiters.handle_verification, - }; - - if limiter.check_key(&client_ip.to_string()).is_err() { - crate::metrics::record_rate_limit_rejection(limiter_name); - return false; - } - - let key = format!("{}:{}", kind.key_prefix(), client_ip); + let key = format!("{}:{}", kind.key_prefix(), key_part); let params = kind.params(); - - if !self + let status = self .distributed_rate_limiter - .check_rate_limit(&key, params.limit, params.window_ms) - .await - { + .consume_rate_limit(&key, params.limit, params.window_ms, points) + .await; + if !status.allowed { crate::metrics::record_rate_limit_rejection(limiter_name); - return false; } - - true + status } } diff --git a/crates/tranquil-ripple/src/crdt/g_counter.rs b/crates/tranquil-ripple/src/crdt/g_counter.rs index 37cac82..997c66a 100644 --- a/crates/tranquil-ripple/src/crdt/g_counter.rs +++ b/crates/tranquil-ripple/src/crdt/g_counter.rs @@ -25,8 +25,12 @@ impl GCounter { } pub fn increment(&mut self, node_id: u64) { + self.increment_by(node_id, 1); + } + + pub fn increment_by(&mut self, node_id: u64, points: u32) { let slot = self.increments.entry(node_id).or_insert(0); - *slot = slot.saturating_add(1); + *slot = slot.saturating_add(u64::from(points)); } pub fn merge(&mut self, other: &GCounter) -> bool { @@ -79,8 +83,19 @@ impl RateLimitStore { window_ms: u64, now_wall_ms: u64, ) -> bool { + self.consume(key, limit, window_ms, 1, now_wall_ms).0 + } + + pub fn consume( + &mut self, + key: &str, + limit: u32, + window_ms: u64, + points: u32, + now_wall_ms: u64, + ) -> (bool, u64, u64) { if window_ms == 0 { - return false; + return (false, 0, 0); } let window_start = Self::aligned_window_start(now_wall_ms, window_ms); @@ -95,12 +110,19 @@ impl RateLimitStore { .or_insert_with(|| GCounter::new(window_start, window_ms)); let current = counter.total(); - if current >= limit as u64 { - return false; + let retry_after_ms = window_start + .saturating_add(window_ms) + .saturating_sub(now_wall_ms); + if current.saturating_add(u64::from(points)) > u64::from(limit) { + return (false, current, retry_after_ms); } - counter.increment(self.node_id); + counter.increment_by(self.node_id, points); self.dirty.insert(key.to_string()); - true + ( + true, + current.saturating_add(u64::from(points)), + retry_after_ms, + ) } pub fn merge_counter(&mut self, key: String, remote: &GCounter) -> bool { @@ -290,6 +312,24 @@ mod tests { assert!(!store.check_and_increment("k", 3, 60_000, 400)); } + #[test] + fn weighted_consumption_is_atomic() { + let mut store = RateLimitStore::new(1); + + assert_eq!(store.consume("k", 5, 60_000, 3, 100), (true, 3, 59_900)); + assert_eq!(store.consume("k", 5, 60_000, 3, 200), (false, 3, 59_800)); + assert_eq!(store.peek_count("k", 60_000, 300), 3); + assert_eq!(store.consume("k", 5, 60_000, 2, 400), (true, 5, 59_600)); + } + + #[test] + fn zero_point_check_does_not_change_count() { + let mut store = RateLimitStore::new(1); + + assert_eq!(store.consume("k", 5, 60_000, 0, 100), (true, 0, 59_900)); + assert_eq!(store.peek_count("k", 60_000, 200), 0); + } + #[test] fn gc_expired_windows() { let mut store = RateLimitStore::new(1); diff --git a/crates/tranquil-ripple/src/crdt/mod.rs b/crates/tranquil-ripple/src/crdt/mod.rs index 808f245..c65f39a 100644 --- a/crates/tranquil-ripple/src/crdt/mod.rs +++ b/crates/tranquil-ripple/src/crdt/mod.rs @@ -122,6 +122,19 @@ impl ShardedCrdtStore { .check_and_increment(key, limit, window_ms, Self::wall_ms_now()) } + pub fn rate_limit_consume( + &self, + key: &str, + limit: u32, + window_ms: u64, + points: u32, + ) -> (bool, u64, u64) { + self.shards[self.shard_for(key)] + .write() + .rate_limits + .consume(key, limit, window_ms, points, Self::wall_ms_now()) + } + pub fn peek_broadcast_delta(&self) -> CrdtDelta { let mut cache_entries: Vec<(String, lww_map::LwwEntry)> = Vec::new(); let mut rate_limit_deltas: Vec = Vec::new(); diff --git a/crates/tranquil-ripple/src/rate_limiter.rs b/crates/tranquil-ripple/src/rate_limiter.rs index 93e84ca..9b362b1 100644 --- a/crates/tranquil-ripple/src/rate_limiter.rs +++ b/crates/tranquil-ripple/src/rate_limiter.rs @@ -1,7 +1,7 @@ use crate::crdt::ShardedCrdtStore; use async_trait::async_trait; use std::sync::Arc; -use tranquil_infra::DistributedRateLimiter; +use tranquil_infra::{DistributedRateLimiter, RateLimitStatus}; pub struct RippleRateLimiter { store: Arc, @@ -19,6 +19,22 @@ impl DistributedRateLimiter for RippleRateLimiter { self.store.rate_limit_check(key, limit, window_ms) } + async fn consume_rate_limit( + &self, + key: &str, + limit: u32, + window_ms: u64, + points: u32, + ) -> RateLimitStatus { + let (allowed, count, retry_after_ms) = + self.store.rate_limit_consume(key, limit, window_ms, points); + if allowed { + RateLimitStatus::allowed(limit, count, retry_after_ms) + } else { + RateLimitStatus::rejected(limit, count, retry_after_ms) + } + } + async fn peek_rate_limit_count(&self, key: &str, window_ms: u64) -> u64 { self.store.rate_limit_peek(key, window_ms) } diff --git a/crates/tranquil-scopes/src/coverage.rs b/crates/tranquil-scopes/src/coverage.rs index 173d1fd..e4c60fd 100644 --- a/crates/tranquil-scopes/src/coverage.rs +++ b/crates/tranquil-scopes/src/coverage.rs @@ -69,6 +69,41 @@ pub fn coverage(granted: &[ParsedScope], requested: &ParsedScope) -> Coverage { }; } + if let ParsedScope::Rpc(r) = requested { + if granted + .iter() + .any(|g| matches!(g, ParsedScope::Rpc(g) if rpc_covers(g, r))) + { + return Coverage::Full; + } + + let intersections: Vec = granted + .iter() + .filter_map(|g| match g { + ParsedScope::Rpc(g) => rpc_intersection(g, r), + _ => None, + }) + .collect(); + let Some(first) = intersections.first() else { + return Coverage::Withheld; + }; + if intersections.iter().any(|scope| scope.aud != first.aud) { + return Coverage::Withheld; + } + let aud = first.aud.clone(); + + let lxms = intersections + .into_iter() + .flat_map(|scope| scope.lxms) + .collect(); + let narrowed = RpcScope { lxms, aud }; + return if narrowed == *r { + Coverage::Full + } else { + Coverage::Narrowed(ParsedScope::Rpc(narrowed)) + }; + } + if granted.iter().any(|g| covers(g, requested)) { Coverage::Full } else { @@ -92,10 +127,11 @@ fn blob_covers(g: &BlobScope, r: &BlobScope) -> bool { } fn rpc_covers(g: &RpcScope, r: &RpcScope) -> bool { - let lxm_ok = r - .lxms - .iter() - .all(|requested| g.lxms.contains("*") || g.lxms.contains(requested)); + let lxm_ok = r.lxms.iter().all(|requested| { + g.lxms + .iter() + .any(|granted| rpc_lxm_covers(granted, requested)) + }); let aud_ok = match &g.aud { None => true, Some(ga) if ga == "*" => true, @@ -104,6 +140,45 @@ fn rpc_covers(g: &RpcScope, r: &RpcScope) -> bool { lxm_ok && aud_ok } +fn rpc_lxm_covers(granted: &str, requested: &str) -> bool { + granted == "*" + || granted == requested + || granted.strip_suffix(".*").is_some_and(|prefix| { + requested.starts_with(prefix) && requested.as_bytes().get(prefix.len()) == Some(&b'.') + }) +} + +fn rpc_intersection(granted: &RpcScope, requested: &RpcScope) -> Option { + let lxms = granted + .lxms + .iter() + .flat_map(|granted_lxm| { + requested.lxms.iter().filter_map(move |requested_lxm| { + if rpc_lxm_covers(granted_lxm, requested_lxm) { + Some(requested_lxm.clone()) + } else if rpc_lxm_covers(requested_lxm, granted_lxm) { + Some(granted_lxm.clone()) + } else { + None + } + }) + }) + .collect::>(); + if lxms.is_empty() { + return None; + } + + let aud = match (&granted.aud, &requested.aud) { + (Some(granted), Some(requested)) if granted != "*" && requested != "*" => { + (granted == requested).then(|| Some(granted.clone()))? + } + (Some(granted), _) if granted != "*" => Some(granted.clone()), + (_, Some(requested)) if requested != "*" => Some(requested.clone()), + (_, requested) => requested.clone(), + }; + Some(RpcScope { lxms, aud }) +} + fn account_covers(g: &AccountScope, r: &AccountScope) -> bool { let attr_ok = g.attr == AccountAttr::Wildcard || g.attr == r.attr; let action_ok = g.actions.contains(&AccountAction::Manage) || r.actions.is_subset(&g.actions); @@ -198,6 +273,10 @@ mod tests { #[test] fn rpc_wildcards() { assert!(c("rpc:*?aud=did:web:x", "rpc:app.bsky.getX?aud=did:web:x")); + assert!(c( + "rpc:app.bsky.*?aud=*", + "rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app" + )); assert!(c( "rpc:app.bsky.getX?aud=*", "rpc:app.bsky.getX?aud=did:web:x" @@ -206,6 +285,35 @@ mod tests { "rpc:app.bsky.getX?aud=did:web:x", "rpc:app.bsky.getY?aud=did:web:x" )); + assert!(!c( + "rpc:app.bsky.*?aud=*", + "rpc:app.bskyextra.getX?aud=did:web:x" + )); + } + + #[test] + fn rpc_namespace_request_is_narrowed_to_granted_methods() { + assert_eq!( + covered( + "rpc:app.bsky.feed.getTimeline?aud=*", + "rpc:app.bsky.*?aud=*" + ), + narrowed_to("rpc:app.bsky.feed.getTimeline?aud=*") + ); + assert_eq!( + covered( + "rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app", + "rpc:app.bsky.*?aud=*" + ), + narrowed_to("rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app") + ); + assert_eq!( + covered( + "rpc:app.bsky.feed.getTimeline?aud=did:web:other.example", + "rpc:app.bsky.*?aud=did:web:api.bsky.app" + ), + Coverage::Withheld + ); } #[test] diff --git a/crates/tranquil-sync/src/repo.rs b/crates/tranquil-sync/src/repo.rs index a06e9d2..d25248f 100644 --- a/crates/tranquil-sync/src/repo.rs +++ b/crates/tranquil-sync/src/repo.rs @@ -10,6 +10,7 @@ use std::str::FromStr; use tracing::error; use tranquil_pds::api::error::ApiError; use tranquil_pds::api::query::XrpcQuery; +use tranquil_pds::rate_limit::{GetRepoLimit, RateLimited}; use tranquil_pds::scheduled::generate_repo_car_from_user_blocks; use tranquil_pds::state::AppState; use tranquil_pds::sync::car::{encode_car_block, encode_car_header}; @@ -120,6 +121,7 @@ pub struct GetRepoQuery { pub async fn get_repo( State(state): State, + _rate_limit: RateLimited, XrpcQuery(query): XrpcQuery, ) -> Response { let did = query.did; diff --git a/docs/2_INSTALL_CONTAINERS.md b/docs/2_INSTALL_CONTAINERS.md index a278c9a..a9ed5ab 100644 --- a/docs/2_INSTALL_CONTAINERS.md +++ b/docs/2_INSTALL_CONTAINERS.md @@ -31,7 +31,9 @@ Tranquil does not request or renew certs. Keep using certbot, acme.sh, lego, ste ### Client IP and forwarded headers -Rate limiting and device records are based on the client IP ofc. Behind a reverse proxy, Tranquil reads it from `X-Forwarded-For`, counting hops from the right. You can set `TRUSTED_PROXY_COUNT` for how many proxies to trust. Leave it unset to let Tranquil assume the count. We assumes 1 proxy when something else terminates TLS, and 0 when Tranquil terminates TLS itself. At 0 it uses the direct conn address and ignores forwarded headers that a direct client could maliciously invent. +Rate limiting and device records use the client IP. Behind reverse proxies, Tranquil reads it from `X-Forwarded-For`, counting trusted hops from the right. Set `TRUSTED_PROXY_COUNT` to every trusted proxy in the request path: use `1` for a single nginx, Caddy, or Traefik proxy, and `2` for a CDN such as Cloudflare in front of that proxy. Each proxy must also be configured to trust forwarded headers only from the proxy or CDN immediately before it. + +When this setting is omitted, Tranquil assumes `1` if something else terminates TLS and `0` when Tranquil terminates TLS itself. At `0`, it uses the direct connection address and ignores forwarded headers that a client could maliciously invent. ## Quickstart (docker/podman compose) diff --git a/example.toml b/example.toml index 21a4a84..4147f72 100644 --- a/example.toml +++ b/example.toml @@ -62,6 +62,17 @@ # Default value: false #disable_rate_limiting = false +# Client IP addresses that bypass all rate limits. +# +# Can also be specified via environment variable `PDS_RATE_LIMIT_BYPASS_IPS`. +#rate_limit_bypass_ips = [] + +# Optional value accepted in the `x-ratelimit-bypass` request header. +# The value must not be empty. +# +# Can also be specified via environment variable `PDS_RATE_LIMIT_BYPASS_KEY`. +#rate_limit_bypass_key = "replace-with-a-random-value" + # Skip the verified-comms-channel gate for login and record writes. # Please keep this off unless you're an invite-only PDS! # @@ -124,6 +135,9 @@ # When left unset, Tranquil will assume: # - 0, if the TLS termination is happening here on Tranquil via the TLS config # - 1, if the TLS termination *isn't* happening here. +# Set this to 2 when a CDN such as Cloudflare sits in front of nginx, Caddy, +# Traefik, or another trusted ingress proxy. Configure each proxy to accept +# forwarded headers only from its trusted upstream networks. # # Can also be specified via environment variable `TRUSTED_PROXY_COUNT`. #trusted_proxy_count =