From b82f50f888f6bd370b0956b41d5660f3b816d5a2 Mon Sep 17 00:00:00 2001 From: Julius Joska Date: Thu, 13 Aug 2026 12:23:17 +0200 Subject: [PATCH] fix(relay): keep-alive on WebSocket, cap delivery attempts, honest receipts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Messages arrived late or not at all, for two independent reasons. Dead connections: neither side sent a ping, and after AUTH the server waited for a frame without a timeout. Mobile NAT drops idle TCP silently, so the client wrote into a void while the relay still held the device in its online map — and therefore never sent a push wake-up either. The ping is a protocol-level one rather than v1::Ping, because dart:io answers those on its own; the fix works with clients already in the wild. Poisoned queue: a client decrypts an envelope (ratchet advances), stores it and acknowledges it. If that ACK is lost, a second decryption attempt can only fail, so the envelope is never acknowledged — and was redelivered on every connect until its 30-day TTL, without ever reaching the recipient. attempt_count existed from the first migration but was never counted; it now bounds redelivery. Receipts reported DELIVERED_ONLINE whenever a write into the bounded mpsc channel succeeded, which it does even for a long-dead socket. Online is now claimed only for devices that recently sent something. Push send failures were logged at debug while the relay runs at info, so a rejected wake-up looked exactly like a working one. --- rust/sepot-relay/src/apns.rs | 6 +- rust/sepot-relay/src/fcm.rs | 8 +- rust/sepot-relay/src/push.rs | 14 +- rust/sepot-relay/src/state.rs | 479 ++++++++++++++++++++++++++++++++-- rust/sepot-relay/src/ws.rs | 99 ++++--- 5 files changed, 545 insertions(+), 61 deletions(-) diff --git a/rust/sepot-relay/src/apns.rs b/rust/sepot-relay/src/apns.rs index 2f2c870..413363c 100644 --- a/rust/sepot-relay/src/apns.rs +++ b/rust/sepot-relay/src/apns.rs @@ -123,11 +123,13 @@ impl ApnsSender { .body(r#"{"aps":{"content-available":1}}"#) .send() .await; + // warn, ne debug — viz shodný komentář ve fcm.rs: na RUST_LOG=info + // bylo selhání pushe neviditelné. match resp { Ok(r) if !r.status().is_success() => { - tracing::debug!(status = %r.status(), "APNs budíček odmítnut"); + tracing::warn!(status = %r.status(), "APNs budíček odmítnut"); } - Err(e) => tracing::debug!("APNs budíček selhal: {e}"), + Err(e) => tracing::warn!("APNs budíček selhal: {e}"), _ => {} } } diff --git a/rust/sepot-relay/src/fcm.rs b/rust/sepot-relay/src/fcm.rs index 8595bc3..b2ab658 100644 --- a/rust/sepot-relay/src/fcm.rs +++ b/rust/sepot-relay/src/fcm.rs @@ -143,11 +143,15 @@ impl FcmSender { "android": { "priority": "HIGH" } } }); + // warn, ne debug: relay běží na RUST_LOG=info, takže na debugu bylo + // selhání pushe neviditelné a tiše se hromadila fronta. Protistranou + // je Google, ne telefon uživatele — chyba tu znamená mrtvý token nebo + // rozbitou konfiguraci projektu, obojí je provozní problém. match http.post(&url).bearer_auth(access).json(&body).send().await { Ok(resp) if !resp.status().is_success() => { - tracing::debug!(status = %resp.status(), "FCM budíček odmítnut"); + tracing::warn!(status = %resp.status(), "FCM budíček odmítnut"); } - Err(e) => tracing::debug!("FCM budíček selhal: {e}"), + Err(e) => tracing::warn!("FCM budíček selhal: {e}"), _ => {} } } diff --git a/rust/sepot-relay/src/push.rs b/rust/sepot-relay/src/push.rs index c5efc20..d3cfe90 100644 --- a/rust/sepot-relay/src/push.rs +++ b/rust/sepot-relay/src/push.rs @@ -85,13 +85,23 @@ pub async fn send_wake( /// Samotný POST budíčku (bez guardu — ten řeší volající). Oddělené kvůli /// testovatelnosti proti loopback mocku (guard by loopback jinak odmítl). async fn post_wake(client: &Client, url: &str) { - let _ = client + let resp = client .post(url) .header("content-type", "application/json") .header("ttl", "2419200") // WebPush TTL (28 dní), ntfy ignoruje .body(WAKE_BODY) .send() .await; + // Výsledek se dřív zahazoval (`let _ =`), takže nešlo poznat, jestli + // budíčky vůbec odcházejí. Logujeme jen stavový kód — endpoint je + // identifikátor uživatele a do logu nesmí. + match resp { + Ok(r) if !r.status().is_success() => { + tracing::warn!(status = %r.status(), "UnifiedPush budíček odmítnut"); + } + Err(e) => tracing::warn!("UnifiedPush budíček selhal: {e}"), + _ => {} + } } /// Povolit jen bezpečné cíle. https vždy; http jen pro .onion (samo šifruje). @@ -191,7 +201,7 @@ mod tests { #[test] fn verejne_adresy_projdou() { - for ip in ["1.1.1.1", "8.8.8.8", "203.0.113.10", "2606:4700:4700::1111"] { + for ip in ["1.1.1.1", "8.8.8.8", "9.9.9.9", "2606:4700:4700::1111"] { assert!(!is_internal_ip(&ip.parse().unwrap()), "{ip} nemělo být interní"); } } diff --git a/rust/sepot-relay/src/state.rs b/rust/sepot-relay/src/state.rs index a1da61d..40a7fca 100644 --- a/rust/sepot-relay/src/state.rs +++ b/rust/sepot-relay/src/state.rs @@ -1,4 +1,6 @@ -use chrono::{Duration, Utc}; +use std::sync::Arc; + +use chrono::{DateTime, Duration, Utc}; use dashmap::DashMap; use prost::Message; use sepot_proto::v1; @@ -7,18 +9,80 @@ use sqlx::PgPool; use tokio::sync::mpsc; use crate::certs::CertAuthority; +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; + +/// 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. +const FLUSH_BATCH: i64 = 64; + +/// Po kolika marných pokusech o doručení obálku zahodit. +/// +/// Ztracený ACK je jednosměrka: klient obálku dešifruje (ratchet postoupí), +/// uloží a pošle ACK — když ten cestou zmizí (mrtvé spojení), relay obálku +/// drží dál, jenže druhý pokus o dešifrování už libsignal odmítne. Klient ji +/// tedy neACKuje a fronta se otravuje: tatáž obálka se posílá při každém +/// připojení až do TTL 30 dní a příjemce ji nikdy nedostane. Strop tomu dá +/// konec. Osm pokusů je velkoryse nad běžné „appka spadla před ACK". +const MAX_DELIVERY_ATTEMPTS: i16 = 8; + +/// Jak dlouho po posledním rámci od klienta se spojení ještě považuje za +/// prokazatelně živé. Klient odpovídá na ping à 30 s (ws::PING_INTERVAL), +/// takže dva zmeškané cykly znamenají, že už nedýchá. +const FRESH_CONNECTION_MS: u64 = 70_000; + +/// Připojené zařízení: kanál na jeho WS + kdy od něj naposled něco přišlo. +/// +/// Samotný `tx` o životnosti spojení nic neříká — je to mpsc s bufferem, do +/// kterého zápis projde i u dávno mrtvého socketu. Právě proto se dřív vracelo +/// falešné DELIVERED_ONLINE a push budíček se vůbec nezkusil. +pub struct OnlineDevice { + pub tx: mpsc::Sender, + /// ms od startu procesu; monotónní, aby s ním neházel NTP. + last_frame_ms: std::sync::atomic::AtomicU64, +} + +impl OnlineDevice { + pub fn new(tx: mpsc::Sender, now_ms: u64) -> Self { + Self { + tx, + last_frame_ms: std::sync::atomic::AtomicU64::new(now_ms), + } + } + + /// Přišel rámec od klienta (Pong/ACK/Envelope) — spojení dýchá. + pub fn touch(&self, now_ms: u64) { + self.last_frame_ms + .store(now_ms, std::sync::atomic::Ordering::Relaxed); + } + + fn is_fresh(&self, now_ms: u64) -> bool { + let last = self.last_frame_ms.load(std::sync::atomic::Ordering::Relaxed); + now_ms.saturating_sub(last) <= FRESH_CONNECTION_MS + } +} + pub struct AppState { pub pool: PgPool, - /// delivery_address → kanál na živé WS spojení zařízení. - pub online: DashMap, mpsc::Sender>, + /// Kdy proces nastartoval — základ monotónních značek v [`OnlineDevice`]. + pub started: std::time::Instant, + /// delivery_address → živé WS spojení zařízení. + pub online: DashMap, Arc>, /// klientská IP → počet souběžných WS spojení (anti-flood cap). /// Jen v paměti, nikam se neloguje ani neukládá. pub ws_conns: DashMap, pub certs: CertAuthority, + /// Trusted proxy sada pro určení klientské IP (WS conn-cap). Sdílená + /// s governor limity — komu se věří forwarded hlavička. + pub trusted_proxies: Arc, /// HTTP klient na push budíčky (bez redirectů, krátký timeout). pub http: reqwest::Client, /// Úložiště šifrovaných příloh (adresář + strop velikosti). @@ -30,12 +94,19 @@ pub struct AppState { } impl AppState { - pub fn new(pool: PgPool, certs: CertAuthority, attachments: crate::attachments::Config) -> Self { + pub fn new( + pool: PgPool, + certs: CertAuthority, + attachments: crate::attachments::Config, + trusted_proxies: Arc, + ) -> Self { Self { pool, + started: std::time::Instant::now(), online: DashMap::new(), ws_conns: DashMap::new(), certs, + trusted_proxies, http: crate::push::build_client(), attachments, fcm: crate::fcm::FcmSender::from_env(), @@ -71,20 +142,57 @@ impl AppState { .bind(expires) .execute(&self.pool) .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?; - if let Some(tx) = self.online.get(&dst) { + if let Some(dev) = self.online.get(&dst).map(|d| d.clone()) { let deliver = v1::ServerFrame { frame: Some(v1::server_frame::Frame::Deliver(env)), }; - if tx.send(deliver).await.is_ok() { + let fresh = dev.is_fresh(self.now_ms()); + let queued = dev.tx.send(deliver).await.is_ok(); + // ONLINE hlásíme jen u spojení, které prokazatelně dýchá. Do + // zatuchlého se sice pošle taky (kdyby to byl jen tichý klient), + // ale propadne se i na push — jinak by mrtvé spojení budíček + // spolklo a příjemce se o zprávě nedozvěděl. + if fresh && queued { return Ok(v1::receipt::Status::DeliveredOnline); } } - // Offline (nebo online kanál selhal) → zkusit push budíček. + // Offline, zatuchlé nebo selhalý kanál → zkusit push budíček. self.wake(&dst).await; Ok(v1::receipt::Status::StoredOffline) } + /// 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 { + 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 + FROM message_queue WHERE dst_address = $1 + ) s WHERE running > $2 + )", + ) + .bind(dst) + .bind(cap_bytes) + .execute(&self.pool) + .await?; + Ok(res.rows_affected()) + } + + /// Monotónní značka pro [`OnlineDevice`] — ms od startu procesu. + pub fn now_ms(&self) -> u64 { + self.started.elapsed().as_millis() as u64 + } + /// Budíček pro offline příjemce: dohledá jeho push token a pošle prázdný /// „ťuk". Best-effort, na pozadí — neblokuje receipt odesílateli. /// Endpoint se NELOGUJE. @@ -172,32 +280,118 @@ impl AppState { Ok(()) } - /// Po AUTH: vyleje nedoručenou frontu zařízení (DELIVER stream). + /// Po AUTH: vyleje nedoručenou frontu zařízení (DELIVER stream). Čte po + /// dávkách keyset kurzorem přes (enqueued_at, msg_uuid) — paměť je + /// omezená na FLUSH_BATCH obálek, ne celá fronta najednou (audit #3). + /// Index idx_queue_dst(dst_address, enqueued_at) kurzor pokrývá. pub async fn flush_queue( &self, delivery_address: &[u8], tx: &mpsc::Sender, ) -> anyhow::Result { - let rows: Vec<(Vec,)> = sqlx::query_as( - "SELECT envelope FROM message_queue - WHERE dst_address = $1 AND expires_at > now() - ORDER BY enqueued_at", + // Obálky, které strop pokusů vyčerpaly, jsou neprůchodné (skoro vždy + // ztracený ACK po úspěšném dešifrování) — zahodit dřív, než se znovu + // pošlou. Maže se až tady, ne hned po inkrementu: poslední pokus + // dostane šanci doběhnout do ACK. + let dropped = sqlx::query( + "DELETE FROM message_queue WHERE dst_address = $1 AND attempt_count >= $2", ) .bind(delivery_address) - .fetch_all(&self.pool) - .await?; + .bind(MAX_DELIVERY_ATTEMPTS) + .execute(&self.pool) + .await? + .rows_affected(); + if dropped > 0 { + // Bez identifikátoru příjemce — jen že se fronta čistila. + tracing::info!(dropped, "nedoručitelné obálky zahozeny (strop pokusů)"); + } - let count = rows.len(); - for (encoded,) in rows { - let env = v1::Envelope::decode(encoded.as_slice())?; - let frame = v1::ServerFrame { - frame: Some(v1::server_frame::Frame::Deliver(env)), + let mut total = 0usize; + // (enqueued_at, msg_uuid) poslední odeslané obálky = kurzor další dávky. + let mut cursor: Option<(DateTime, Vec)> = None; + loop { + let rows: Vec<(Vec, DateTime, Vec)> = match &cursor { + None => { + sqlx::query_as( + "SELECT envelope, enqueued_at, msg_uuid FROM message_queue + WHERE dst_address = $1 AND expires_at > now() + AND attempt_count < $2 + ORDER BY enqueued_at, msg_uuid + LIMIT $3", + ) + .bind(delivery_address) + .bind(MAX_DELIVERY_ATTEMPTS) + .bind(FLUSH_BATCH) + .fetch_all(&self.pool) + .await? + } + Some((ts, uuid)) => { + sqlx::query_as( + "SELECT envelope, enqueued_at, msg_uuid FROM message_queue + WHERE dst_address = $1 AND expires_at > now() + AND attempt_count < $2 + AND (enqueued_at, msg_uuid) > ($3, $4) + ORDER BY enqueued_at, msg_uuid + LIMIT $5", + ) + .bind(delivery_address) + .bind(MAX_DELIVERY_ATTEMPTS) + .bind(ts) + .bind(uuid) + .bind(FLUSH_BATCH) + .fetch_all(&self.pool) + .await? + } }; - if tx.send(frame).await.is_err() { + let batch_len = rows.len(); + if batch_len == 0 { break; } + let last = rows + .last() + .map(|(_, ts, uuid)| (*ts, uuid.clone())) + .expect("dávka není prázdná"); + let mut sent_uuids: Vec> = Vec::with_capacity(batch_len); + for (encoded, _, uuid) in &rows { + let env = v1::Envelope::decode(encoded.as_slice())?; + let frame = v1::ServerFrame { + frame: Some(v1::server_frame::Frame::Deliver(env)), + }; + if tx.send(frame).await.is_err() { + // Příjemce odpadl uprostřed dávky — započítat jen to, co + // reálně odešlo, ať pokus „zdarma" nekrátí strop. + self.bump_attempts(&sent_uuids).await; + return Ok(total); + } + sent_uuids.push(uuid.clone()); + total += 1; + } + self.bump_attempts(&sent_uuids).await; + if (batch_len as i64) < FLUSH_BATCH { + break; // poslední (neúplná) dávka + } + cursor = Some(last); + } + Ok(total) + } + + /// Započítá pokus o doručení odeslaným obálkám. Selhání se jen zaloguje — + /// obálky už jsou na drátě a shodit kvůli tomu flush by bylo horší než + /// nepřesné počítadlo. + async fn bump_attempts(&self, msg_uuids: &[Vec]) { + if msg_uuids.is_empty() { + return; + } + if let Err(e) = sqlx::query( + "UPDATE message_queue SET attempt_count = attempt_count + 1 + WHERE msg_uuid = ANY($1)", + ) + .bind(msg_uuids) + .execute(&self.pool) + .await + { + tracing::warn!("attempt_count se nepodařilo zvýšit: {e}"); } - Ok(count) } /// ACK příjemce → hard-delete z fronty (žádná retence po doručení). @@ -228,7 +422,8 @@ mod db_tests { .await .expect("dev DB"); let certs = crate::certs::CertAuthority::init(&pool).await.unwrap(); - AppState::new(pool, certs, crate::attachments::Config::from_env()) + let trusted = Arc::new(crate::net::TrustedProxies::from_env()); + AppState::new(pool, certs, crate::attachments::Config::from_env(), trusted) } fn rand_bytes(n: usize) -> Vec { @@ -413,4 +608,244 @@ mod db_tests { // idempotence: druhé smazání neexistujícího účtu nesmí spadnout state.delete_identity(&account_id).await.unwrap(); } + + /// Audit #3: flush přes hranici dávky (150 > FLUSH_BATCH) nesmí keyset + /// kurzorem žádnou obálku ztratit ani zdvojit. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn flush_queue_davkuje_pres_hranici_bez_ztraty() { + let state = test_state().await; + let dst = rand_bytes(32); + let n = 150usize; + let mut expected = std::collections::HashSet::new(); + for i in 0..n { + let msg_uuid = rand_bytes(16); + expected.insert(msg_uuid.clone()); + let env = v1::Envelope { + dst_address: dst.clone(), + msg_uuid: msg_uuid.clone(), + content_hint: 0, + payload: vec![0x11; 32], + }; + sqlx::query( + "INSERT INTO message_queue (msg_uuid, dst_address, envelope, enqueued_at, expires_at) + VALUES ($1, $2, $3, now() + (interval '1 millisecond' * $4), now() + interval '30 days')", + ) + .bind(&msg_uuid) + .bind(&dst) + .bind(env.encode_to_vec()) + .bind(i as i32) + .execute(&state.pool) + .await + .unwrap(); + } + + let (tx, mut rx) = mpsc::channel::(256); + let count = state.flush_queue(&dst, &tx).await.unwrap(); + drop(tx); + assert_eq!(count, n); + + let mut got = std::collections::HashSet::new(); + while let Some(frame) = rx.recv().await { + if let Some(v1::server_frame::Frame::Deliver(env)) = frame.frame { + assert!(got.insert(env.msg_uuid), "obálka doručena dvakrát"); + } + } + assert_eq!(got, expected, "sada doručených ≠ sada zařazených"); + + sqlx::query("DELETE FROM message_queue WHERE dst_address = $1") + .bind(&dst) + .execute(&state.pool) + .await + .unwrap(); + } + + /// Zapsat do mpsc bufferu jde i u dávno mrtvého spojení — DELIVERED_ONLINE + /// se proto smí hlásit jen u zařízení, od kterého nedávno něco přišlo. + /// Zatuchlé musí spadnout na offline větev, jinak spolkne push budíček. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn route_envelope_zatuchle_spojeni_neni_online() { + let mut state = test_state().await; + // Posunout start procesu do minulosti, ať jde značka „naposled se + // ozval na začátku" vyjádřit bez čekání reálných 70 s. + state.started = std::time::Instant::now() + - std::time::Duration::from_millis(FRESH_CONNECTION_MS + 10_000); + let (_account_id, _device_uuid, delivery_address) = seed_device(&state).await; + + // Kanál nikdo nečte, ale buffer zápis spolkne — přesně situace + // mrtvého klienta, kterou dřív relay hlásil jako doručeno. + let (tx, _rx) = mpsc::channel::(64); + let device = Arc::new(OnlineDevice::new(tx, 0)); + state + .online + .insert(delivery_address.clone(), device.clone()); + + let env = v1::Envelope { + dst_address: delivery_address.clone(), + msg_uuid: rand_bytes(16), + content_hint: 0, + payload: vec![0xCD; 64], + }; + assert_eq!( + state.route_envelope(env).await.unwrap(), + v1::receipt::Status::StoredOffline, + "zatuchlé spojení se nesmí hlásit jako doručené online" + ); + + // A naopak: čerstvě označené zařízení online být má. + device.touch(state.now_ms()); + let env2 = v1::Envelope { + dst_address: delivery_address.clone(), + msg_uuid: rand_bytes(16), + content_hint: 0, + payload: vec![0xCE; 64], + }; + assert_eq!( + state.route_envelope(env2).await.unwrap(), + v1::receipt::Status::DeliveredOnline, + "živé spojení má být online" + ); + + state.online.remove(&delivery_address); + sqlx::query("DELETE FROM message_queue WHERE dst_address = $1") + .bind(&delivery_address) + .execute(&state.pool) + .await + .unwrap(); + } + + /// Flush počítá pokusy — bez toho by strop nikdy nezabral. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn flush_queue_pocita_pokusy() { + let state = test_state().await; + let dst = rand_bytes(32); + let msg_uuid = rand_bytes(16); + let env = v1::Envelope { + dst_address: dst.clone(), + msg_uuid: msg_uuid.clone(), + content_hint: 0, + payload: vec![0x22; 32], + }; + sqlx::query( + "INSERT INTO message_queue (msg_uuid, dst_address, envelope, expires_at) + VALUES ($1, $2, $3, now() + interval '30 days')", + ) + .bind(&msg_uuid) + .bind(&dst) + .bind(env.encode_to_vec()) + .execute(&state.pool) + .await + .unwrap(); + + // Dva flushe bez ACK = dva započtené pokusy. + for _ in 0..2 { + let (tx, rx) = mpsc::channel::(16); + assert_eq!(state.flush_queue(&dst, &tx).await.unwrap(), 1); + drop(rx); + } + + let attempts: (i16,) = + sqlx::query_as("SELECT attempt_count FROM message_queue WHERE msg_uuid = $1") + .bind(&msg_uuid) + .fetch_one(&state.pool) + .await + .unwrap(); + assert_eq!(attempts.0, 2, "každý flush musí započítat pokus"); + + sqlx::query("DELETE FROM message_queue WHERE dst_address = $1") + .bind(&dst) + .execute(&state.pool) + .await + .unwrap(); + } + + /// Obálka, kterou příjemce nikdy neACKne (ztracený ACK → ratchet už + /// postoupil), musí po stropu zmizet, ne otravovat frontu 30 dní. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn flush_queue_zahodi_nepruchodnou_obalku() { + let state = test_state().await; + let dst = rand_bytes(32); + let msg_uuid = rand_bytes(16); + let env = v1::Envelope { + dst_address: dst.clone(), + msg_uuid: msg_uuid.clone(), + content_hint: 0, + payload: vec![0x33; 32], + }; + // Rovnou na stropu = obálka, která se marně posílala MAX_DELIVERY_ATTEMPTS×. + sqlx::query( + "INSERT INTO message_queue (msg_uuid, dst_address, envelope, expires_at, attempt_count) + VALUES ($1, $2, $3, now() + interval '30 days', $4)", + ) + .bind(&msg_uuid) + .bind(&dst) + .bind(env.encode_to_vec()) + .bind(MAX_DELIVERY_ATTEMPTS) + .execute(&state.pool) + .await + .unwrap(); + + let (tx, rx) = mpsc::channel::(16); + assert_eq!( + state.flush_queue(&dst, &tx).await.unwrap(), + 0, + "vyčerpaná obálka se nesmí znovu poslat" + ); + drop(rx); + + let zbylo: (i64,) = + sqlx::query_as("SELECT count(*) FROM message_queue WHERE dst_address = $1") + .bind(&dst) + .fetch_one(&state.pool) + .await + .unwrap(); + assert_eq!(zbylo.0, 0, "vyčerpaná obálka měla být zahozena"); + } + + /// Audit #3: FIFO strop zahodí nejstarší přes limit, nejnovější nechá. + #[tokio::test] + #[ignore = "vyžaduje běžící Postgres (dev DB)"] + async fn trim_queue_fifo_zahodi_nejstarsi() { + let state = test_state().await; + let dst = rand_bytes(32); + // 5 obálek po 100 B, rostoucí enqueued_at. Strop 250 B → nejnovější + // dvě (2×100 ≤ 250) zůstanou, tři nejstarší padnou. + 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; 100]) + .bind(i as i32) + .execute(&state.pool) + .await + .unwrap(); + } + + let dropped = state.trim_queue(&dst, 250).await.unwrap(); + assert_eq!(dropped, 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(); + } } diff --git a/rust/sepot-relay/src/ws.rs b/rust/sepot-relay/src/ws.rs index bed533e..6d42aff 100644 --- a/rust/sepot-relay/src/ws.rs +++ b/rust/sepot-relay/src/ws.rs @@ -12,6 +12,7 @@ use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use std::time::Duration; +use axum::body::Bytes; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::{ConnectInfo, State}; use axum::http::{HeaderMap, StatusCode}; @@ -26,38 +27,26 @@ use tokio::time::Instant; use crate::state::AppState; -/// Max souběžných WS spojení z jedné IP (za traefikem = reálná klientská -/// IP z X-Real-Ip; tor kontejner sdílí jednu — proto ne moc těsně). +/// Max souběžných WS spojení z jedné IP. Klientská IP jde přes trusted-proxy +/// logiku (viz net.rs): za traefikem reálná IP z forwarded hlavičky, onion +/// klienti sdílí IP tor kontejneru (jeden bucket) — proto ne moc těsně. const MAX_CONNS_PER_IP: u32 = 32; /// Neautentizované spojení zavřít po tomto čase. const AUTH_DEADLINE: Duration = Duration::from_secs(10); /// Rate limit rámců od klienta: max FRAME_WINDOW_MAX za FRAME_WINDOW. const FRAME_WINDOW: Duration = Duration::from_secs(10); const FRAME_WINDOW_MAX: u32 = 200; - -/// Klientská IP: hlavičky od trusted proxy (na relay dosáhne jen traefik -/// z docker sítě), fallback peer adresa (dev/tor). +/// Jak často server posílá WS Ping autentizovanému klientovi. /// -/// TRUST MODEL — POZOR: X-Real-Ip / X-Forwarded-For se přebírají bezpodmínečně -/// a řídí per-IP limity (cap spojení, rate limit). To je bezpečné VÝHRADNĚ -/// proto, že před relayem stojí důvěryhodná reverzní proxy (traefik), která -/// tyhle hlavičky vždy přepisuje reálnou klientskou IP. Klientem poslané -/// hodnoty tak nikdy neprojdou. Kdyby byl relay vystaven přímo do internetu -/// (bez proxy, která hlavičky sanitizuje), útočník by je mohl libovolně -/// podvrhnout, tím rozprostřít provoz přes falešné IP a všechny per-IP limity -/// obejít. Relay proto NIKDY nenasazovat bez takové proxy. -fn client_ip(headers: &HeaderMap, peer: SocketAddr) -> IpAddr { - let from_header = |name: &str| { - headers - .get(name) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.split(',').next()) - .and_then(|v| v.trim().parse::().ok()) - }; - from_header("x-real-ip") - .or_else(|| from_header("x-forwarded-for")) - .unwrap_or_else(|| peer.ip()) -} +/// Mobilní NAT zahazuje nečinné TCP tiše — bez keepalive se ani jedna strana +/// nedozví, že spojení je mrtvé: klient hlásí „připojeno" a píše do prázdna, +/// relay drží záznam v `online` a proto ani nepošle push budíček. Ping je +/// PROTOKOLOVÝ (ne `v1::Ping`), protože na ten dart:io odpovídá Pongem sám — +/// funguje tedy i s klienty, které jsou už venku a aplikační Ping neznají. +const PING_INTERVAL: Duration = Duration::from_secs(30); +/// Bez jediného rámce od klienta (stačí Pong) je spojení mrtvé. Tři zmeškané +/// pingy — velkoryse, ať krátký výpadek mobilní sítě spojení nezabije. +const IDLE_TIMEOUT: Duration = Duration::from_secs(95); /// Drop-guard: úbytek čítače spojení i při panice/cancelu handleru. struct ConnGuard { @@ -80,7 +69,7 @@ pub async fn ws_handler( headers: HeaderMap, ws: WebSocketUpgrade, ) -> impl IntoResponse { - let ip = client_ip(&headers, peer); + let ip = state.trusted_proxies.client_ip(peer.ip(), &headers); let over_cap = { let mut entry = state.ws_conns.entry(ip).or_insert(0); @@ -134,11 +123,28 @@ async fn handle(state: Arc, socket: WebSocket) { let (mut sink, mut stream) = socket.split(); let (tx, mut rx) = mpsc::channel::(64); + // Writer drží sink sám, takže si i pinguje — jinak by control rámce + // musely téct přes `tx`, který nese jen aplikační ServerFrame. let writer = tokio::spawn(async move { - while let Some(frame) = rx.recv().await { - let bytes = frame.encode_to_vec(); - if sink.send(WsMessage::Binary(bytes.into())).await.is_err() { - break; + let mut ping = tokio::time::interval(PING_INTERVAL); + // Zpožděné ticky se nedohánějí — po pauze chceme jeden ping, ne dávku. + ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ping.tick().await; // první tick padne hned, ten nechceme + + loop { + tokio::select! { + frame = rx.recv() => { + let Some(frame) = frame else { break }; + let bytes = frame.encode_to_vec(); + if sink.send(WsMessage::Binary(bytes.into())).await.is_err() { + break; + } + } + _ = ping.tick() => { + if sink.send(WsMessage::Ping(Bytes::new())).await.is_err() { + break; + } + } } } }); @@ -147,6 +153,9 @@ 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; + // 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; // Anti-abuse okno: počítá VŠECHNY rámce od klienta (envelope/ack/pong) let mut window_start = Instant::now(); @@ -163,10 +172,29 @@ async fn handle(state: Arc, socket: WebSocket) { } } } else { - stream.next().await + // Po AUTH hlídá spojení idle timeout: klient, který dýchá, pošle + // aspoň Pong na náš ping. Ticho = mrtvé spojení → zavřít, ať + // záznam zmizí z `online` a další zpráva jde přes push budíček. + // Bez tohohle tam zombie visel, dokud traefiku nedošel readTimeout + // (30 min), a push se celou tu dobu vůbec nezkusil. + match tokio::time::timeout(IDLE_TIMEOUT, stream.next()).await { + Ok(next) => next, + Err(_) => { + // info, ne debug: relay běží na RUST_LOG=info a kolik + // spojení umírá na ticho je provozní signál (kvalita sítě + // klientů), ne ladicí detail. Bez identifikátoru zařízení. + tracing::info!("ws zavřeno: klient neodpovídá na ping"); + break; + } + } }; let Some(Ok(msg)) = next else { break }; + // Cokoliv od klienta (i Pong na náš ping) = spojení dýchá. + if let Some(dev) = &online_device { + dev.touch(state.now_ms()); + } + let now = Instant::now(); if now.duration_since(window_start) > FRAME_WINDOW { window_start = now; @@ -242,9 +270,14 @@ async fn handle(state: Arc, socket: WebSocket) { break; } + let device = Arc::new(crate::state::OnlineDevice::new( + tx.clone(), + state.now_ms(), + )); state .online - .insert(pd.delivery_address.clone(), tx.clone()); + .insert(pd.delivery_address.clone(), device.clone()); + online_device = Some(device); authed_address = Some(pd.delivery_address.clone()); // Potvrzení AUTH — klient až teď smí hlásit „připojeno" @@ -254,7 +287,7 @@ async fn handle(state: Arc, socket: WebSocket) { ))) .await; - // last_seen jen v hrubé granularitě (privacy — zaokrouhleno na hodinu). + // last_seen jen v hrubé granularitě (privacy, §1.2). let _ = sqlx::query( "UPDATE devices SET last_seen_at = date_trunc('hour', now()) WHERE device_uuid = $1",