From 5222c301ff838507e29a62675909dfeba6ecb43d Mon Sep 17 00:00:00 2001 From: Julius Joska Date: Fri, 23 Oct 2026 11:04:33 +0200 Subject: [PATCH] fix(relay): bounded resources for traffic spikes - Reject envelopes for addresses without a device (404) instead of queueing them for 30 days; random addresses were the cheapest way to fill the relay's disk. - Per-sender hourly quota on stored envelopes (count and bytes, 429) and a count cap on each recipient's queue next to the byte cap. - Per-account attachment quota: bytes per day and bytes stored within the TTL, kept in memory only (no account-to-blob link is stored). - Disk watermarks: below a configurable share of free space the relay refuses new attachments, then new offline envelopes (507), before Postgres runs out of space. - WebSocket buffers 4 KiB read / unbuffered write: ~16.6 KiB per connection instead of ~137 KiB (10 500 connections measured at ~180 MiB including the DB pool). - reqwest with HTTP/2: APNs only accepts HTTP/2, so no iOS wake-up could ever be delivered. APNs registration with only a VoIP token is stored instead of deleted. - All limits configurable via environment (limits.rs). Defaults keep previous behaviour except registration per IP (50 + 1/10 s, for carrier-grade NAT) guarded by a global cap (40/s), separate Tor buckets (SEPOT_TOR_PEERS) and a rate limit on opening WebSockets. SEPOT_REGISTRATION_PAUSED=1 pauses new registrations (503). - Envelopes whose encoding exceeds the queue column limit are 400 before they count against the sender quota. Rate limiting moves from tower_governor to governor directly, so Tor and clearnet can use different quotas on the same route. --- rust/Cargo.lock | 147 ++---------- rust/sepot-relay/Cargo.toml | 8 +- rust/sepot-relay/src/apns.rs | 22 ++ rust/sepot-relay/src/attachments.rs | 93 ++++++-- rust/sepot-relay/src/disk.rs | 113 +++++++++ rust/sepot-relay/src/limits.rs | 336 +++++++++++++++++++++++++++ rust/sepot-relay/src/main.rs | 89 +++----- rust/sepot-relay/src/net.rs | 104 +++++---- rust/sepot-relay/src/quota.rs | 194 ++++++++++++++++ rust/sepot-relay/src/rest.rs | 13 +- rust/sepot-relay/src/state.rs | 340 ++++++++++++++++++++++++---- rust/sepot-relay/src/ws.rs | 47 +++- 12 files changed, 1181 insertions(+), 325 deletions(-) create mode 100644 rust/sepot-relay/src/disk.rs create mode 100644 rust/sepot-relay/src/limits.rs create mode 100644 rust/sepot-relay/src/quota.rs diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 9be946b..836d6e5 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -633,16 +633,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "forwarded-header-value" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8835f84f38484cc86f110a805655697908257fb9a7af005234060891557198e9" -dependencies = [ - "nonempty", - "thiserror 1.0.69", -] - [[package]] name = "futures-channel" version = "0.3.32" @@ -1059,19 +1049,6 @@ dependencies = [ "webpki-roots 1.0.8", ] -[[package]] -name = "hyper-timeout" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" -dependencies = [ - "hyper", - "hyper-util", - "pin-project-lite", - "tokio", - "tower-service", -] - [[package]] name = "hyper-util" version = "0.1.20" @@ -1440,7 +1417,7 @@ dependencies = [ "rand 0.9.4", "sha2", "subtle", - "thiserror 2.0.18", + "thiserror", "uuid", "x25519-dalek", "zerocopy", @@ -1488,7 +1465,7 @@ dependencies = [ "signal-crypto", "spqr", "subtle", - "thiserror 2.0.18", + "thiserror", "uuid", "zerocopy", ] @@ -1590,12 +1567,6 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" -[[package]] -name = "nonempty" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e9e591e719385e6ebaeb5ce5d3887f7d5676fceca6411d1925ccc95745f3d6f7" - [[package]] name = "nonzero_ext" version = "0.3.0" @@ -1788,26 +1759,6 @@ dependencies = [ "indexmap", ] -[[package]] -name = "pin-project" -version = "1.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" -dependencies = [ - "pin-project-internal", -] - -[[package]] -name = "pin-project-internal" -version = "1.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "pin-project-lite" version = "0.2.17" @@ -2071,7 +2022,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror 2.0.18", + "thiserror", "tokio", "tracing", "web-time", @@ -2092,7 +2043,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror 2.0.18", + "thiserror", "tinyvec", "tracing", "web-time", @@ -2277,6 +2228,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "h2", "http", "http-body", "http-body-util", @@ -2445,6 +2397,7 @@ dependencies = [ "chrono", "dashmap", "futures-util", + "governor", "hex", "hmac", "ipnet", @@ -2463,7 +2416,6 @@ dependencies = [ "tokio", "tokio-util", "tower-http 0.7.0", - "tower_governor", "tracing", "tracing-subscriber", ] @@ -2603,7 +2555,7 @@ dependencies = [ "sha1", "sha2", "subtle", - "thiserror 2.0.18", + "thiserror", "zeroize", ] @@ -2635,7 +2587,7 @@ checksum = "0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d" dependencies = [ "num-bigint", "num-traits", - "thiserror 2.0.18", + "thiserror", "time", ] @@ -2718,7 +2670,7 @@ dependencies = [ "rand_core 0.9.5", "sha2", "sorted-vec", - "thiserror 2.0.18", + "thiserror", ] [[package]] @@ -2763,7 +2715,7 @@ dependencies = [ "serde_json", "sha2", "smallvec", - "thiserror 2.0.18", + "thiserror", "tokio", "tokio-stream", "tracing", @@ -2847,7 +2799,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror 2.0.18", + "thiserror", "tracing", "whoami", ] @@ -2885,7 +2837,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror 2.0.18", + "thiserror", "tracing", "whoami", ] @@ -2910,7 +2862,7 @@ dependencies = [ "serde", "serde_urlencoded", "sqlx-core", - "thiserror 2.0.18", + "thiserror", "tracing", "url", ] @@ -2982,33 +2934,13 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "thiserror" -version = "1.0.69" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" -dependencies = [ - "thiserror-impl 1.0.69", -] - [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl 2.0.18", -] - -[[package]] -name = "thiserror-impl" -version = "1.0.69" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" -dependencies = [ - "proc-macro2", - "quote", - "syn", + "thiserror-impl", ] [[package]] @@ -3190,35 +3122,6 @@ dependencies = [ "winnow", ] -[[package]] -name = "tonic" -version = "0.14.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" -dependencies = [ - "async-trait", - "axum", - "base64", - "bytes", - "h2", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-timeout", - "hyper-util", - "percent-encoding", - "pin-project", - "socket2", - "sync_wrapper", - "tokio", - "tokio-stream", - "tower", - "tower-layer", - "tower-service", - "tracing", -] - [[package]] name = "tower" version = "0.5.3" @@ -3227,12 +3130,9 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", - "indexmap", "pin-project-lite", - "slab", "sync_wrapper", "tokio", - "tokio-util", "tower-layer", "tower-service", "tracing", @@ -3285,23 +3185,6 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" -[[package]] -name = "tower_governor" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44de9b94d849d3c46e06a883d72d408c2de6403367b39df2b1c9d9e7b6736fe6" -dependencies = [ - "axum", - "forwarded-header-value", - "governor", - "http", - "pin-project", - "thiserror 2.0.18", - "tonic", - "tower", - "tracing", -] - [[package]] name = "tracing" version = "0.1.44" @@ -3383,7 +3266,7 @@ dependencies = [ "log", "rand 0.9.4", "sha1", - "thiserror 2.0.18", + "thiserror", ] [[package]] diff --git a/rust/sepot-relay/Cargo.toml b/rust/sepot-relay/Cargo.toml index 3426e37..a6afcd4 100644 --- a/rust/sepot-relay/Cargo.toml +++ b/rust/sepot-relay/Cargo.toml @@ -42,13 +42,15 @@ tokio-util = { version = "0.7", features = ["io"] } # při AUTH. Relay NIKDY nešifruje/nedešifruje zprávy (hloupý relay). libsignal-protocol = { git = "https://github.com/signalapp/libsignal", tag = "v0.96.4" } libsignal-core = { git = "https://github.com/signalapp/libsignal", tag = "v0.96.4" } -tower_governor = "0.8.0" +# Rate limity (token bucket per IP / Tor / globálně, limits.rs) +governor = "0.10" tower-http = { version = "0.7.0", features = ["timeout"] } # Push budíčky (UnifiedPush/WebPush = POST na endpoint). rustls, ať build # zůstane bez OpenSSL (stejně jako sqlx tls-rustls). Bez default featur — -# nechceme native-tls ani zbytečné kodeky. -reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } +# nechceme native-tls ani zbytečné kodeky. http2: APNs jiný protokol +# nepřijme (HTTP/1.1 spojení odmítne) — bez něj nedorazil žádný iOS budíček. +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "http2"] } # FCM fast mode: RS256 JWT pro OAuth2 service account flow jsonwebtoken = "9" # setrlimit(RLIMIT_NOFILE) při startu — docker dává soft limit 1024 diff --git a/rust/sepot-relay/src/apns.rs b/rust/sepot-relay/src/apns.rs index 59154cf..84ffa1c 100644 --- a/rust/sepot-relay/src/apns.rs +++ b/rust/sepot-relay/src/apns.rs @@ -201,4 +201,26 @@ mod tests { assert!(limit <= 30 * 86_400, "APNs nedrží notifikace déle než 30 dní"); } + + /// APNs mluví jen HTTP/2. Bez featury `http2` v reqwestu relay spojení + /// vůbec nenaváže a iPhone žádný budíček nedostane. Bez platného JWT + /// vrátí Apple 403 (nebo 400) — to stačí: spojení i protokol sedí. + #[tokio::test] + #[ignore = "vyžaduje síť (api.push.apple.com)"] + async fn apns_spojeni_jde_pres_http2() { + let client = crate::push::build_client(); + let res = client + .post(format!("{PROD_HOST}/3/device/00")) + .header("apns-topic", "cz.sepot.sepot") + .body("{}") + .send() + .await + .expect("spojení s APNs"); + assert_eq!(res.version(), reqwest::Version::HTTP_2); + assert!( + matches!(res.status().as_u16(), 400 | 403), + "čekána odpověď APNs bez platného tokenu, přišlo {}", + res.status() + ); + } } diff --git a/rust/sepot-relay/src/attachments.rs b/rust/sepot-relay/src/attachments.rs index 98bedc6..d146452 100644 --- a/rust/sepot-relay/src/attachments.rs +++ b/rust/sepot-relay/src/attachments.rs @@ -111,23 +111,85 @@ pub async fn upload( headers: HeaderMap, body: Body, ) -> ApiResult<(StatusCode, Json)> { - crate::rest::authenticate_device(&state, &headers, "POST /v1/attachments").await?; + let authed = + crate::rest::authenticate_device(&state, &headers, "POST /v1/attachments").await?; let cfg = &state.attachments; // Rychlé odmítnutí podle Content-Length, než tečou data - if let Some(len) = headers + let declared = headers .get(header::CONTENT_LENGTH) .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) + .and_then(|v| v.parse::().ok()); + if declared.is_some_and(|len| len > cfg.max_bytes) { + return Err(( + StatusCode::PAYLOAD_TOO_LARGE, + "příloha je příliš velká".into(), + )); + } + + // Místo na disku dřív než cokoli dalšího: přílohy a DB sdílí filesystém + // a plný disk by zastavil i zprávy (disk.rs). + if state + .disk + .below(state.limits.disk_min_free_pct_attachments) { - if len > cfg.max_bytes { - return Err(( - StatusCode::PAYLOAD_TOO_LARGE, - "příloha je příliš velká".into(), - )); + return Err(( + StatusCode::INSUFFICIENT_STORAGE, + "úložiště příloh je plné, zkus to později".into(), + )); + } + + // Kvóta účtu: rezervuje se ohlášená velikost (bez Content-Length strop + // jedné přílohy) a po zápisu se nespotřebovaný zbytek vrátí. Jen + // v paměti — vazba účet → blob se neukládá (viz hlavička modulu). + let reserved = declared.unwrap_or(cfg.max_bytes); + let account = authed.account_id; + if !state.attachment_budget.try_take(&account, reserved) { + return Err(( + StatusCode::TOO_MANY_REQUESTS, + "denní limit příloh vyčerpán, zkus to později".into(), + )); + } + if !state.attachment_stored.try_reserve(&account, reserved) { + state.attachment_budget.refund(&account, reserved); + return Err(( + StatusCode::TOO_MANY_REQUESTS, + "příliš mnoho uložených příloh, počkej, až je příjemci stáhnou".into(), + )); + } + let refund = |unused: u64| { + state.attachment_budget.refund(&account, unused); + state.attachment_stored.release(&account, unused); + }; + let result = store_upload(&state, body).await; + match &result { + Ok(stored) => refund(reserved.saturating_sub(stored.size_bytes)), + Err(_) => refund(reserved), + } + result.map(|stored| (StatusCode::CREATED, Json(stored.into()))) +} + +/// Výsledek zápisu včetně skutečné velikosti (pro vyúčtování kvóty). +struct Stored { + blob_id: String, + expires_at_ms: i64, + max_bytes: u64, + size_bytes: u64, +} + +impl From for UploadOut { + fn from(s: Stored) -> Self { + Self { + blob_id: s.blob_id, + expires_at_ms: s.expires_at_ms, + max_bytes: s.max_bytes, } } +} + +async fn store_upload(state: &AppState, body: Body) -> ApiResult { + let cfg = &state.attachments; // Globální strop: součet živých příloh + tahle nesmí přesáhnout // total_bytes. Jeden SELECT nad malou tabulkou; nic o zařízení se @@ -192,14 +254,12 @@ pub async fn upload( // Log bez blob_id — jen velikost (kapacitní metrika) tracing::debug!(size = written, "příloha uložena"); - Ok(( - StatusCode::CREATED, - Json(UploadOut { - blob_id: blob_hex, - expires_at_ms: expires_at.timestamp_millis(), - max_bytes: cfg.max_bytes, - }), - )) + Ok(Stored { + blob_id: blob_hex, + expires_at_ms: expires_at.timestamp_millis(), + max_bytes: cfg.max_bytes, + size_bytes: written, + }) } async fn write_stream(body: Body, path: &FsPath, max_bytes: u64) -> ApiResult { @@ -343,6 +403,7 @@ mod db_tests { total_bytes: 1 << 30, }, trusted, + crate::limits::Limits::from_env(), ) } diff --git a/rust/sepot-relay/src/disk.rs b/rust/sepot-relay/src/disk.rs new file mode 100644 index 0000000..3ad5385 --- /dev/null +++ b/rust/sepot-relay/src/disk.rs @@ -0,0 +1,113 @@ +//! Hlídání volného místa na datovém filesystému. +//! +//! Přílohy i DB leží v produkci na jednom filesystému (svazky Dockeru v CT). +//! Když dojde místo, Postgres přestane zapisovat a relay je mrtvý pro všechny +//! — i pro registrace a doručení online. Relay proto přestane přijímat to, co +//! disk plní (přílohy, pak offline fronta) o kus dřív, než místo dojde. + +use std::ffi::CString; +use std::path::Path; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +/// Jak často se místo měří. Zápis 25 MiB přílohy trvá déle než tohle okno, +/// takže práh nepřeskočí ani souběžný nápor. +const MEASURE_EVERY: Duration = Duration::from_secs(5); +/// Pod 20 % volna se do logu zapíše varování (dřív, než zabere kterýkoli +/// práh odmítání), aby si toho Loki alert všiml s předstihem. +const WARN_BELOW_BP: u32 = 2000; + +/// Volné místo v setinách procenta; `u32::MAX` = ještě neměřeno / nejde měřit +/// (pak se nic neblokuje — chyba měření nesmí shodit provoz). +pub struct DiskWatch { + free_bp: AtomicU32, +} + +impl DiskWatch { + pub fn unknown() -> Self { + Self { + free_bp: AtomicU32::new(u32::MAX), + } + } + + /// Volné místo v procentech, `None` = neznámé. + pub fn free_pct(&self) -> Option { + match self.free_bp.load(Ordering::Relaxed) { + u32::MAX => None, + bp => Some(f64::from(bp) / 100.0), + } + } + + /// Je volného místa méně než `min_pct`? + pub fn below(&self, min_pct: f64) -> bool { + self.free_pct().is_some_and(|free| free < min_pct) + } + + #[cfg(test)] + pub fn set_free_pct(&self, pct: f64) { + self.free_bp.store((pct * 100.0) as u32, Ordering::Relaxed); + } + + pub fn spawn(self: &Arc, path: &Path) { + let watch = self.clone(); + let path = path.to_path_buf(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(MEASURE_EVERY); + let mut was_low = false; + loop { + tick.tick().await; + let measured = { + let path = path.clone(); + tokio::task::spawn_blocking(move || free_basis_points(&path)) + .await + .ok() + .flatten() + }; + let Some(bp) = measured else { + tracing::warn!("volné místo na disku nejde změřit"); + continue; + }; + watch.free_bp.store(bp, Ordering::Relaxed); + // Změnu stavu zalogovat jednou, ne každých 5 s. + let low = bp < WARN_BELOW_BP; + if low != was_low { + tracing::warn!(volno_pct = f64::from(bp) / 100.0, "volné místo na disku"); + was_low = low; + } + } + }); + } +} + +/// statvfs: místo dostupné neprivilegovanému procesu (`f_bavail`) vůči +/// velikosti filesystému. Na ZFS s refquota je velikost = kvóta datasetu. +fn free_basis_points(path: &Path) -> Option { + let c_path = CString::new(path.as_os_str().as_encoded_bytes()).ok()?; + // SAFETY: statvfs jen zapisuje do předaného structu, cesta je platný C řetězec. + let mut st: libc::statvfs = unsafe { std::mem::zeroed() }; + if unsafe { libc::statvfs(c_path.as_ptr(), &mut st) } != 0 || st.f_blocks == 0 { + return None; + } + Some(((st.f_bavail as u128 * 10_000) / st.f_blocks as u128) as u32) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn mereni_vrati_rozumne_cislo() { + let bp = free_basis_points(Path::new("/")).expect("statvfs /"); + assert!(bp <= 10_000); + } + + #[test] + fn nezname_misto_nic_neblokuje() { + let w = DiskWatch::unknown(); + assert!(!w.below(99.0)); + w.set_free_pct(5.0); + assert!(w.below(8.0)); + assert!(!w.below(4.0)); + } +} diff --git a/rust/sepot-relay/src/limits.rs b/rust/sepot-relay/src/limits.rs new file mode 100644 index 0000000..bb4af48 --- /dev/null +++ b/rust/sepot-relay/src/limits.rs @@ -0,0 +1,336 @@ +//! Limity relaye — všechny nastavitelné přes env. +//! +//! Proč env a ne konstanty: limit, který se pod skutečnou zátěží ukáže +//! špatně (sdílené CGNAT adresy, Tor, nápor po zveřejnění), se musí dát +//! změnit úpravou `.env` a restartem (sekundy), ne novým buildem. +//! +//! Výchozí hodnoty drží chování dřívějších verzí se třemi vědomými +//! výjimkami popsanými u polí: registrace per IP je výrazně volnější (mobilní +//! CGNAT dá za jednu IPv4 stovky lidí) a DB místo toho chrání globální strop; +//! Tor klienti mají vlastní, větší buckety (všichni přicházejí z jedné +//! adresy); a nové kvóty (odesílatel, přílohy na účet, místo na disku) dřív +//! neexistovaly. +//! +//! Rate limity jsou token buckety: `*_BURST` tokenů naráz, jeden token zpět +//! každých `*_PERIOD_MS` ms. Kvóty bajtů/kusů: 0 = bez omezení. + +use std::net::{IpAddr, SocketAddr}; +use std::num::NonZeroU32; +use std::sync::Arc; +use std::time::Duration; + +use axum::extract::{ConnectInfo, Request, State}; +use axum::http::{header, HeaderValue, StatusCode}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use governor::clock::{Clock, DefaultClock}; +use governor::{DefaultDirectRateLimiter, DefaultKeyedRateLimiter, Quota}; + +use crate::net::TrustedProxies; + +const MIB: u64 = 1024 * 1024; + +fn env_parse(name: &str, default: T) -> T { + match std::env::var(name) { + Ok(raw) => match raw.trim().parse() { + Ok(v) => v, + Err(_) => { + tracing::warn!(name, "neplatná hodnota, platí výchozí"); + default + } + }, + Err(_) => default, + } +} + +/// Token bucket: `burst` požadavků naráz, pak jeden za `period`. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Rate { + pub burst: u32, + pub period: Duration, +} + +impl Rate { + pub const fn new(burst: u32, period_ms: u64) -> Self { + Self { + burst, + period: Duration::from_millis(period_ms), + } + } + + /// `{prefix}_BURST` a `{prefix}_PERIOD_MS`, každá s vlastní výchozí hodnotou. + fn from_env(prefix: &str, default: Rate) -> Self { + let burst = env_parse(&format!("{prefix}_BURST"), default.burst).max(1); + let period_ms = env_parse( + &format!("{prefix}_PERIOD_MS"), + default.period.as_millis() as u64, + ) + .max(1); + Self::new(burst, period_ms) + } + + fn quota(&self) -> Quota { + Quota::with_period(self.period) + .expect("period > 0") + .allow_burst(NonZeroU32::new(self.burst).expect("burst > 0")) + } +} + +/// Oddělené buckety: clearnet klienti (podle IP klienta) a všichni Tor +/// klienti dohromady (sdílí adresu kontejneru Toru). +#[derive(Clone, Copy, Debug)] +pub struct RatePair { + pub clearnet: Rate, + pub tor: Rate, +} + +#[derive(Clone, Debug)] +pub struct Limits { + /// Spojení relaye do Postgresu (`SEPOT_DB_POOL`, výchozí 8). + pub db_pool: u32, + + /// `SEPOT_RL_REGISTER_*` na IP: 50 naráz, pak 1 za 10 s (360/h). + /// Dřív 10 + 1/5 min: za CGNAT mobilního operátora čekal jedenáctý + /// člověk na účet minuty. + pub register: RatePair, + /// `SEPOT_RL_REGISTER_GLOBAL_*`: všechny registrace dohromady, 600 naráz, + /// pak 40/s. Chrání DB (změřeno 117–147 registrací/s na 2 jádrech), když + /// registruje mnoho adres naráz; BURST 0 = vypnuto. + pub register_global: Option, + /// `SEPOT_REGISTRATION_PAUSED=1`: nové registrace dostanou 503 + Retry-After + /// (nouzový vypínač při záplavě registrací); existující zařízení jedou dál. + pub registration_paused: bool, + /// `SEPOT_RL_GENERAL_*`: server-info, bundly klíčů, certifikát, smazání. + pub general: RatePair, + /// `SEPOT_RL_ATTACHMENTS_*`: upload i stažení příloh. + pub attachments: RatePair, + /// `SEPOT_RL_WS_*`: otevírání WS spojení (každý AUTH = dotaz do DB). + pub ws_upgrade: RatePair, + + /// `SEPOT_MAX_CONNS_PER_IP` (výchozí 32) a `SEPOT_TOR_MAX_CONNS` + /// (výchozí 5000, všichni Tor klienti dohromady). + pub max_conns_per_ip: u32, + pub max_conns_tor: u32, + /// `SEPOT_WS_FRAMES_PER_10S`: drahé rámce na spojení (výchozí 200). + pub ws_frames_per_window: u32, + + /// `SEPOT_OTP_ISSUE_PER_HOUR`: vydaných one-time prekeys na adresu za hodinu. + pub otp_issue_per_hour: u32, + /// `SEPOT_MAX_STORED_OTP`: zásoba one-time prekeys na zařízení. + pub max_stored_otp: i64, + /// `SEPOT_CALL_WAKES_PER_10MIN` a `SEPOT_CALL_WAKE_MIN_INTERVAL_S`. + pub call_wakes_per_window: usize, + pub call_wake_min_interval: Duration, + + /// `SEPOT_QUEUE_MAX_BYTES_PER_RECIPIENT` (50 MiB) a + /// `SEPOT_QUEUE_MAX_PER_RECIPIENT` (10 000 obálek): přebytek padá od nejstarších. + pub queue_max_bytes_per_recipient: i64, + pub queue_max_per_recipient: i64, + /// `SEPOT_SEND_MAX_PER_HOUR` / `SEPOT_SEND_MAX_BYTES_PER_HOUR` na odesílající + /// účet (ukládané obálky; „píše…“ se nepočítá). + pub send_per_hour: u64, + pub send_bytes_per_hour: u64, + + /// `SEPOT_ATT_ACCOUNT_BYTES_PER_DAY` (100 MiB) a + /// `SEPOT_ATT_ACCOUNT_STORED_BYTES` (250 MiB živých v rámci TTL). + pub att_account_bytes_per_day: u64, + pub att_account_stored_bytes: u64, + + /// `SEPOT_DISK_MIN_FREE_PCT_ATTACHMENTS` (15) a `SEPOT_DISK_MIN_FREE_PCT_QUEUE` + /// (8): pod tímto podílem volného místa na datovém filesystému relay + /// odmítá nové přílohy, resp. nové offline obálky (507) — Postgres si tak + /// nechá místo na WAL a na všechno, co už přijal. + pub disk_min_free_pct_attachments: f64, + pub disk_min_free_pct_queue: f64, +} + +impl Limits { + pub fn from_env() -> Self { + let pair = |prefix: &str, clearnet: Rate, tor: Rate| RatePair { + clearnet: Rate::from_env(prefix, clearnet), + tor: Rate::from_env(&format!("{prefix}_TOR"), tor), + }; + let global = Rate::from_env("SEPOT_RL_REGISTER_GLOBAL", Rate::new(600, 25)); + let global_off = env_parse("SEPOT_RL_REGISTER_GLOBAL_BURST", 600u32) == 0; + Self { + db_pool: env_parse("SEPOT_DB_POOL", 8u32).max(1), + register: pair( + "SEPOT_RL_REGISTER", + Rate::new(50, 10_000), + Rate::new(100, 3_000), + ), + register_global: (!global_off).then_some(global), + registration_paused: matches!( + std::env::var("SEPOT_REGISTRATION_PAUSED").as_deref(), + Ok("1") | Ok("true") + ), + general: pair("SEPOT_RL_GENERAL", Rate::new(60, 500), Rate::new(600, 20)), + attachments: pair( + "SEPOT_RL_ATTACHMENTS", + Rate::new(30, 1_000), + Rate::new(300, 100), + ), + ws_upgrade: pair("SEPOT_RL_WS", Rate::new(120, 100), Rate::new(600, 20)), + max_conns_per_ip: env_parse("SEPOT_MAX_CONNS_PER_IP", 32u32), + max_conns_tor: env_parse("SEPOT_TOR_MAX_CONNS", 5000u32), + ws_frames_per_window: env_parse("SEPOT_WS_FRAMES_PER_10S", 200u32), + otp_issue_per_hour: env_parse("SEPOT_OTP_ISSUE_PER_HOUR", 30u32), + max_stored_otp: env_parse("SEPOT_MAX_STORED_OTP", 100i64), + call_wakes_per_window: env_parse("SEPOT_CALL_WAKES_PER_10MIN", 6usize), + call_wake_min_interval: Duration::from_secs(env_parse( + "SEPOT_CALL_WAKE_MIN_INTERVAL_S", + 10u64, + )), + queue_max_bytes_per_recipient: env_parse( + "SEPOT_QUEUE_MAX_BYTES_PER_RECIPIENT", + (50 * MIB) as i64, + ), + queue_max_per_recipient: env_parse("SEPOT_QUEUE_MAX_PER_RECIPIENT", 10_000i64), + send_per_hour: env_parse("SEPOT_SEND_MAX_PER_HOUR", 5_000u64), + send_bytes_per_hour: env_parse("SEPOT_SEND_MAX_BYTES_PER_HOUR", 32 * MIB), + att_account_bytes_per_day: env_parse("SEPOT_ATT_ACCOUNT_BYTES_PER_DAY", 100 * MIB), + att_account_stored_bytes: env_parse("SEPOT_ATT_ACCOUNT_STORED_BYTES", 250 * MIB), + disk_min_free_pct_attachments: env_parse("SEPOT_DISK_MIN_FREE_PCT_ATTACHMENTS", 15.0), + disk_min_free_pct_queue: env_parse("SEPOT_DISK_MIN_FREE_PCT_QUEUE", 8.0), + } + } +} + +/// Rate limiter skupiny rout, jen v paměti. Nic se nezapisuje ani neloguje +/// (adresy klientů jsou osobní údaj). +pub struct RateLimit { + clearnet: DefaultKeyedRateLimiter, + tor: DefaultDirectRateLimiter, + global: Option, + trusted: Arc, +} + +impl RateLimit { + pub fn new(rates: RatePair, global: Option, trusted: Arc) -> Arc { + let limiter = Arc::new(Self { + clearnet: DefaultKeyedRateLimiter::keyed(rates.clearnet.quota()), + tor: DefaultDirectRateLimiter::direct(rates.tor.quota()), + global: global.map(|g| DefaultDirectRateLimiter::direct(g.quota())), + trusted, + }); + // Zapomenout nečinné buckety, ať mapa neroste s každou kdy viděnou IP. + let weak = Arc::downgrade(&limiter); + tokio::spawn(async move { + let mut tick = tokio::time::interval(Duration::from_secs(60)); + loop { + tick.tick().await; + let Some(limiter) = weak.upgrade() else { break }; + limiter.clearnet.retain_recent(); + } + }); + limiter + } + + /// `Ok` = povoleno, `Err(wait)` = odmítnuto, zkusit znovu za `wait`. + pub fn check(&self, peer: IpAddr, headers: &axum::http::HeaderMap) -> Result<(), Duration> { + let now = DefaultClock::default().now(); + let own = if self.trusted.is_tor(peer) { + self.tor.check() + } else { + self.clearnet + .check_key(&self.trusted.client_ip(peer, headers)) + }; + own.map_err(|not_until| not_until.wait_time_from(now))?; + if let Some(global) = &self.global { + global.check().map_err(|not_until| not_until.wait_time_from(now))?; + } + Ok(()) + } +} + +/// axum middleware: 429 s Retry-After, když je bucket prázdný. +pub async fn enforce( + State(limit): State>, + ConnectInfo(peer): ConnectInfo, + request: Request, + next: Next, +) -> Response { + match limit.check(peer.ip(), request.headers()) { + Ok(()) => next.run(request).await, + Err(wait) => too_many(wait), + } +} + +fn too_many(wait: Duration) -> Response { + let secs = wait.as_secs() + u64::from(wait.subsec_nanos() > 0); + let mut response = (StatusCode::TOO_MANY_REQUESTS, "příliš mnoho požadavků").into_response(); + if let Ok(value) = HeaderValue::from_str(&secs.max(1).to_string()) { + response.headers_mut().insert(header::RETRY_AFTER, value); + } + response +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::HeaderMap; + + fn trusted(proxies: &str, tor: &str) -> Arc { + Arc::new(TrustedProxies::parse(proxies, tor)) + } + + fn xff(ip: &str) -> HeaderMap { + let mut h = HeaderMap::new(); + h.insert("x-forwarded-for", ip.parse().unwrap()); + h + } + + #[tokio::test] + async fn clearnet_bucket_je_na_ip_klienta() { + let rates = RatePair { + clearnet: Rate::new(2, 3_600_000), + tor: Rate::new(100, 1), + }; + let rl = RateLimit::new(rates, None, trusted("10.0.0.1", "")); + let proxy: IpAddr = "10.0.0.1".parse().unwrap(); + assert!(rl.check(proxy, &xff("198.51.100.1")).is_ok()); + assert!(rl.check(proxy, &xff("198.51.100.1")).is_ok()); + let wait = rl.check(proxy, &xff("198.51.100.1")).unwrap_err(); + assert!(wait > Duration::from_secs(3000), "Retry-After odpovídá periodě"); + assert!(rl.check(proxy, &xff("198.51.100.2")).is_ok(), "jiný klient limit nesdílí"); + } + + #[tokio::test] + async fn tor_ma_vlastni_bucket_a_nepodvrhne_ip() { + let rates = RatePair { + clearnet: Rate::new(1, 3_600_000), + tor: Rate::new(3, 3_600_000), + }; + let rl = RateLimit::new(rates, None, trusted("10.0.0.1", "172.20.1.0/24")); + let tor_peer: IpAddr = "172.20.1.3".parse().unwrap(); + for i in 0..3 { + // podvržená XFF z Toru nesmí založit nové buckety + assert!(rl.check(tor_peer, &xff(&format!("203.0.113.{i}"))).is_ok()); + } + assert!(rl.check(tor_peer, &xff("203.0.113.99")).is_err()); + // vyčerpaný Tor bucket clearnet klienta neomezí + let proxy: IpAddr = "10.0.0.1".parse().unwrap(); + assert!(rl.check(proxy, &xff("198.51.100.1")).is_ok()); + } + + #[tokio::test] + async fn globalni_strop_plati_pres_klienty() { + let rates = RatePair { + clearnet: Rate::new(100, 1), + tor: Rate::new(100, 1), + }; + let rl = RateLimit::new(rates, Some(Rate::new(2, 3_600_000)), trusted("10.0.0.1", "")); + let proxy: IpAddr = "10.0.0.1".parse().unwrap(); + assert!(rl.check(proxy, &xff("198.51.100.1")).is_ok()); + assert!(rl.check(proxy, &xff("198.51.100.2")).is_ok()); + assert!(rl.check(proxy, &xff("198.51.100.3")).is_err()); + } + + #[test] + fn rate_z_env_padne_na_vychozi_po_polich() { + std::env::set_var("SEPOT_TEST_RL_BURST", "7"); + std::env::set_var("SEPOT_TEST_RL_PERIOD_MS", "nonsense"); + let rate = Rate::from_env("SEPOT_TEST_RL", Rate::new(1, 250)); + assert_eq!(rate, Rate::new(7, 250)); + } +} diff --git a/rust/sepot-relay/src/main.rs b/rust/sepot-relay/src/main.rs index 85cfdf2..57f2230 100644 --- a/rust/sepot-relay/src/main.rs +++ b/rust/sepot-relay/src/main.rs @@ -8,9 +8,12 @@ mod apns; mod attachments; mod certs; +mod disk; mod fcm; +mod limits; mod net; mod push; +mod quota; mod rest; mod state; mod turn; @@ -22,14 +25,14 @@ use std::time::Duration; use axum::extract::DefaultBodyLimit; use axum::http::StatusCode; +use axum::middleware::from_fn_with_state; use axum::routing::{delete, get, post}; use axum::serve::ListenerExt; use axum::Router; use sqlx::postgres::PgPoolOptions; -use tower_governor::governor::GovernorConfigBuilder; -use tower_governor::GovernorLayer; -use crate::net::{TrustedProxies, TrustedProxyIp}; +use crate::limits::{Limits, RateLimit}; +use crate::net::TrustedProxies; use tower_http::timeout::TimeoutLayer; use tracing_subscriber::EnvFilter; @@ -49,8 +52,9 @@ async fn main() -> anyhow::Result<()> { .unwrap_or_else(|_| "postgres://sepot:sepot_dev@127.0.0.1/sepot_relay".into()); let bind = std::env::var("SEPOT_BIND").unwrap_or_else(|_| "127.0.0.1:8470".into()); + let limits = Limits::from_env(); let pool = PgPoolOptions::new() - .max_connections(8) + .max_connections(limits.db_pool) .connect(&database_url) .await?; sqlx::migrate!("./migrations").run(&pool).await?; @@ -66,60 +70,23 @@ async fn main() -> anyhow::Result<()> { certs, attachments_cfg, trusted.clone(), + limits.clone(), )); attachments::spawn_cleanup(state.clone()); + state.disk.spawn(&state.attachments.dir); + tracing::info!(?limits, "limity"); - // ── Rate limiting (minimální sada; plný PoW/Privacy Pass později) ── + // ── Rate limiting ── // - // Klíč = klientská IP přes TrustedProxyIp: X-Forwarded-For/X-Real-Ip se - // uzná JEN od trusted proxy (traefik), jinak peer adresa. Onion klienti - // přicházejí z tor kontejneru (mimo trusted sadu) → sdílí jeden bucket - // a nemohou si IP nastavit hlavičkou (SECURITY audit #1). - // In-memory token bucket — nikdy se nezapisuje na disk (privacy). - - // registrace: zápis do DB bez auth → nejpřísnější (burst 10, +1/5 min) - let gov_register = Arc::new( - GovernorConfigBuilder::default() - .key_extractor(TrustedProxyIp(trusted.clone())) - .burst_size(10) - .period(Duration::from_secs(300)) - .finish() - .expect("governor config"), - ); - // čtecí API (bundle žere OTP klíče → drain ochrana): burst 60, +2/s - let gov_general = Arc::new( - GovernorConfigBuilder::default() - .key_extractor(TrustedProxyIp(trusted.clone())) - .burst_size(60) - .per_second(2) - .finish() - .expect("governor config"), - ); - - // přílohy: upload je autentizovaný, ale drahý (disk) → burst 30, +1/s - let gov_attachments = Arc::new( - GovernorConfigBuilder::default() - .key_extractor(TrustedProxyIp(trusted.clone())) - .burst_size(30) - .per_second(1) - .finish() - .expect("governor config"), - ); - - // Governor drží mapu IP→bucket v paměti — periodicky vyhazovat staré - for limiter in [ - gov_register.limiter().clone(), - gov_general.limiter().clone(), - gov_attachments.limiter().clone(), - ] { - tokio::spawn(async move { - let mut tick = tokio::time::interval(Duration::from_secs(60)); - loop { - tick.tick().await; - limiter.retain_recent(); - } - }); - } + // Klíč = klientská IP: X-Forwarded-For se uzná JEN od trusted proxy + // (traefik), jinak peer adresa. Onion klienti přicházejí z kontejneru Toru + // (SEPOT_TOR_PEERS) a mají vlastní společný bucket — hlavičkou si IP + // nastavit nemohou (SECURITY audit #1). Jen v paměti, nikdy na disk. + // Hodnoty a jejich zdůvodnění: limits.rs. + let rl_register = RateLimit::new(limits.register, limits.register_global, trusted.clone()); + let rl_general = RateLimit::new(limits.general, None, trusted.clone()); + let rl_attachments = RateLimit::new(limits.attachments, None, trusted.clone()); + let rl_ws = RateLimit::new(limits.ws_upgrade, None, trusted.clone()); // Úklid značek budíčků a OTP oken (jen paměť) { @@ -135,7 +102,7 @@ async fn main() -> anyhow::Result<()> { let register_routes = Router::new() .route("/v1/register", post(rest::register)) - .layer(GovernorLayer::new(gov_register)); + .layer(from_fn_with_state(rl_register, crate::limits::enforce)); let general_routes = Router::new() .route("/v1/server-info", get(rest::server_info)) @@ -144,7 +111,7 @@ async fn main() -> anyhow::Result<()> { .route("/v1/keys", post(rest::upload_keys)) .route("/v1/certificate/delivery", get(rest::certificate_delivery)) .route("/v1/identity", delete(rest::delete_identity)) - .layer(GovernorLayer::new(gov_general)); + .layer(from_fn_with_state(rl_general, crate::limits::enforce)); let rest_routes = register_routes .merge(general_routes) @@ -159,7 +126,7 @@ async fn main() -> anyhow::Result<()> { let attachment_routes = Router::new() .route("/v1/attachments", post(attachments::upload)) .route("/v1/attachments/{blob_id}", get(attachments::download)) - .layer(GovernorLayer::new(gov_attachments)) + .layer(from_fn_with_state(rl_attachments, crate::limits::enforce)) .layer(TimeoutLayer::with_status_code( StatusCode::REQUEST_TIMEOUT, Duration::from_secs(120), @@ -169,8 +136,12 @@ async fn main() -> anyhow::Result<()> { // health bez limitů — interní docker/uptime checky; ready ověří i DB .route("/v1/health", get(rest::health)) .route("/v1/ready", get(rest::ready)) - // WS má vlastní limity (conn cap per IP, auth deadline, frame rate) - .route("/v1/ws", get(ws::ws_handler)) + // WS: tempo otevírání spojení + vlastní limity v ws.rs (strop + // souběžných spojení na IP, AUTH deadline, tempo rámců) + .route( + "/v1/ws", + get(ws::ws_handler).layer(from_fn_with_state(rl_ws, crate::limits::enforce)), + ) .merge(rest_routes) .merge(attachment_routes) .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)) diff --git a/rust/sepot-relay/src/net.rs b/rust/sepot-relay/src/net.rs index 30ac2d0..883b263 100644 --- a/rust/sepot-relay/src/net.rs +++ b/rust/sepot-relay/src/net.rs @@ -14,19 +14,34 @@ //! proxy. Onion peer (tor kontejner) tam není → použije se jeho IP a všichni //! onion klienti sdílí jeden bucket (přesně jak počítal původní komentář). -use std::net::{IpAddr, SocketAddr}; -use std::sync::Arc; +use std::net::IpAddr; -use axum::extract::ConnectInfo; -use axum::http::{HeaderMap, Request}; +use axum::http::HeaderMap; use ipnet::IpNet; -use tower_governor::key_extractor::KeyExtractor; -use tower_governor::GovernorError; /// Sada sítí, jejichž forwarded hlavičkám se věří (traefik/pangolin). /// Konfiguruje se přes `SEPOT_TRUSTED_PROXIES` (čárkou oddělené IP nebo CIDR). pub struct TrustedProxies { nets: Vec, + /// Adresy, ze kterých přichází Tor (kontejner `tor` v síti tornet, + /// `SEPOT_TOR_PEERS`). Takový peer nikdy není trusted proxy a dostává + /// vlastní, společné limity — všichni onion klienti mají jeho IP. + tor: Vec, +} + +fn parse_nets(raw: &str) -> Vec { + raw.split(',') + .filter_map(|s| { + let s = s.trim(); + if s.is_empty() { + return None; + } + // "10.0.0.0/24" i holé "10.0.0.5" (bereme jako /32, resp. /128) + s.parse::() + .ok() + .or_else(|| s.parse::().ok().map(IpNet::from)) + }) + .collect() } impl TrustedProxies { @@ -34,31 +49,37 @@ impl TrustedProxies { /// nikomu (vždy peer IP) — fail-safe: chybějící konfigurace nesmí otevřít /// spoofing, maximálně sloučí clearnet klienty do jednoho bucketu. pub fn from_env() -> Self { - let raw = std::env::var("SEPOT_TRUSTED_PROXIES").unwrap_or_default(); - let nets: Vec = raw - .split(',') - .filter_map(|s| { - let s = s.trim(); - if s.is_empty() { - return None; - } - // "10.0.0.0/24" i holé "10.0.0.5" (bereme jako /32, resp. /128) - s.parse::() - .ok() - .or_else(|| s.parse::().ok().map(IpNet::from)) - }) - .collect(); - if nets.is_empty() { + let proxies = Self::parse( + &std::env::var("SEPOT_TRUSTED_PROXIES").unwrap_or_default(), + &std::env::var("SEPOT_TOR_PEERS").unwrap_or_default(), + ); + if proxies.nets.is_empty() { tracing::warn!( "SEPOT_TRUSTED_PROXIES prázdné — forwarded hlavičky se ignorují, \ anti-abuse klíč = peer IP. Za reverse proxy nastav CIDR proxy sítě." ); } - Self { nets } + proxies + } + + /// Prázdné/nenastavené proxy = nevěří se nikomu (vždy peer IP) — fail-safe: + /// chybějící konfigurace nesmí otevřít spoofing, maximálně sloučí clearnet + /// klienty do jednoho bucketu. Tor má přednost: síť, která je zároveň + /// v obou sadách, se jako proxy nebere (onion klient by si jinak IP + /// nastavil hlavičkou). + pub fn parse(proxies: &str, tor: &str) -> Self { + Self { + nets: parse_nets(proxies), + tor: parse_nets(tor), + } + } + + pub fn is_tor(&self, peer: IpAddr) -> bool { + self.tor.iter().any(|n| n.contains(&peer)) } fn trusts(&self, peer: IpAddr) -> bool { - self.nets.iter().any(|n| n.contains(&peer)) + !self.is_tor(peer) && self.nets.iter().any(|n| n.contains(&peer)) } /// Klientská IP pro anti-abuse. Forwarded hlavičku uznej jen od trusted @@ -83,24 +104,6 @@ impl TrustedProxies { } } -/// Governor key extractor s ověřením trusted proxy. Nahrazuje -/// `SmartIpKeyExtractor`, který forwarded hlavičkám věří bezpodmínečně. -#[derive(Clone)] -pub struct TrustedProxyIp(pub Arc); - -impl KeyExtractor for TrustedProxyIp { - type Key = IpAddr; - - fn extract(&self, req: &Request) -> Result { - let peer = req - .extensions() - .get::>() - .map(|ci| ci.0.ip()) - .ok_or(GovernorError::UnableToExtractKey)?; - Ok(self.0.client_ip(peer, req.headers())) - } -} - #[cfg(test)] mod tests { use super::*; @@ -108,15 +111,7 @@ mod tests { use axum::http::HeaderMap; fn proxies(cfg: &[&str]) -> TrustedProxies { - let nets = cfg - .iter() - .filter_map(|s| { - s.parse::() - .ok() - .or_else(|| s.parse::().ok().map(IpNet::from)) - }) - .collect(); - TrustedProxies { nets } + TrustedProxies::parse(&cfg.join(","), "") } fn ip(s: &str) -> IpAddr { @@ -191,4 +186,13 @@ mod tests { let h = headers(&[("x-forwarded-for", "2001:db8::1")]); assert_eq!(tp.client_ip(peer, &h), ip("2001:db8::1")); } + + /// Síť Toru v sadě proxy omylem: onion peer si IP hlavičkou nastavit nesmí. + #[test] + fn tor_peer_neni_proxy_ani_pri_prekryvu() { + let tp = TrustedProxies::parse("172.20.0.0/16", "172.20.1.0/24"); + let tor = ip("172.20.1.3"); + assert!(tp.is_tor(tor)); + assert_eq!(tp.client_ip(tor, &headers(&[("x-forwarded-for", "1.2.3.4")])), tor); + } } diff --git a/rust/sepot-relay/src/quota.rs b/rust/sepot-relay/src/quota.rs new file mode 100644 index 0000000..ed0b21c --- /dev/null +++ b/rust/sepot-relay/src/quota.rs @@ -0,0 +1,194 @@ +//! Kvóty na účet — jen v paměti, nikam se neukládají ani nelogují. +//! +//! Proč v paměti: relay záměrně nikde nedrží vazbu „účet → co poslal / jakou +//! přílohu nahrál“ (metadata). Cena: restart relaye kvóty vynuluje. Proti +//! zaplnění disku proto stojí ještě globální strop příloh a hlídání volného +//! místa (`disk.rs`), které restart nevynuluje. + +use std::collections::VecDeque; +use std::time::{Duration, Instant}; + +use dashmap::DashMap; + +/// Token bucket počtu i bajtů na klíč: plný na `count`/`bytes`, za `period` +/// se doplní celý. 0 = daná dimenze bez omezení. +pub struct Budgets { + count: u64, + bytes: u64, + period: Duration, + map: DashMap, Bucket>, +} + +struct Bucket { + count: f64, + bytes: f64, + last: Instant, +} + +impl Budgets { + pub fn new(count: u64, bytes: u64, period: Duration) -> Self { + Self { + count, + bytes, + period, + map: DashMap::new(), + } + } + + fn refill(&self, b: &mut Bucket, now: Instant) { + let share = now.duration_since(b.last).as_secs_f64() / self.period.as_secs_f64(); + b.count = (b.count + share * self.count as f64).min(self.count as f64); + b.bytes = (b.bytes + share * self.bytes as f64).min(self.bytes as f64); + b.last = now; + } + + /// Odečte jednu položku o `bytes` bajtech; `false` = kvóta vyčerpaná + /// (nic se neodečte). + pub fn try_take(&self, key: &[u8], bytes: u64) -> bool { + self.try_take_at(key, bytes, Instant::now()) + } + + fn try_take_at(&self, key: &[u8], bytes: u64, now: Instant) -> bool { + let mut b = self.map.entry(key.to_vec()).or_insert_with(|| Bucket { + count: self.count as f64, + bytes: self.bytes as f64, + last: now, + }); + self.refill(&mut b, now); + let count_ok = self.count == 0 || b.count >= 1.0; + let bytes_ok = self.bytes == 0 || b.bytes >= bytes as f64; + if !(count_ok && bytes_ok) { + return false; + } + b.count -= 1.0; + b.bytes -= bytes as f64; + true + } + + /// Vrátí bajty, které se nakonec nespotřebovaly (upload menší než + /// rezervace, chyba zápisu). + pub fn refund(&self, key: &[u8], bytes: u64) { + if let Some(mut b) = self.map.get_mut(key) { + b.bytes = (b.bytes + bytes as f64).min(self.bytes as f64); + } + } + + /// Úklid plných bucketů (účet dlouho nic neposlal) — paměť jen pro aktivní. + pub fn prune(&self) { + let now = Instant::now(); + self.map.retain(|_, b| { + let share = now.duration_since(b.last).as_secs_f64() / self.period.as_secs_f64(); + share < 1.0 + }); + } +} + +/// Kolik bajtů má účet právě uložených (příloh v rámci TTL). Účetnictví po +/// dnech: na účet nejvýš `ttl_days` čísel, ne seznam blobů — paměť je pevná +/// a vazba na jednotlivé bloby nevzniká. +pub struct StoredLedger { + cap: u64, + ttl_days: i64, + map: DashMap, VecDeque<(i64, u64)>>, +} + +impl StoredLedger { + pub fn new(cap: u64, ttl_days: i64) -> Self { + Self { + cap, + ttl_days, + map: DashMap::new(), + } + } + + fn today() -> i64 { + chrono::Utc::now().timestamp().div_euclid(86_400) + } + + /// Zarezervuje `bytes`; `false` = účet by přesáhl strop uloženého. + pub fn try_reserve(&self, key: &[u8], bytes: u64) -> bool { + self.try_reserve_on(key, bytes, Self::today()) + } + + fn try_reserve_on(&self, key: &[u8], bytes: u64, day: i64) -> bool { + if self.cap == 0 { + return true; + } + let mut days = self.map.entry(key.to_vec()).or_default(); + // Den nahrání + TTL dní: pak je blob úklidem pryč (zaokrouhleno nahoru). + while days.front().is_some_and(|(d, _)| *d + self.ttl_days < day) { + days.pop_front(); + } + let used: u64 = days.iter().map(|(_, b)| b).sum(); + if used.saturating_add(bytes) > self.cap { + return false; + } + match days.back_mut() { + Some((d, b)) if *d == day => *b += bytes, + _ => days.push_back((day, bytes)), + } + true + } + + pub fn release(&self, key: &[u8], bytes: u64) { + if let Some(mut days) = self.map.get_mut(key) { + if let Some((_, b)) = days.back_mut() { + *b = b.saturating_sub(bytes); + } + } + } + + pub fn prune(&self) { + let horizon = Self::today() - self.ttl_days; + self.map + .retain(|_, days| days.back().is_some_and(|(d, _)| *d >= horizon)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn budget_odmitne_nad_bajty_i_kusy_a_doplni_se() { + let q = Budgets::new(3, 1000, Duration::from_secs(3600)); + let t0 = Instant::now(); + assert!(q.try_take_at(b"a", 600, t0)); + assert!(!q.try_take_at(b"a", 600, t0), "bajty došly"); + assert!(q.try_take_at(b"a", 100, t0)); + assert!(q.try_take_at(b"a", 100, t0)); + assert!(!q.try_take_at(b"a", 1, t0), "kusy došly"); + assert!(q.try_take_at(b"b", 600, t0), "jiný účet má vlastní kvótu"); + // za půl periody se doplní polovina + let t1 = t0 + Duration::from_secs(1800); + assert!(q.try_take_at(b"a", 400, t1)); + } + + #[test] + fn budget_nula_znamena_bez_omezeni() { + let q = Budgets::new(0, 0, Duration::from_secs(1)); + for _ in 0..1000 { + assert!(q.try_take(b"a", u64::MAX / 2)); + } + } + + #[test] + fn refund_vrati_nespotrebovane() { + let q = Budgets::new(0, 1000, Duration::from_secs(3600)); + assert!(q.try_take(b"a", 1000)); + q.refund(b"a", 700); + assert!(q.try_take(b"a", 700)); + } + + #[test] + fn ledger_drzi_strop_a_stare_dny_uvolni() { + let l = StoredLedger::new(1000, 30); + assert!(l.try_reserve_on(b"a", 800, 100)); + assert!(!l.try_reserve_on(b"a", 300, 100)); + assert!(l.try_reserve_on(b"b", 300, 100), "jiný účet"); + l.release(b"a", 500); + assert!(l.try_reserve_on(b"a", 600, 101)); + // po TTL dnech od nahrání je místo zase volné + assert!(l.try_reserve_on(b"a", 1000, 132)); + } +} diff --git a/rust/sepot-relay/src/rest.rs b/rust/sepot-relay/src/rest.rs index e257769..1c5e87a 100644 --- a/rust/sepot-relay/src/rest.rs +++ b/rust/sepot-relay/src/rest.rs @@ -132,6 +132,12 @@ pub async fn register( headers: HeaderMap, Json(body): Json, ) -> ApiResult<(StatusCode, Json)> { + if state.limits.registration_paused { + return Err(( + StatusCode::SERVICE_UNAVAILABLE, + "registrace jsou dočasně pozastavené, zkus to za pár minut".into(), + )); + } if body.one_time_prekeys.len() > MAX_ONE_TIME_PREKEYS { return Err(bad("příliš mnoho one-time prekeys")); } @@ -489,9 +495,6 @@ pub struct UploadKeysOut { pub one_time_prekeys: i64, } -/// Strop zásoby OTP na zařízení — nad něj se další zahodí (klient posílá -/// dávky po 50, relay zůstane u sta jako při registraci). -const MAX_STORED_ONE_TIME_PREKEYS: i64 = 100; /// Po kolika dnech se starý signed prekey smaže (po rotaci ho relay ještě /// drží, protože bundle s ním mohl někdo stáhnout těsně před výměnou). @@ -595,7 +598,9 @@ pub async fn upload_keys( .await .map_err(internal)?; for otp in &body.one_time_prekeys { - if stored >= MAX_STORED_ONE_TIME_PREKEYS { + // Strop zásoby OTP (`Limits::max_stored_otp`, výchozí 100) — nad něj se + // další zahodí (klient posílá dávky po 50, jako při registraci sto). + if stored >= state.limits.max_stored_otp { break; } let public = b64("one_time_prekeys.public_key", &otp.public_key, Some(33))?; diff --git a/rust/sepot-relay/src/state.rs b/rust/sepot-relay/src/state.rs index 0721ef2..c23c03f 100644 --- a/rust/sepot-relay/src/state.rs +++ b/rust/sepot-relay/src/state.rs @@ -15,11 +15,8 @@ use crate::net::TrustedProxies; /// TTL offline fronty. pub const QUEUE_TTL_DAYS: i64 = 30; -/// Strop celkové velikosti fronty na příjemce (SECURITY audit #3). Bez něj -/// autentizovaný útočník nasype neomezeně obálek (256 KiB × N) a při flushi -/// z toho vznikne OOM, na disku neomezený růst. Přetečení = FIFO zahození -/// nejstarších. 50 MiB ≈ 200 max obálek na příjemce. -const MAX_QUEUE_BYTES_PER_RECIPIENT: i64 = 50 * 1024 * 1024; +/// Největší zakódovaná obálka, kterou fronta přijme (CHECK v migraci 0001). +const MAX_STORED_ENVELOPE_BYTES: usize = 262_144; /// Velikost dávky při vylévání fronty (SECURITY audit #3). Strop paměti = /// FLUSH_BATCH × MAX_ENVELOPE_PAYLOAD (~16 MiB), ne celá fronta najednou. @@ -40,29 +37,61 @@ const MAX_DELIVERY_ATTEMPTS: i16 = 8; /// takže dva zmeškané cykly znamenají, že už nedýchá. const FRESH_CONNECTION_MS: u64 = 70_000; -/// Výchozí strop souběžných WS spojení z jedné IP. Env -/// `SEPOT_MAX_CONNS_PER_IP` ho přepisuje — za CGNAT mobilních operátorů -/// sdílí jednu IP stovky lidí. -const DEFAULT_MAX_CONNS_PER_IP: u32 = 32; - /// Běžný budíček na jeden cíl nejvýš jednou za tohle okno: třicet zpráv ve /// skupině během výpadku sítě = jeden push, ne třicet (APNs i FCM nápor /// tichých pushů škrtí, baterie taky). const WAKE_MIN_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); /// Budíček hovoru (VoIP push / FCM k=call) zvoní na telefonu hned a bez /// ověření — proto má vlastní, přísnější strop na cíl: nejvýš jeden za -/// CALL_WAKE_MIN_INTERVAL a CALL_WAKE_MAX_PER_WINDOW za CALL_WAKE_WINDOW. -/// Nad strop se obálka uloží a jde jen běžný budíček (phantom -/// hovory od kohokoli, kdo zná adresu). -const CALL_WAKE_MIN_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10); +/// `Limits::call_wake_min_interval` a `Limits::call_wakes_per_window` za +/// CALL_WAKE_WINDOW. Nad strop se obálka uloží a jde jen běžný budíček +/// (phantom hovory od kohokoli, kdo zná adresu). const CALL_WAKE_WINDOW: std::time::Duration = std::time::Duration::from_secs(600); -const CALL_WAKE_MAX_PER_WINDOW: usize = 6; -/// Kolik one-time prekeys smí relay vydat na jednu adresu za hodinu. -/// Noví kontakti jsou jednotky za den; sto GET /v1/keys za dvacet sekund -/// je vyčerpání zásoby cizím. Nad strop jde bundle bez OTP. -const OTP_ISSUE_MAX_PER_HOUR: u32 = 30; +/// Okno pro `Limits::otp_issue_per_hour`: noví kontakti jsou jednotky za den; +/// sto GET /v1/keys za dvacet sekund je vyčerpání zásoby cizím. Nad strop +/// jde bundle bez OTP. const OTP_ISSUE_WINDOW: std::time::Duration = std::time::Duration::from_secs(3600); +/// Okno kvóty odesílatele (`Limits::send_*_per_hour`). +const SEND_BUDGET_PERIOD: std::time::Duration = std::time::Duration::from_secs(3600); +/// Okno kvóty příloh na účet (`Limits::att_account_bytes_per_day`). +const ATTACHMENT_BUDGET_PERIOD: std::time::Duration = std::time::Duration::from_secs(86_400); + +/// Proč relay obálku nepřijal. Klient dostane Error rámec s kódem, spojení +/// zůstává otevřené. +#[derive(Debug)] +pub enum RouteError { + /// 400: formát (délky, prázdný payload, nad strop velikosti). + Invalid, + /// 404: adresa nepatří žádnému zařízení. Ukládat by ji nikdo nevyzvedl + /// — a fronta na náhodné adresy byla nejlevnější cesta, jak zaplnit disk. + /// Nic nového to neprozradí: GET /v1/keys vrací na neznámou adresu 404 taky. + UnknownRecipient, + /// 429: odesílající účet vyčerpal hodinovou kvótu. + SenderQuota, + /// 507: na disku zbývá méně než `Limits::disk_min_free_pct_queue`. + DiskFull, + Internal(anyhow::Error), +} + +impl RouteError { + pub fn code_and_message(&self) -> (u32, &'static str) { + match self { + Self::Invalid => (400, "obálka odmítnuta"), + Self::UnknownRecipient => (404, "příjemce neexistuje"), + Self::SenderQuota => (429, "příliš mnoho zpráv, zkus to později"), + Self::DiskFull => (507, "relay je dočasně plný, zkus to později"), + Self::Internal(_) => (500, "interní chyba"), + } + } +} + +impl From for RouteError { + fn from(e: sqlx::Error) -> Self { + Self::Internal(e.into()) + } +} + /// Značky budíčků jednoho cíle (jen v paměti, nikam se neukládají). #[derive(Default)] pub struct WakeMarks { @@ -157,8 +186,15 @@ pub struct AppState { pub apns: Option>, /// TURN pro hovory (None = relay údaje nevydává). pub turn: Option, - /// Strop WS spojení na IP. - pub max_conns_per_ip: u32, + /// Všechny limity relaye (env, viz limits.rs). + pub limits: crate::limits::Limits, + /// Kvóta ukládaných obálek na odesílající účet (paměť). + pub send_budget: crate::quota::Budgets, + /// Kvóta příloh na účet: tempo za den a uloženo celkem (paměť). + pub attachment_budget: crate::quota::Budgets, + pub attachment_stored: crate::quota::StoredLedger, + /// Volné místo na datovém filesystému (přílohy i DB). + pub disk: Arc, /// delivery_address → značky budíčků (throttling pushů a hovorů). pub wake_marks: DashMap, WakeMarks>, /// delivery_address → (začátek okna, vydaných OTP) — drain ochrana. @@ -174,6 +210,7 @@ impl AppState { certs: CertAuthority, attachments: crate::attachments::Config, trusted_proxies: Arc, + limits: crate::limits::Limits, ) -> Self { Self { pool, @@ -187,10 +224,22 @@ impl AppState { fcm: crate::fcm::FcmSender::from_env(), apns: crate::apns::ApnsSender::from_env(), turn: crate::turn::TurnConfig::from_env(), - max_conns_per_ip: std::env::var("SEPOT_MAX_CONNS_PER_IP") - .ok() - .and_then(|v| v.parse().ok()) - .unwrap_or(DEFAULT_MAX_CONNS_PER_IP), + send_budget: crate::quota::Budgets::new( + limits.send_per_hour, + limits.send_bytes_per_hour, + SEND_BUDGET_PERIOD, + ), + attachment_budget: crate::quota::Budgets::new( + 0, + limits.att_account_bytes_per_day, + ATTACHMENT_BUDGET_PERIOD, + ), + attachment_stored: crate::quota::StoredLedger::new( + limits.att_account_stored_bytes, + crate::attachments::ATTACHMENT_TTL_DAYS, + ), + disk: Arc::new(crate::disk::DiskWatch::unknown()), + limits, wake_marks: DashMap::new(), otp_issued: DashMap::new(), source_url: std::env::var("SEPOT_SOURCE_URL") @@ -212,6 +261,9 @@ impl AppState { }); self.otp_issued .retain(|_, (start, _)| now.duration_since(*start) < OTP_ISSUE_WINDOW); + self.send_budget.prune(); + self.attachment_budget.prune(); + self.attachment_stored.prune(); } /// Smí se vydat další OTP pro tuhle adresu? Počítá v hodinovém okně. @@ -224,7 +276,7 @@ impl AppState { if now.duration_since(entry.0) >= OTP_ISSUE_WINDOW { *entry = (now, 0); } - if entry.1 >= OTP_ISSUE_MAX_PER_HOUR { + if entry.1 >= self.limits.otp_issue_per_hour { return false; } entry.1 += 1; @@ -255,11 +307,11 @@ impl AppState { .call_wakes .retain(|t| now.duration_since(*t) < CALL_WAKE_WINDOW); if let Some(last) = marks.call_wakes.back() { - if now.duration_since(*last) < CALL_WAKE_MIN_INTERVAL { + if now.duration_since(*last) < self.limits.call_wake_min_interval { return false; } } - if marks.call_wakes.len() >= CALL_WAKE_MAX_PER_WINDOW { + if marks.call_wakes.len() >= self.limits.call_wakes_per_window { return false; } marks.call_wakes.push_back(now); @@ -277,19 +329,22 @@ impl AppState { Ok(n) } - /// Přijme obálku: validace → perzistentní fronta (dedup) → pokud je - /// příjemce online, doručí hned. At-least-once: záznam z fronty maže - /// až ACK příjemce, ne odeslání. + /// Přijme obálku: validace → existence příjemce → kvóta odesílatele a + /// místo na disku → perzistentní fronta (dedup) → pokud je příjemce + /// online, doručí hned. At-least-once: záznam z fronty maže až ACK + /// příjemce, ne odeslání. `sender_account` = ověřený účet spojení (jen + /// klíč kvóty v paměti; do fronty se neukládá — sealed sender). pub async fn route_envelope( &self, env: v1::Envelope, - ) -> anyhow::Result { + sender_account: &[u8], + ) -> Result { if env.dst_address.len() != 32 || env.msg_uuid.len() != 16 || env.payload.is_empty() || env.payload.len() > MAX_ENVELOPE_PAYLOAD { - anyhow::bail!("neplatná obálka"); + return Err(RouteError::Invalid); } if env.content_hint == v1::ContentHint::Ephemeral as i32 { @@ -297,7 +352,26 @@ impl AppState { } let dst = env.dst_address.clone(); + // Online příjemce existuje; jinak jeden indexovaný dotaz (UNIQUE). + if self.online.get(&dst).is_none() && !self.address_exists(&dst).await? { + return Err(RouteError::UnknownRecipient); + } + if self.disk.below(self.limits.disk_min_free_pct_queue) { + return Err(RouteError::DiskFull); + } let encoded = env.encode_to_vec(); + // Strop zakódované obálky = CHECK message_queue.envelope v 0001; payload + // těsně pod MAX_ENVELOPE_PAYLOAD ho s hlavičkou protobufu přesáhne a DB + // by zápis odmítla až po odečtení kvóty (dřív jako 400 přes chybu DB). + if encoded.len() > MAX_STORED_ENVELOPE_BYTES { + return Err(RouteError::Invalid); + } + if !self + .send_budget + .try_take(sender_account, encoded.len() as u64) + { + return Err(RouteError::SenderQuota); + } let expires = Utc::now() + Duration::days(QUEUE_TTL_DAYS); sqlx::query( "INSERT INTO message_queue (msg_uuid, dst_address, envelope, expires_at) @@ -311,7 +385,13 @@ impl AppState { .await?; // FIFO strop na příjemce (audit #3). Nově vložená má nejmladší // enqueued_at → sama se nikdy nezahodí, padne jen starý přebytek. - self.trim_queue(&dst, MAX_QUEUE_BYTES_PER_RECIPIENT).await?; + self.trim_queue( + &dst, + self.limits.queue_max_bytes_per_recipient, + self.limits.queue_max_per_recipient, + ) + .await + .map_err(RouteError::Internal)?; let is_call = env.content_hint == v1::ContentHint::Call as i32; if let Some(dev) = self.online.get(&dst) { @@ -341,6 +421,13 @@ impl AppState { Ok(v1::receipt::Status::StoredOffline) } + async fn address_exists(&self, dst: &[u8]) -> Result { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM devices WHERE delivery_address = $1)") + .bind(dst) + .fetch_one(&self.pool) + .await + } + /// Jen živé doručení („píše…“): do fronty nic, push nic. Zatuchlé /// spojení se nepočítá — obálka by v mrtvém kanálu zmizela stejně /// a odesílatel se má dozvědět, že příjemce nikdo neposlouchá. @@ -362,23 +449,27 @@ impl AppState { } /// FIFO strop fronty příjemce: nech nejnovější obálky do `cap_bytes` - /// (běžný součet od nejnovější), starší přebytek zahoď. Vrací počet - /// zahozených. Oddělené kvůli přímému testu bez vkládání 50 MiB. - async fn trim_queue(&self, dst: &[u8], cap_bytes: i64) -> anyhow::Result { + /// (běžný součet od nejnovější) a nejvýš `cap_count` kusů, starší + /// přebytek zahoď. Strop počtu chrání před miliony drobných obálek, kde + /// řádek v DB stojí víc než payload. Vrací počet zahozených. Oddělené + /// kvůli přímému testu bez vkládání 50 MiB. + async fn trim_queue(&self, dst: &[u8], cap_bytes: i64, cap_count: i64) -> anyhow::Result { let res = sqlx::query( "DELETE FROM message_queue WHERE dst_address = $1 AND msg_uuid IN ( SELECT msg_uuid FROM ( SELECT msg_uuid, - sum(octet_length(envelope)) OVER ( - ORDER BY enqueued_at DESC, msg_uuid DESC - ROWS UNBOUNDED PRECEDING) AS running + sum(octet_length(envelope)) OVER newest AS running, + row_number() OVER newest AS position FROM message_queue WHERE dst_address = $1 - ) s WHERE running > $2 + WINDOW newest AS (ORDER BY enqueued_at DESC, msg_uuid DESC + ROWS UNBOUNDED PRECEDING) + ) s WHERE running > $2 OR position > $3 )", ) .bind(dst) .bind(cap_bytes) + .bind(cap_count) .execute(&self.pool) .await?; Ok(res.rows_affected()) @@ -475,7 +566,11 @@ impl AppState { endpoint: Option<&[u8]>, voip_token: Option<&[u8]>, ) -> anyhow::Result<()> { - if platform == crate::push::PLATFORM_NONE || endpoint.is_none() { + // APNs s jediným VoIP tokenem (appka hovory zvoní přes PushKit, běžný + // APNs token ještě nemá) se ukládá taky — dřív by ji tahle větev + // smazala a hovor by offline iPhone nikdy nevzbudil. + let voip_only = platform == crate::push::PLATFORM_APNS && voip_token.is_some(); + if platform == crate::push::PLATFORM_NONE || (endpoint.is_none() && !voip_only) { sqlx::query("DELETE FROM push_tokens WHERE device_uuid = $1") .bind(device_uuid) .execute(&self.pool) @@ -693,6 +788,9 @@ mod db_tests { use super::*; use rand::RngCore; + /// Účet odesílatele v testech — kvóta je jen v paměti, DB ho nezná. + const SENDER: &[u8] = b"test-odesilatel"; + async fn test_state() -> AppState { let url = std::env::var("DATABASE_URL") .unwrap_or_else(|_| "postgres://sepot:sepot_dev@127.0.0.1/sepot_relay".into()); @@ -703,7 +801,13 @@ mod db_tests { .expect("dev DB"); let certs = crate::certs::CertAuthority::init(&pool).await.unwrap(); let trusted = Arc::new(crate::net::TrustedProxies::from_env()); - AppState::new(pool, certs, crate::attachments::Config::from_env(), trusted) + AppState::new( + pool, + certs, + crate::attachments::Config::from_env(), + trusted, + crate::limits::Limits::from_env(), + ) } fn rand_bytes(n: usize) -> Vec { @@ -826,7 +930,7 @@ mod db_tests { content_hint: v1::ContentHint::Call as i32, payload: vec![0xCD; 64], }; - let status = state.route_envelope(env).await.unwrap(); + let status = state.route_envelope(env, SENDER).await.unwrap(); assert_eq!(status, v1::receipt::Status::StoredOffline); let queued: i64 = sqlx::query_scalar("SELECT count(*) FROM message_queue WHERE dst_address = $1") @@ -875,7 +979,7 @@ mod db_tests { content_hint: 0, payload: vec![0xAB; 64], }; - let status = state.route_envelope(env).await.unwrap(); + let status = state.route_envelope(env, SENDER).await.unwrap(); assert_eq!(status, v1::receipt::Status::StoredOffline); // obálka je ve frontě (příjemce byl offline) @@ -902,7 +1006,7 @@ mod db_tests { content_hint: v1::ContentHint::Ephemeral as i32, payload: vec![0xAB; 64], }; - let status = state.route_envelope(env).await.unwrap(); + let status = state.route_envelope(env, SENDER).await.unwrap(); assert_eq!(status, v1::receipt::Status::Dropped); // „píše…“ pro offline příjemce nesmí skončit ve frontě @@ -1055,7 +1159,7 @@ mod db_tests { payload: vec![0xCD; 64], }; assert_eq!( - state.route_envelope(env).await.unwrap(), + state.route_envelope(env, SENDER).await.unwrap(), v1::receipt::Status::StoredOffline, "zatuchlé spojení se nesmí hlásit jako doručené online" ); @@ -1069,7 +1173,7 @@ mod db_tests { payload: vec![0xCE; 64], }; assert_eq!( - state.route_envelope(env2).await.unwrap(), + state.route_envelope(env2, SENDER).await.unwrap(), v1::receipt::Status::DeliveredOnline, "živé spojení má být online" ); @@ -1109,7 +1213,7 @@ mod db_tests { payload: vec![0xEF; 64], }; assert_eq!( - state.route_envelope(env).await.unwrap(), + state.route_envelope(env, SENDER).await.unwrap(), v1::receipt::Status::DeliveredOnline, "konec zombie spojení nesmí vyřadit nové spojení" ); @@ -1237,7 +1341,7 @@ mod db_tests { .unwrap(); } - let dropped = state.trim_queue(&dst, 250).await.unwrap(); + let dropped = state.trim_queue(&dst, 250, 1000).await.unwrap(); assert_eq!(dropped, 3); let remaining: Vec> = sqlx::query_scalar( @@ -1255,4 +1359,140 @@ mod db_tests { .await .unwrap(); } + + fn envelope(dst: &[u8], payload_len: usize) -> v1::Envelope { + v1::Envelope { + dst_address: dst.to_vec(), + msg_uuid: rand_bytes(16), + content_hint: 0, + payload: vec![0x5Au8; payload_len], + } + } + + async fn queued(state: &AppState, dst: &[u8]) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM message_queue WHERE dst_address = $1") + .bind(dst) + .fetch_one(&state.pool) + .await + .unwrap() + } + + /// Obálka na adresu bez zařízení se neuloží (dřív 30 dní ve frontě = + /// nejlevnější cesta, jak zaplnit disk relaye). + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn obalka_na_neexistujici_adresu_se_neulozi() { + let state = test_state().await; + let dst = rand_bytes(32); + let err = state.route_envelope(envelope(&dst, 64), SENDER).await.unwrap_err(); + assert!(matches!(err, RouteError::UnknownRecipient), "{err:?}"); + assert_eq!(err.code_and_message().0, 404); + assert_eq!(queued(&state, &dst).await, 0); + } + + /// Kvóta odesílatele: nad hodinový strop 429, jiný účet jede dál. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn kvota_odesilatele_odmitne_nad_strop() { + let mut state = test_state().await; + state.send_budget = crate::quota::Budgets::new(3, 0, SEND_BUDGET_PERIOD); + let (account_id, _device, dst) = seed_device(&state).await; + let sender = rand_bytes(16); + for _ in 0..3 { + state.route_envelope(envelope(&dst, 64), &sender).await.unwrap(); + } + let err = state.route_envelope(envelope(&dst, 64), &sender).await.unwrap_err(); + assert!(matches!(err, RouteError::SenderQuota), "{err:?}"); + state + .route_envelope(envelope(&dst, 64), &rand_bytes(16)) + .await + .expect("jiný odesílatel má vlastní kvótu"); + assert_eq!(queued(&state, &dst).await, 4); + cleanup(&state, &account_id, &dst).await; + } + + /// Málo místa na disku: offline fronta stojí (507), nic se nezapíše. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn plny_disk_zastavi_frontu() { + let state = test_state().await; + let (account_id, _device, dst) = seed_device(&state).await; + state.disk.set_free_pct(state.limits.disk_min_free_pct_queue - 1.0); + let err = state.route_envelope(envelope(&dst, 64), SENDER).await.unwrap_err(); + assert!(matches!(err, RouteError::DiskFull), "{err:?}"); + assert_eq!(queued(&state, &dst).await, 0); + cleanup(&state, &account_id, &dst).await; + } + + /// Strop počtu obálek na příjemce drží nejnovější, i když bajty stačí. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn trim_queue_strop_poctu() { + let state = test_state().await; + let dst = rand_bytes(32); + let mut uuids = Vec::new(); + for i in 0..5 { + let msg_uuid = rand_bytes(16); + uuids.push(msg_uuid.clone()); + sqlx::query( + "INSERT INTO message_queue (msg_uuid, dst_address, envelope, enqueued_at, expires_at) + VALUES ($1, $2, $3, now() + (interval '1 second' * $4), now() + interval '30 days')", + ) + .bind(&msg_uuid) + .bind(&dst) + .bind(vec![0u8; 10]) + .bind(i as i32) + .execute(&state.pool) + .await + .unwrap(); + } + assert_eq!(state.trim_queue(&dst, 1 << 30, 2).await.unwrap(), 3); + let remaining: Vec> = sqlx::query_scalar( + "SELECT msg_uuid FROM message_queue WHERE dst_address = $1 ORDER BY enqueued_at", + ) + .bind(&dst) + .fetch_all(&state.pool) + .await + .unwrap(); + assert_eq!(remaining, vec![uuids[3].clone(), uuids[4].clone()]); + sqlx::query("DELETE FROM message_queue WHERE dst_address = $1") + .bind(&dst) + .execute(&state.pool) + .await + .unwrap(); + } + + /// iPhone s jediným VoIP tokenem (běžný APNs token ještě nemá) musí jít + /// vzbudit hovorem — registrace se nesmí zahodit. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn apns_jen_s_voip_tokenem_se_ulozi() { + let state = test_state().await; + let (account_id, device_uuid, dst) = seed_device(&state).await; + state + .register_push(&device_uuid, crate::push::PLATFORM_APNS, None, Some(b"voip")) + .await + .unwrap(); + let target = state.push_target_full(&dst).await.expect("VoIP cíl"); + assert_eq!(target.voip_token.as_deref(), Some(&b"voip"[..])); + assert!(target.token.is_none()); + assert!(state.push_target(&dst).await.is_none(), "běžný budíček nemá kam"); + cleanup(&state, &account_id, &dst).await; + } + + /// Obálka, která se zakódovaná nevejde do CHECK fronty, je 400 dřív, než + /// ukrojí z kvóty odesílatele — ne 500 z databáze. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn obalka_nad_strop_fronty_je_neplatna() { + let state = test_state().await; + let (account_id, _device, dst) = seed_device(&state).await; + let err = state + .route_envelope(envelope(&dst, MAX_ENVELOPE_PAYLOAD), SENDER) + .await + .unwrap_err(); + assert!(matches!(err, RouteError::Invalid), "{err:?}"); + assert_eq!(queued(&state, &dst).await, 0); + cleanup(&state, &account_id, &dst).await; + } } diff --git a/rust/sepot-relay/src/ws.rs b/rust/sepot-relay/src/ws.rs index c554d89..319ed16 100644 --- a/rust/sepot-relay/src/ws.rs +++ b/rust/sepot-relay/src/ws.rs @@ -33,13 +33,20 @@ use crate::state::AppState; const AUTH_DEADLINE: Duration = Duration::from_secs(10); /// Kolik AUTH pokusů smí jedno spojení udělat (každý = dotaz do DB). const MAX_AUTH_ATTEMPTS: u32 = 3; -/// Rate limit rámců od klienta: max FRAME_WINDOW_MAX za FRAME_WINDOW. +/// Rate limit rámců od klienta: max `Limits::ws_frames_per_window` za FRAME_WINDOW. /// Počítají se jen rámce, které něco stojí (obálka, push registrace, TURN); /// ACK a Pong ne — klient po vylití velké fronty posílá ACK + doručenku na /// každou obálku a 429 by ho odpojilo uprostřed, fronta by se po osmi /// kolech smazala (ztráta zpráv bez útočníka). const FRAME_WINDOW: Duration = Duration::from_secs(10); -const FRAME_WINDOW_MAX: u32 = 200; +/// Buffery jednoho spojení. Výchozí v axum (tungstenite) jsou 128 KiB čtecí +/// + 128 KiB zapisovací = ~137 KiB RAM na spojení; při 512 MiB stropu +/// kontejneru OOM kolem 3,7–7 tisíc připojených (změřeno). 4 KiB čtecí buffer +/// a zápis bez bufferu (každý rámec se hned flushne) = ~16,6 KiB na spojení, +/// 8 990 spojení = 149 MiB. Velké obálky (256 KiB) tím netrpí: čtecí buffer +/// je jen blok, zprávu skládá tungstenite zvlášť do `max_message_size`. +const WS_READ_BUFFER: usize = 4096; +const WS_WRITE_BUFFER: usize = 0; /// Push endpoint (URL distributora / FCM token) a VoIP token mají rozumný /// strop; APNs token je 64 hex, FCM do ~200 znaků, UnifiedPush URL < 1 KiB. const MAX_PUSH_ENDPOINT_BYTES: usize = 2048; @@ -83,10 +90,16 @@ pub async fn ws_handler( ws: WebSocketUpgrade, ) -> impl IntoResponse { let ip = state.trusted_proxies.client_ip(peer.ip(), &headers); + // Všichni onion klienti mají IP kontejneru Toru → vlastní, vyšší strop. + let cap = if state.trusted_proxies.is_tor(peer.ip()) { + state.limits.max_conns_tor + } else { + state.limits.max_conns_per_ip + }; let over_cap = { let mut entry = state.ws_conns.entry(ip).or_insert(0); - if *entry >= state.max_conns_per_ip { + if *entry >= cap { true } else { *entry += 1; @@ -103,7 +116,9 @@ pub async fn ws_handler( ip, }; - ws.max_message_size(MAX_ENVELOPE_PAYLOAD + 4096) + ws.read_buffer_size(WS_READ_BUFFER) + .write_buffer_size(WS_WRITE_BUFFER) + .max_message_size(MAX_ENVELOPE_PAYLOAD + 4096) .max_frame_size(MAX_ENVELOPE_PAYLOAD + 4096) .on_upgrade(move |socket| async move { let _guard = guard; @@ -119,6 +134,7 @@ struct Pending { delivery_address: Vec, identity_key: Vec, device_uuid: Vec, + account_id: Vec, } fn server_frame(frame: v1::server_frame::Frame) -> v1::ServerFrame { @@ -166,6 +182,8 @@ async fn handle(state: Arc, socket: WebSocket) { let mut authed_address: Option> = None; // device_uuid ověřeného zařízení — klíč pro push registraci. let mut authed_device: Option> = None; + // Účet ověřeného zařízení — klíč kvóty odesílatele (jen v paměti). + let mut authed_account: Option> = None; // Záznam v `online` mapě — značkuje se každým rámcem od klienta, aby // routing poznal živé spojení od zatuchlého. let mut online_device: Option> = None; @@ -233,7 +251,7 @@ async fn handle(state: Arc, socket: WebSocket) { window_frames = 0; } window_frames += 1; - if window_frames > FRAME_WINDOW_MAX { + if window_frames > state.limits.ws_frames_per_window { let _ = tx.send(error_frame(429, "příliš mnoho rámců")).await; break; } @@ -250,8 +268,8 @@ async fn handle(state: Arc, socket: WebSocket) { let _ = tx.send(error_frame(400, "neplatný AUTH")).await; break; } - let row: Option<(Vec, Vec)> = sqlx::query_as( - "SELECT d.delivery_address, i.identity_key + let row: Option<(Vec, Vec, Vec)> = sqlx::query_as( + "SELECT d.delivery_address, i.identity_key, d.account_id FROM devices d JOIN identities i USING (account_id) WHERE d.device_uuid = $1", ) @@ -260,7 +278,7 @@ async fn handle(state: Arc, socket: WebSocket) { .await .ok() .flatten(); - let Some((delivery_address, identity_key)) = row else { + let Some((delivery_address, identity_key, account_id)) = row else { let _ = tx.send(error_frame(401, "neautorizováno")).await; break; }; @@ -279,6 +297,7 @@ async fn handle(state: Arc, socket: WebSocket) { delivery_address, identity_key, device_uuid: auth.device_uuid, + account_id, }); } @@ -347,6 +366,7 @@ async fn handle(state: Arc, socket: WebSocket) { } authed_device = Some(pd.device_uuid); + authed_account = Some(pd.account_id); } Some(v1::client_frame::Frame::RegisterPush(reg)) if authed_address.is_some() => { @@ -377,7 +397,8 @@ async fn handle(state: Arc, socket: WebSocket) { Some(v1::client_frame::Frame::Envelope(envelope)) if authed_address.is_some() => { let msg_uuid = envelope.msg_uuid.clone(); - match state.route_envelope(envelope).await { + let sender = authed_account.as_deref().unwrap_or_default(); + match state.route_envelope(envelope, sender).await { Ok(status) => { let _ = tx .send(server_frame(v1::server_frame::Frame::Receipt( @@ -388,8 +409,12 @@ async fn handle(state: Arc, socket: WebSocket) { ))) .await; } - Err(_) => { - let _ = tx.send(error_frame(400, "obálka odmítnuta")).await; + Err(e) => { + if let crate::state::RouteError::Internal(err) = &e { + tracing::warn!("obálka neuložena: {err}"); + } + let (code, message) = e.code_and_message(); + let _ = tx.send(error_frame(code, message)).await; } } }