diff --git a/deploy/relay/Dockerfile b/deploy/relay/Dockerfile index bd10022..b208327 100644 --- a/deploy/relay/Dockerfile +++ b/deploy/relay/Dockerfile @@ -4,7 +4,7 @@ # Build stage potřebuje: protoc (prost-build generuje z proto/sepot.proto), # git (libsignal-protocol je git dependency s [patch.crates-io] zrcadlem). -FROM rust:1.96-slim AS build +FROM rust:1.99-slim AS build RUN apt-get update && apt-get install -y --no-install-recommends \ protobuf-compiler git pkg-config build-essential cmake \ && rm -rf /var/lib/apt/lists/* @@ -21,6 +21,12 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ && cp target/release/sepot-relay /sepot-relay FROM debian:bookworm-slim +# Revize zdroje: label pro `docker inspect` a env pro /v1/server-info +# (AGPL §13 — uživatel relaye má vědět, který commit běží). +ARG GIT_SHA=dev +LABEL org.opencontainers.image.revision=$GIT_SHA \ + org.opencontainers.image.source="https://github.com/juliusjoska/sepot" +ENV SEPOT_REVISION=$GIT_SHA RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates curl \ && rm -rf /var/lib/apt/lists/* \ diff --git a/deploy/relay/tor/Dockerfile b/deploy/relay/tor/Dockerfile index 0404ac0..4b431d8 100644 --- a/deploy/relay/tor/Dockerfile +++ b/deploy/relay/tor/Dockerfile @@ -1,6 +1,6 @@ # Tor hidden service pro sepot-relay — vlastní minimální image # (žádný third-party tor image; privacy produkt nemá kam outsourcovat důvěru) -FROM alpine:3.22 +FROM alpine:3.24 RUN apk add --no-cache tor \ && mkdir -p /var/lib/tor \ && chown -R tor:tor /var/lib/tor \ diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 5e31c15..9be946b 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -274,15 +274,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "concurrent-queue" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "const-oid" version = "0.9.6" @@ -348,9 +339,9 @@ dependencies = [ [[package]] name = "crossbeam-epoch" -version = "0.9.18" +version = "0.9.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" dependencies = [ "crossbeam-utils", ] @@ -572,11 +563,10 @@ dependencies = [ [[package]] name = "event-listener" -version = "5.4.1" +version = "5.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" dependencies = [ - "concurrent-queue", "parking", "pin-project-lite", ] @@ -826,9 +816,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.15" +version = "0.4.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +checksum = "7d29020232d6aa3fb1daca64c1127cf662cf97f254ae16c18c05b8ab635fc118" dependencies = [ "atomic-waker", "bytes", @@ -2380,9 +2370,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.41" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "once_cell", "ring", @@ -2404,9 +2394,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.13" +version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ "ring", "rustls-pki-types", @@ -2456,8 +2446,10 @@ dependencies = [ "dashmap", "futures-util", "hex", + "hmac", "ipnet", "jsonwebtoken", + "libc", "libsignal-core", "libsignal-protocol", "prost 0.13.5", @@ -2466,6 +2458,7 @@ dependencies = [ "sepot-proto", "serde", "serde_json", + "sha1", "sqlx", "tokio", "tokio-util", diff --git a/rust/sepot-relay/Cargo.toml b/rust/sepot-relay/Cargo.toml index 95fccde..3426e37 100644 --- a/rust/sepot-relay/Cargo.toml +++ b/rust/sepot-relay/Cargo.toml @@ -51,6 +51,8 @@ tower-http = { version = "0.7.0", features = ["timeout"] } reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } # FCM fast mode: RS256 JWT pro OAuth2 service account flow jsonwebtoken = "9" +# setrlimit(RLIMIT_NOFILE) při startu — docker dává soft limit 1024 +libc = "0.2" [dev-dependencies] tokio = { workspace = true, features = ["full"] } diff --git a/rust/sepot-relay/src/apns.rs b/rust/sepot-relay/src/apns.rs index d7672f3..1e079e6 100644 --- a/rust/sepot-relay/src/apns.rs +++ b/rust/sepot-relay/src/apns.rs @@ -140,6 +140,8 @@ impl ApnsSender { .header("apns-push-type", "background") .header("apns-priority", "5") .header("apns-expiration", wake_expiration(Utc::now()).to_string()) + // Víc budíčků pro jedno zařízení APNs sloučí do jednoho. + .header("apns-collapse-id", "wake") .body(r#"{"aps":{"content-available":1}}"#) .send() .await; @@ -149,7 +151,8 @@ impl ApnsSender { Ok(r) if !r.status().is_success() => { tracing::warn!(status = %r.status(), "APNs budíček odmítnut"); } - Err(e) => tracing::warn!("APNs budíček selhal: {e}"), + // without_url: URL nese device token. + Err(e) => tracing::warn!("APNs budíček selhal: {}", e.without_url()), _ => {} } } @@ -178,7 +181,7 @@ impl ApnsSender { Ok(r) if !r.status().is_success() => { tracing::warn!(status = %r.status(), "APNs VoIP push odmítnut"); } - Err(e) => tracing::warn!("APNs VoIP push selhal: {e}"), + Err(e) => tracing::warn!("APNs VoIP push selhal: {}", e.without_url()), _ => {} } } @@ -197,12 +200,7 @@ mod tests { assert_eq!(limit, crate::state::QUEUE_TTL_DAYS * 86_400); assert!(limit <= 30 * 86_400, "APNs nedrží notifikace déle než 30 dní"); } -} - -#[cfg(test)] -mod tests { - use super::*; - + #[test] fn budicek_drzi_apns_stejne_dlouho_jako_fronta() { let ted = DateTime::parse_from_rfc3339("2026-09-29T18:00:00Z") diff --git a/rust/sepot-relay/src/attachments.rs b/rust/sepot-relay/src/attachments.rs index fe7d13c..98bedc6 100644 --- a/rust/sepot-relay/src/attachments.rs +++ b/rust/sepot-relay/src/attachments.rs @@ -37,6 +37,12 @@ pub const ATTACHMENT_TTL_DAYS: i64 = 30; /// Výchozí strop velikosti (šifrovaný blob). Přepis env `SEPOT_MAX_ATTACHMENT_BYTES`. const DEFAULT_MAX_BYTES: u64 = 25 * 1024 * 1024; +/// Výchozí strop součtu všech živých příloh na disku. Přepis env +/// `SEPOT_ATTACHMENTS_TOTAL_BYTES`. Bez něj jeden registrovaný klient +/// zaplní celý filesystem (v produkci 16 GB sdílených s Postgresem) za +/// minuty a relay přestane zapisovat do DB. +const DEFAULT_TOTAL_BYTES: u64 = 4 * 1024 * 1024 * 1024; + /// Osiřelé `.part` soubory (pád mezi zápisem a přejmenováním) starší než /// tohle se při úklidu mažou. const PART_MAX_AGE: StdDuration = StdDuration::from_secs(24 * 3600); @@ -45,6 +51,7 @@ const PART_MAX_AGE: StdDuration = StdDuration::from_secs(24 * 3600); pub struct Config { pub dir: PathBuf, pub max_bytes: u64, + pub total_bytes: u64, } impl Config { @@ -55,9 +62,14 @@ impl Config { .ok() .and_then(|v| v.parse().ok()) .unwrap_or(DEFAULT_MAX_BYTES); + let total_bytes = std::env::var("SEPOT_ATTACHMENTS_TOTAL_BYTES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(DEFAULT_TOTAL_BYTES); Self { dir: PathBuf::from(dir), max_bytes, + total_bytes, } } @@ -117,6 +129,21 @@ pub async fn upload( } } + // 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 + // k blobu neváže. + let (used,): (i64,) = + sqlx::query_as("SELECT coalesce(sum(size_bytes), 0)::bigint FROM attachments") + .fetch_one(&state.pool) + .await + .map_err(internal)?; + if (used.max(0) as u64).saturating_add(cfg.max_bytes) > cfg.total_bytes { + return Err(( + StatusCode::INSUFFICIENT_STORAGE, + "úložiště příloh je plné, zkus to později".into(), + )); + } + // ThreadRng je !Send — nesmí přežít přes await, proto vlastní scope. let blob_id = { let mut rng = rand::rng(); @@ -229,6 +256,10 @@ pub async fn download( .header(header::CONTENT_TYPE, "application/octet-stream") .header(header::CONTENT_LENGTH, size) .header(header::CACHE_CONTROL, "no-store") + // Blob je bearer URL na důvěryhodné doméně — prohlížeč ho nesmí + // ani zkoušet vykreslit (nosniff + attachment). + .header(header::X_CONTENT_TYPE_OPTIONS, "nosniff") + .header(header::CONTENT_DISPOSITION, "attachment") .body(Body::from_stream(stream)) .map_err(internal) } @@ -309,6 +340,7 @@ mod db_tests { Config { dir, max_bytes: 1024, + total_bytes: 1 << 30, }, trusted, ) diff --git a/rust/sepot-relay/src/fcm.rs b/rust/sepot-relay/src/fcm.rs index 6ad586d..04bc94e 100644 --- a/rust/sepot-relay/src/fcm.rs +++ b/rust/sepot-relay/src/fcm.rs @@ -133,7 +133,9 @@ impl FcmSender { "message": { "token": device_token, "data": { "v": "1" }, - "android": { "priority": "HIGH" } + // collapse_key: čekající budíčky pro zařízení mimo síť se + // slijí do jednoho (třicet zpráv ≠ třicet pushů). + "android": { "priority": "HIGH", "collapse_key": "wake" } } }); self.send(http, body).await; @@ -171,7 +173,7 @@ impl FcmSender { Ok(resp) if !resp.status().is_success() => { tracing::warn!(status = %resp.status(), "FCM budíček odmítnut"); } - Err(e) => tracing::warn!("FCM budíček selhal: {e}"), + Err(e) => tracing::warn!("FCM budíček selhal: {}", e.without_url()), _ => {} } } diff --git a/rust/sepot-relay/src/main.rs b/rust/sepot-relay/src/main.rs index 205cbed..85cfdf2 100644 --- a/rust/sepot-relay/src/main.rs +++ b/rust/sepot-relay/src/main.rs @@ -121,6 +121,18 @@ async fn main() -> anyhow::Result<()> { }); } + // Úklid značek budíčků a OTP oken (jen paměť) + { + let state = state.clone(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(Duration::from_secs(600)); + loop { + tick.tick().await; + state.prune_marks(); + } + }); + } + let register_routes = Router::new() .route("/v1/register", post(rest::register)) .layer(GovernorLayer::new(gov_register)); @@ -128,6 +140,8 @@ async fn main() -> anyhow::Result<()> { let general_routes = Router::new() .route("/v1/server-info", get(rest::server_info)) .route("/v1/keys/{delivery_address}", get(rest::fetch_bundle)) + // doplnění one-time prekeys / rotace signed a Kyber prekey (auth) + .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)); @@ -152,16 +166,18 @@ async fn main() -> anyhow::Result<()> { )); let app = Router::new() - // health bez limitů — interní docker/uptime checky + // 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)) .merge(rest_routes) .merge(attachment_routes) .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)) - .with_state(state); + .with_state(state.clone()); - tracing::info!(%bind, "sepot-relay startuje"); + raise_nofile_limit(); + tracing::info!(%bind, revision = %state.revision, "sepot-relay startuje"); let listener = tokio::net::TcpListener::bind(&bind).await?.tap_io(|tcp| { // Bez TCP_NODELAY čeká každý další malý zápis na ACK klienta (Nagle // × zpožděné ACK = ~40 ms). Nejvíc to bolí hned po AUTH: AUTH_OK a @@ -171,11 +187,48 @@ async fn main() -> anyhow::Result<()> { tracing::debug!("TCP_NODELAY nenastaven: {e}"); } }); - // ConnectInfo: peer adresa pro rate-limit fallback a WS conn cap + // ConnectInfo: peer adresa pro rate-limit fallback a WS conn cap. + // Graceful shutdown: SIGTERM od dockeru nechá doběhnout + // rozjeté uploady; WS klienti se jen přepojí. axum::serve( listener, app.into_make_service_with_connect_info::(), ) + .with_graceful_shutdown(shutdown_signal()) .await?; Ok(()) } + +async fn shutdown_signal() { + let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("SIGTERM handler"); + tokio::select! { + _ = term.recv() => {} + _ = tokio::signal::ctrl_c() => {} + } + tracing::info!("sepot-relay končí (signál)"); +} + +/// Soft limit otevřených souborů na hard limit. Každé WS spojení je jeden +/// deskriptor; docker dává soft 1024 → strop ~1 000 klientů, zatímco hard +/// bývá 2^19. Selhání se jen zaloguje. +fn raise_nofile_limit() { + let mut lim = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + // SAFETY: getrlimit/setrlimit jen čtou a zapisují předaný struct. + unsafe { + if libc::getrlimit(libc::RLIMIT_NOFILE, &mut lim) != 0 { + return; + } + if lim.rlim_cur < lim.rlim_max { + lim.rlim_cur = lim.rlim_max; + if libc::setrlimit(libc::RLIMIT_NOFILE, &lim) != 0 { + tracing::warn!("RLIMIT_NOFILE se nepodařilo zvednout"); + return; + } + } + } + tracing::info!(nofile = lim.rlim_cur, "limit otevřených souborů"); +} diff --git a/rust/sepot-relay/src/push.rs b/rust/sepot-relay/src/push.rs index c3ced34..63da749 100644 --- a/rust/sepot-relay/src/push.rs +++ b/rust/sepot-relay/src/push.rs @@ -95,11 +95,29 @@ pub async fn send_wake( match platform { PLATFORM_UNIFIEDPUSH | PLATFORM_WEBPUSH => { let Ok(url) = std::str::from_utf8(token) else { return }; - if !endpoint_allowed(url).await { + let Some(pinned) = endpoint_allowed(url).await else { tracing::debug!("push endpoint odmítnut (scheme/SSRF guard)"); return; + }; + // Adresy z guardu se připnou i pro vlastní POST — reqwest by + // jinak resolvoval znovu a DNS s TTL 0 by napodruhé vrátil + // vnitřní adresu (DNS rebinding). + match pinned { + Some((host, addrs)) => { + let pinned_client = Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(Duration::from_secs(5)) + .connect_timeout(Duration::from_secs(3)) + .user_agent("sepot-relay") + .resolve_to_addrs(&host, &addrs) + .build(); + match pinned_client { + Ok(c) => post_wake(&c, url).await, + Err(_) => post_wake(client, url).await, + } + } + None => post_wake(client, url).await, } - post_wake(client, url).await; } PLATFORM_FCM => { // Opt-in fast mode — jen když je relay nakonfigurován @@ -139,7 +157,9 @@ async fn post_wake(client: &Client, url: &str) { 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}"), + // without_url: chyba reqwest nese URL = endpoint = identifikátor + // uživatele; do logu nesmí. + Err(e) => tracing::warn!("UnifiedPush budíček selhal: {}", e.without_url()), _ => {} } } @@ -147,32 +167,24 @@ async fn post_wake(client: &Client, url: &str) { /// Povolit jen bezpečné cíle. https vždy; http jen pro .onion (samo šifruje). /// Odmítnout hostitele resolvující na interní adresy (blind-SSRF ochrana). /// -/// ZNÁMÁ LIMITACE (DNS rebinding / TOCTOU): guard resolvuje host přes -/// `lookup_host` zde, ale samotný POST v `post_wake` resolvuje znovu. Mezi -/// kontrolou a requestem může útočníkem řízené DNS vrátit napřed veřejnou a -/// pak interní adresu, čímž guard obejde. Plný fix = resolvovat jednou a -/// připojit se přímo na ověřenou IP (reqwest `resolve()` / vlastní connector), -/// což je mimo rozsah tohoto vydání. Dopad je zúžen tím, že tělo je konstantní -/// `{"v":1}`, redirecty jsou vypnuté a odpověď se nikam nevrací (blind). -async fn endpoint_allowed(url: &str) -> bool { - let Ok(parsed) = reqwest::Url::parse(url) else { - return false; - }; - let Some(host) = parsed.host_str() else { - return false; - }; +/// Vrací `None` = odmítnuto; `Some(None)` = povoleno bez DNS (.onion); +/// `Some(Some((host, adresy)))` = povoleno, a tyhle adresy se mají pro POST +/// připnout (ochrana proti DNS rebindingu mezi kontrolou a požadavkem). +async fn endpoint_allowed(url: &str) -> Option)>> { + let parsed = reqwest::Url::parse(url).ok()?; + let host = parsed.host_str()?; let is_onion = host.ends_with(".onion"); match parsed.scheme() { "https" => {} // .onion šifruje sám; relay stejně nemá tor egress, request tiše // selže — ale schéma povolíme, ať je to připravené. - "http" if is_onion => return true, - _ => return false, + "http" if is_onion => return Some(None), + _ => return None, } // .onion přes https (neobvyklé) — neresolvuje se přes DNS, pustit. if is_onion { - return true; + return Some(None); } // Vlastnit host+port, ať `parsed` nepřežívá await na lookup_host. @@ -181,17 +193,18 @@ async fn endpoint_allowed(url: &str) -> bool { drop(parsed); let resolved = tokio::net::lookup_host((host.as_str(), port)).await; - let Ok(addrs) = resolved else { - return false; // host nejde resolvovat → nepouštět - }; - let mut saw_any = false; + let addrs = resolved.ok()?; // host nejde resolvovat → nepouštět + let mut pinned = Vec::new(); for a in addrs { - saw_any = true; if is_internal_ip(&a.ip()) { - return false; // aspoň jedna interní adresa → celé odmítnout + return None; // aspoň jedna interní adresa → celé odmítnout } + pinned.push(a); } - saw_any + if pinned.is_empty() { + return None; + } + Some(Some((host, pinned))) } fn is_internal_v4(v4: &std::net::Ipv4Addr) -> bool { @@ -214,7 +227,14 @@ fn is_internal_ip(ip: &IpAddr) -> bool { || v6.is_unspecified() || (o[0] & 0xfe) == 0xfc // fc00::/7 ULA || (o[0] == 0xfe && (o[1] & 0xc0) == 0x80) // fe80::/10 link-local + || (o[0] == 0xfe && (o[1] & 0xc0) == 0xc0) // fec0::/10 site-local || v6.to_ipv4_mapped().map(|v4| is_internal_v4(&v4)).unwrap_or(false) + // 64:ff9b::/96 NAT64 a 2002::/16 6to4 nesou vnořenou IPv4 + || (o[..12] == [0, 0x64, 0xff, 0x9b, 0, 0, 0, 0, 0, 0, 0, 0] + && is_internal_v4(&std::net::Ipv4Addr::new(o[12], o[13], o[14], o[15]))) + || (o[0] == 0x20 + && o[1] == 0x02 + && is_internal_v4(&std::net::Ipv4Addr::new(o[2], o[3], o[4], o[5]))) } } } @@ -252,20 +272,33 @@ mod tests { assert!(is_internal_ip(&mapped)); } + #[test] + fn nat64_a_6to4_s_vnorenou_privatni_v4_jsou_interni() { + let nat64: IpAddr = "64:ff9b::10.0.0.1".parse().unwrap(); + assert!(is_internal_ip(&nat64)); + let sixtofour: IpAddr = "2002:c0a8:101::1".parse().unwrap(); // 192.168.1.1 + assert!(is_internal_ip(&sixtofour)); + let verejna_nat64: IpAddr = "64:ff9b::1.1.1.1".parse().unwrap(); + assert!(!is_internal_ip(&verejna_nat64)); + } + #[tokio::test] async fn schema_a_ssrf_guard() { // http (ne-onion) se odmítne - assert!(!endpoint_allowed("http://ntfy.sh/abc").await); + assert!(endpoint_allowed("http://ntfy.sh/abc").await.is_none()); // loopback se odmítne i přes https - assert!(!endpoint_allowed("https://127.0.0.1/abc").await); - assert!(!endpoint_allowed("https://localhost/abc").await); + assert!(endpoint_allowed("https://127.0.0.1/abc").await.is_none()); + assert!(endpoint_allowed("https://localhost/abc").await.is_none()); // interní IP se odmítne - assert!(!endpoint_allowed("https://10.0.0.5:8000/x").await); - // http na .onion se pustí (schéma OK) - assert!(endpoint_allowed("http://cipjsctoeyf4fcrntvdvt22o2ansgr73qvthmewo246vuntd7cqmykyd.onion/x").await); + assert!(endpoint_allowed("https://10.0.0.5:8000/x").await.is_none()); + // http na .onion se pustí (schéma OK), bez DNS → nic k připnutí + assert_eq!( + endpoint_allowed("http://cipjsctoeyf4fcrntvdvt22o2ansgr73qvthmewo246vuntd7cqmykyd.onion/x").await, + Some(None) + ); // nesmysl se odmítne - assert!(!endpoint_allowed("není url").await); - assert!(!endpoint_allowed("ftp://ntfy.sh/x").await); + assert!(endpoint_allowed("není url").await.is_none()); + assert!(endpoint_allowed("ftp://ntfy.sh/x").await.is_none()); } #[test] diff --git a/rust/sepot-relay/src/rest.rs b/rust/sepot-relay/src/rest.rs index 56eba21..e257769 100644 --- a/rust/sepot-relay/src/rest.rs +++ b/rust/sepot-relay/src/rest.rs @@ -40,14 +40,60 @@ pub async fn health() -> Json { Json(serde_json::json!({ "status": "ok" })) } +/// Readiness: health je jen „proces žije", tohle ověří i DB. +/// Pro sondy monitoringu — DB mimo provoz = 503, ne falešné 200. +pub async fn ready(State(state): State>) -> ApiResult> { + let probe = tokio::time::timeout( + std::time::Duration::from_secs(1), + sqlx::query("SELECT 1").execute(&state.pool), + ) + .await; + match probe { + Ok(Ok(_)) => Ok(Json(serde_json::json!({ "status": "ready" }))), + _ => Err((StatusCode::SERVICE_UNAVAILABLE, "databáze neodpovídá".into())), + } +} + +/// Veřejné informace o relayi. `source` + `revision` plní AGPL §13 — kdo +/// s relayem mluví, má vědět, který zdroj právě běží. pub async fn server_info(State(state): State>) -> Json { Json(serde_json::json!({ "frame_version": 1, "trust_root": BASE64_STANDARD.encode(&state.certs.trust_root_public), - "capabilities": ["pqxdh", "sealed-sender"], + "capabilities": ["pqxdh", "sealed-sender", "prekey-upload", "identity-reclaim"], + "source": state.source_url, + "revision": state.revision, })) } +/// Ověří hlavičku `X-Sepot-Identity-Proof: .` — podpis +/// identity klíčem nad `"\n"`, stejné okno jako +/// X-Sepot-Auth. Důkaz držení privátního klíče bez znalosti device_uuid: +/// obnova ze 24 slov na novém telefonu nemá nic jiného. +fn identity_proof_valid(headers: &HeaderMap, identity: &PublicKey, method_path: &str) -> bool { + let Some(header) = headers + .get("x-sepot-identity-proof") + .and_then(|v| v.to_str().ok()) + else { + return false; + }; + let Some((ts_str, sig_b64)) = header.split_once('.') else { + return false; + }; + let Ok(ts) = ts_str.parse::() else { + return false; + }; + let Ok(signature) = BASE64_STANDARD.decode(sig_b64) else { + return false; + }; + let now_ms = chrono::Utc::now().timestamp_millis(); + if now_ms.abs_diff(ts) > AUTH_WINDOW_MS as u64 { + return false; + } + let message = format!("{method_path}\n{ts}"); + identity.verify_signature(message.as_bytes(), &signature) +} + // ── POST /v1/register ────────────────────────────────────────────────── #[derive(Deserialize)] @@ -83,11 +129,17 @@ const MAX_ONE_TIME_PREKEYS: usize = 100; pub async fn register( State(state): State>, + headers: HeaderMap, Json(body): Json, ) -> ApiResult<(StatusCode, Json)> { if body.one_time_prekeys.len() > MAX_ONE_TIME_PREKEYS { return Err(bad("příliš mnoho one-time prekeys")); } + // libsignal DeviceId je 1..=127; registration_id kladné 14bitové číslo. + // Mimo rozsah by registrace prošla a GET /v1/certificate pak padal na 500. + if !(1..=127).contains(&body.device_id) || !(1..=16_383).contains(&body.registration_id) { + return Err(bad("device_id nebo registration_id mimo rozsah")); + } let identity_key = b64("identity_key", &body.identity_key, Some(33))?; let delivery_address = b64("delivery_address", &body.delivery_address, Some(32))?; let spk_pub = b64("signed_prekey.public_key", &body.signed_prekey.public_key, Some(33))?; @@ -133,9 +185,39 @@ pub async fn register( .execute(&mut *tx) .await .map_err(internal)?; - if inserted.rows_affected() == 0 { - return Err((StatusCode::CONFLICT, "identita už existuje".into())); - } + // Identita už existuje = obnova ze 24 slov na novém telefonu (nebo + // reinstalace). Bez důkazu držení klíče zůstává 409 (anti-squatting); + // s důkazem účet převezme nové zařízení: stará zařízení, jejich + // prekeys, push tokeny i nedoručená fronta pryč (nové zařízení staré + // ratchet relace stejně nemá), account_id zůstává — kontakty ho znají. + let account_id = if inserted.rows_affected() == 0 { + if !identity_proof_valid(&headers, &identity, "POST /v1/register") { + return Err((StatusCode::CONFLICT, "identita už existuje".into())); + } + let (existing,): (Vec,) = + sqlx::query_as("SELECT account_id FROM identities WHERE identity_key = $1") + .bind(&identity_key) + .fetch_one(&mut *tx) + .await + .map_err(internal)?; + sqlx::query( + "DELETE FROM message_queue WHERE dst_address IN ( + SELECT delivery_address FROM devices WHERE account_id = $1)", + ) + .bind(&existing) + .execute(&mut *tx) + .await + .map_err(internal)?; + sqlx::query("DELETE FROM devices WHERE account_id = $1") + .bind(&existing) + .execute(&mut *tx) + .await + .map_err(internal)?; + tracing::info!("identita převzata novým zařízením (obnova)"); + existing + } else { + account_id + }; sqlx::query( "INSERT INTO devices (device_uuid, account_id, device_id, registration_id, delivery_address) @@ -274,18 +356,25 @@ pub async fn fetch_bundle( .await .map_err(internal)?; - // Spotřebuj jeden one-time prekey (DELETE ... RETURNING). - let otp: Option<(i32, Vec)> = sqlx::query_as( - "DELETE FROM one_time_prekeys - WHERE (device_uuid, key_id) IN ( - SELECT device_uuid, key_id FROM one_time_prekeys - WHERE device_uuid = $1 ORDER BY key_id LIMIT 1) - RETURNING key_id, public_key", - ) - .bind(&device_uuid) - .fetch_optional(&state.pool) - .await - .map_err(internal)?; + // Spotřebuj jeden one-time prekey (DELETE ... RETURNING) — + // ale jen v mezích hodinového stropu na adresu: bundle bez OTP je + // povolený fallback, vyčerpaná zásoba cizím ne. + let otp: Option<(i32, Vec)> = if state.allow_otp_issue(&address) { + sqlx::query_as( + "DELETE FROM one_time_prekeys + WHERE (device_uuid, key_id) IN ( + SELECT device_uuid, key_id FROM one_time_prekeys + WHERE device_uuid = $1 ORDER BY key_id + FOR UPDATE SKIP LOCKED LIMIT 1) + RETURNING key_id, public_key", + ) + .bind(&device_uuid) + .fetch_optional(&state.pool) + .await + .map_err(internal)? + } else { + None + }; out.push(DeviceBundleOut { device_id, @@ -321,6 +410,7 @@ pub(crate) struct AuthedDevice { pub account_id: Vec, pub identity_key: Vec, pub device_id: i32, + pub device_uuid: Vec, } /// Ověří hlavičku `X-Sepot-Auth: ..`, @@ -349,7 +439,8 @@ pub(crate) async fn authenticate_device( let signature = BASE64_STANDARD.decode(sig_b64).map_err(|_| unauthorized())?; let now_ms = chrono::Utc::now().timestamp_millis(); - if (now_ms - ts).abs() > AUTH_WINDOW_MS { + // abs_diff: `(now - ts).abs()` by u extrémního ts v release přetekl. + if now_ms.abs_diff(ts) > AUTH_WINDOW_MS as u64 { return Err(unauthorized()); } @@ -376,9 +467,157 @@ pub(crate) async fn authenticate_device( account_id, identity_key, device_id, + device_uuid, }) } +// ── POST /v1/keys ────────────────────────────────────────────────────── + +#[derive(Deserialize)] +pub struct UploadKeysIn { + #[serde(default)] + pub signed_prekey: Option, + #[serde(default)] + pub kyber_prekey: Option, + #[serde(default)] + pub one_time_prekeys: Vec, +} + +#[derive(Serialize)] +pub struct UploadKeysOut { + /// Kolik OTP zařízení má po nahrání. + 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). +const SIGNED_PREKEY_GRACE_DAYS: i64 = 30; + +/// Doplnění one-time prekeys a rotace signed / Kyber prekey. Auth podpis: `"POST /v1/keys\n"`. +/// Podpisy nových prekeys se ověřují proti identity klíči zařízení. +pub async fn upload_keys( + State(state): State>, + headers: HeaderMap, + Json(body): Json, +) -> ApiResult> { + let authed = authenticate_device(&state, &headers, "POST /v1/keys").await?; + if body.one_time_prekeys.len() > MAX_ONE_TIME_PREKEYS { + return Err(bad("příliš mnoho one-time prekeys")); + } + let identity = + PublicKey::deserialize(&authed.identity_key).map_err(|_| bad("neplatný identity klíč"))?; + + let mut tx = state.pool.begin().await.map_err(internal)?; + + if let Some(spk) = &body.signed_prekey { + let public = b64("signed_prekey.public_key", &spk.public_key, Some(33))?; + let sig = b64( + "signed_prekey.signature", + spk.signature.as_deref().ok_or_else(|| bad("chybí podpis signed prekey"))?, + Some(64), + )?; + if !identity.verify_signature(&public, &sig) { + return Err(bad("podpis signed prekey nesedí")); + } + sqlx::query( + "INSERT INTO signed_prekeys (device_uuid, key_id, public_key, signature) + VALUES ($1, $2, $3, $4) + ON CONFLICT (device_uuid, key_id) DO NOTHING", + ) + .bind(&authed.device_uuid) + .bind(spk.key_id) + .bind(&public) + .bind(&sig) + .execute(&mut *tx) + .await + .map_err(internal)?; + // Starší než grace a zároveň ne nejnovější → pryč. + sqlx::query( + "DELETE FROM signed_prekeys WHERE device_uuid = $1 + AND created_at < now() - make_interval(days => $2) + AND key_id <> (SELECT key_id FROM signed_prekeys WHERE device_uuid = $1 + ORDER BY created_at DESC LIMIT 1)", + ) + .bind(&authed.device_uuid) + .bind(SIGNED_PREKEY_GRACE_DAYS as i32) + .execute(&mut *tx) + .await + .map_err(internal)?; + } + + if let Some(kyber) = &body.kyber_prekey { + let public = b64("kyber_prekey.public_key", &kyber.public_key, None)?; + if public.len() > 2048 { + return Err(bad("kyber prekey je příliš dlouhý")); + } + let sig = b64( + "kyber_prekey.signature", + kyber.signature.as_deref().ok_or_else(|| bad("chybí podpis kyber prekey"))?, + Some(64), + )?; + if !identity.verify_signature(&public, &sig) { + return Err(bad("podpis kyber prekey nesedí")); + } + sqlx::query( + "INSERT INTO kyber_prekeys (device_uuid, key_id, public_key, signature, is_last_resort) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (device_uuid, key_id) DO NOTHING", + ) + .bind(&authed.device_uuid) + .bind(kyber.key_id) + .bind(&public) + .bind(&sig) + .bind(kyber.is_last_resort) + .execute(&mut *tx) + .await + .map_err(internal)?; + sqlx::query( + "DELETE FROM kyber_prekeys WHERE device_uuid = $1 AND is_last_resort + AND created_at < now() - make_interval(days => $2) + AND key_id <> (SELECT key_id FROM kyber_prekeys WHERE device_uuid = $1 + AND is_last_resort ORDER BY created_at DESC LIMIT 1)", + ) + .bind(&authed.device_uuid) + .bind(SIGNED_PREKEY_GRACE_DAYS as i32) + .execute(&mut *tx) + .await + .map_err(internal)?; + } + + let (mut stored,): (i64,) = + sqlx::query_as("SELECT count(*) FROM one_time_prekeys WHERE device_uuid = $1") + .bind(&authed.device_uuid) + .fetch_one(&mut *tx) + .await + .map_err(internal)?; + for otp in &body.one_time_prekeys { + if stored >= MAX_STORED_ONE_TIME_PREKEYS { + break; + } + let public = b64("one_time_prekeys.public_key", &otp.public_key, Some(33))?; + let res = sqlx::query( + "INSERT INTO one_time_prekeys (device_uuid, key_id, public_key) VALUES ($1, $2, $3) + ON CONFLICT (device_uuid, key_id) DO NOTHING", + ) + .bind(&authed.device_uuid) + .bind(otp.key_id) + .bind(&public) + .execute(&mut *tx) + .await + .map_err(internal)?; + stored += res.rows_affected() as i64; + } + + tx.commit().await.map_err(internal)?; + Ok(Json(UploadKeysOut { + one_time_prekeys: stored, + })) +} + // ── DELETE /v1/identity ──────────────────────────────────────────────── /// Deregistrace účtu (GDPR čl. 17, store požadavek na smazání účtu). diff --git a/rust/sepot-relay/src/state.rs b/rust/sepot-relay/src/state.rs index f182ed3..0721ef2 100644 --- a/rust/sepot-relay/src/state.rs +++ b/rust/sepot-relay/src/state.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::time::Instant; use chrono::{DateTime, Duration, Utc}; use dashmap::DashMap; @@ -39,6 +40,36 @@ 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); +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; +const OTP_ISSUE_WINDOW: std::time::Duration = std::time::Duration::from_secs(3600); + +/// Značky budíčků jednoho cíle (jen v paměti, nikam se neukládají). +#[derive(Default)] +pub struct WakeMarks { + last_wake: Option, + call_wakes: std::collections::VecDeque, +} + /// 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 @@ -126,6 +157,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, + /// 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. + pub otp_issued: DashMap, (Instant, u32)>, + /// Odkaz na zdroj a revize běžící binárky (AGPL §13, /v1/server-info). + pub source_url: String, + pub revision: String, } impl AppState { @@ -147,9 +187,96 @@ 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), + wake_marks: DashMap::new(), + otp_issued: DashMap::new(), + source_url: std::env::var("SEPOT_SOURCE_URL") + .unwrap_or_else(|_| "https://github.com/juliusjoska/sepot".into()), + revision: std::env::var("SEPOT_REVISION").unwrap_or_else(|_| "dev".into()), } } + /// Úklid značek starších než největší okno — volá periodická úloha. + pub fn prune_marks(&self) { + let now = Instant::now(); + self.wake_marks.retain(|_, m| { + m.call_wakes + .retain(|t| now.duration_since(*t) < CALL_WAKE_WINDOW); + m.last_wake + .map(|t| now.duration_since(t) < CALL_WAKE_WINDOW) + .unwrap_or(false) + || !m.call_wakes.is_empty() + }); + self.otp_issued + .retain(|_, (start, _)| now.duration_since(*start) < OTP_ISSUE_WINDOW); + } + + /// Smí se vydat další OTP pro tuhle adresu? Počítá v hodinovém okně. + pub fn allow_otp_issue(&self, dst_address: &[u8]) -> bool { + let now = Instant::now(); + let mut entry = self + .otp_issued + .entry(dst_address.to_vec()) + .or_insert((now, 0)); + if now.duration_since(entry.0) >= OTP_ISSUE_WINDOW { + *entry = (now, 0); + } + if entry.1 >= OTP_ISSUE_MAX_PER_HOUR { + return false; + } + entry.1 += 1; + true + } + + /// Běžný budíček smí odejít nejvýš jednou za WAKE_MIN_INTERVAL na cíl. + fn allow_wake(&self, dst_address: &[u8]) -> bool { + let now = Instant::now(); + let mut marks = self.wake_marks.entry(dst_address.to_vec()).or_default(); + if marks + .last_wake + .map(|t| now.duration_since(t) < WAKE_MIN_INTERVAL) + .unwrap_or(false) + { + return false; + } + marks.last_wake = Some(now); + true + } + + /// Budíček hovoru: vlastní strop na cíl (viz CALL_WAKE_*). Vrací false, + /// když je cíl v okně už vyčerpaný — pak jde nanejvýš běžný budíček. + fn allow_call_wake(&self, dst_address: &[u8]) -> bool { + let now = Instant::now(); + let mut marks = self.wake_marks.entry(dst_address.to_vec()).or_default(); + marks + .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 { + return false; + } + } + if marks.call_wakes.len() >= CALL_WAKE_MAX_PER_WINDOW { + return false; + } + marks.call_wakes.push_back(now); + marks.last_wake = Some(now); + true + } + + /// Kolik one-time prekeys zařízení ještě má. + pub async fn one_time_prekey_count(&self, device_uuid: &[u8]) -> anyhow::Result { + let (n,): (i64,) = + sqlx::query_as("SELECT count(*) FROM one_time_prekeys WHERE device_uuid = $1") + .bind(device_uuid) + .fetch_one(&self.pool) + .await?; + 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í. @@ -192,7 +319,11 @@ impl AppState { frame: Some(v1::server_frame::Frame::Deliver(env)), }; let fresh = dev.is_fresh(self.now_ms()); - let queued = dev.tx.send(deliver).await.is_ok(); + // try_send, ne send().await: plný kanál znamená, že příjemce ze + // socketu nečte. Čekat na něj by drželo odesílatelovo spojení + // (a jeho paměť) rukojmím — nečtoucí příjemci by tak shodili + // relay. Obálka je v DB, půjde budíček. + let queued = dev.tx.try_send(deliver).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 @@ -201,10 +332,10 @@ impl AppState { return Ok(v1::receipt::Status::DeliveredOnline); } } - // Offline, zatuchlé nebo selhalý kanál → zkusit push budíček. - if is_call { + // Offline, zatuchlé nebo plný kanál → zkusit push budíček. + if is_call && self.allow_call_wake(&dst) { self.wake_call(&dst).await; - } else { + } else if self.allow_wake(&dst) { self.wake(&dst).await; } Ok(v1::receipt::Status::StoredOffline) @@ -223,7 +354,7 @@ impl AppState { let deliver = v1::ServerFrame { frame: Some(v1::server_frame::Frame::Deliver(env)), }; - if dev.tx.send(deliver).await.is_ok() { + if dev.tx.try_send(deliver).is_ok() { v1::receipt::Status::DeliveredOnline } else { v1::receipt::Status::Dropped diff --git a/rust/sepot-relay/src/turn.rs b/rust/sepot-relay/src/turn.rs index 091d6f0..06b8f3c 100644 --- a/rust/sepot-relay/src/turn.rs +++ b/rust/sepot-relay/src/turn.rs @@ -12,8 +12,9 @@ use sepot_proto::v1; use sha1::Sha1; /// Platnost údajů. Pokrývá i dlouhý hovor — coturn kontroluje expiraci jen -/// při alokaci a jejím obnovení, ne u běžícího hovoru. -pub const TTL_SECONDS: u32 = 24 * 3600; +/// při alokaci a jejím obnovení, ne u běžícího hovoru. Dvanáct hodin stačí +/// (dřív 24 h: zbytečně dlouhé okno pro údaje, které přežijí i smazání účtu). +pub const TTL_SECONDS: u32 = 12 * 3600; pub struct TurnConfig { secret: Vec, diff --git a/rust/sepot-relay/src/ws.rs b/rust/sepot-relay/src/ws.rs index b3c18d8..c554d89 100644 --- a/rust/sepot-relay/src/ws.rs +++ b/rust/sepot-relay/src/ws.rs @@ -27,15 +27,28 @@ use tokio::time::Instant; use crate::state::AppState; -/// 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. +/// Neautentizované spojení zavřít po tomto čase od navázání — absolutně, +/// ne „od posledního rámce": Pong rámce před AUTH by jinak držely spojení +/// bez konce. 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. +/// 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; +/// 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; +const MAX_VOIP_TOKEN_BYTES: usize = 256; +/// TURN údaje platí hodiny — nové vydat nejdřív po tomhle čase, jinak se +/// na jednom spojení dají farmit stovky přihlášení (kvóta coturnu je na jméno). +const TURN_REISSUE_AFTER: Duration = Duration::from_secs(600); +/// Pod tímhle počtem one-time prekeys relay po AUTH pošle PrekeysLow. +pub const PREKEYS_LOW_THRESHOLD: i64 = 20; /// Jak často server posílá WS Ping autentizovanému klientovi. /// /// Mobilní NAT zahazuje nečinné TCP tiše — bez keepalive se ani jedna strana @@ -73,7 +86,7 @@ pub async fn ws_handler( let over_cap = { let mut entry = state.ws_conns.entry(ip).or_insert(0); - if *entry >= MAX_CONNS_PER_IP { + if *entry >= state.max_conns_per_ip { true } else { *entry += 1; @@ -157,14 +170,18 @@ async fn handle(state: Arc, socket: WebSocket) { // 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) + // Anti-abuse okno: počítá rámce, které něco stojí (viz FRAME_WINDOW_MAX) let mut window_start = Instant::now(); let mut window_frames: u32 = 0; + let auth_deadline = Instant::now() + AUTH_DEADLINE; + let mut auth_attempts: u32 = 0; + // Naposledy vydané TURN údaje — do TURN_REISSUE_AFTER se vrací tytéž. + let mut turn_issued: Option<(Instant, v1::TurnCredentials)> = None; loop { - // Před AUTH platí deadline — viset neautentizovaný nejde + // Před AUTH platí absolutní deadline — viset neautentizovaný nejde let next = if authed_address.is_none() { - match tokio::time::timeout(AUTH_DEADLINE, stream.next()).await { + match tokio::time::timeout_at(auth_deadline, stream.next()).await { Ok(next) => next, Err(_) => { tracing::debug!("ws zavřeno: AUTH deadline"); @@ -195,17 +212,6 @@ async fn handle(state: Arc, socket: WebSocket) { dev.touch(state.now_ms()); } - let now = Instant::now(); - if now.duration_since(window_start) > FRAME_WINDOW { - window_start = now; - window_frames = 0; - } - window_frames += 1; - if window_frames > FRAME_WINDOW_MAX { - let _ = tx.send(error_frame(429, "příliš mnoho rámců")).await; - break; - } - let data = match msg { WsMessage::Binary(b) => b, WsMessage::Close(_) => break, @@ -216,8 +222,30 @@ async fn handle(state: Arc, socket: WebSocket) { break; }; + let cheap = matches!( + frame.frame, + Some(v1::client_frame::Frame::Ack(_)) | Some(v1::client_frame::Frame::Pong(_)) + ); + if !cheap { + let now = Instant::now(); + if now.duration_since(window_start) > FRAME_WINDOW { + window_start = now; + window_frames = 0; + } + window_frames += 1; + if window_frames > FRAME_WINDOW_MAX { + let _ = tx.send(error_frame(429, "příliš mnoho rámců")).await; + break; + } + } + match frame.frame { Some(v1::client_frame::Frame::Auth(auth)) if authed_address.is_none() => { + auth_attempts += 1; + if auth_attempts > MAX_AUTH_ATTEMPTS { + let _ = tx.send(error_frame(429, "příliš mnoho AUTH pokusů")).await; + break; + } if auth.device_uuid.len() != 16 || auth.nonce_client.len() != 16 { let _ = tx.send(error_frame(400, "neplatný AUTH")).await; break; @@ -303,14 +331,33 @@ async fn handle(state: Arc, socket: WebSocket) { Err(e) => tracing::warn!("flush fronty selhal: {e}"), } + // Docházející one-time prekeys: klient si je doplní přes + // POST /v1/keys. Bez tohohle hintu zůstal účet po vyčerpání + // natrvalo bez OTP. + if let Ok(remaining) = state.one_time_prekey_count(&pd.device_uuid).await { + if remaining < PREKEYS_LOW_THRESHOLD { + let _ = tx + .send(server_frame(v1::server_frame::Frame::PrekeysLow( + v1::PrekeysLow { + remaining: remaining.max(0) as u32, + }, + ))) + .await; + } + } + authed_device = Some(pd.device_uuid); } Some(v1::client_frame::Frame::RegisterPush(reg)) if authed_address.is_some() => { if let Some(device) = &authed_device { - let platform = reg.platform as i16; + let platform = i16::try_from(reg.platform).unwrap_or(-1); if platform < 0 || platform > crate::push::PLATFORM_MAX { let _ = tx.send(error_frame(400, "neznámá push platforma")).await; + } else if reg.endpoint.len() > MAX_PUSH_ENDPOINT_BYTES + || reg.voip_token.len() > MAX_VOIP_TOKEN_BYTES + { + let _ = tx.send(error_frame(400, "push endpoint je příliš dlouhý")).await; } else { let endpoint = (!reg.endpoint.is_empty()).then_some(reg.endpoint.as_slice()); @@ -357,9 +404,21 @@ async fn handle(state: Arc, socket: WebSocket) { Some(v1::client_frame::Frame::TurnRequest(_)) if authed_address.is_some() => { let frame = match &state.turn { - Some(turn) => server_frame(v1::server_frame::Frame::TurnCredentials( - turn.credentials(chrono::Utc::now().timestamp()), - )), + Some(turn) => { + let fresh = turn_issued + .as_ref() + .filter(|(at, _)| at.elapsed() < TURN_REISSUE_AFTER) + .map(|(_, c)| c.clone()); + let creds = match fresh { + Some(c) => c, + None => { + let c = turn.credentials(chrono::Utc::now().timestamp()); + turn_issued = Some((Instant::now(), c.clone())); + c + } + }; + server_frame(v1::server_frame::Frame::TurnCredentials(creds)) + } None => error_frame(503, "TURN není nakonfigurován"), }; let _ = tx.send(frame).await;