diff --git a/fluxer_push/src/providers/web_push.rs b/fluxer_push/src/providers/web_push.rs index f10e8c60c..f2c503e4e 100644 --- a/fluxer_push/src/providers/web_push.rs +++ b/fluxer_push/src/providers/web_push.rs @@ -94,7 +94,7 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen Err(error) => { let unreachable = Unreachable::of(&error); warn!( - error = %error, + error = %error.without_url(), kind = unreachable.label(), endpoint = %origin_of(&sub.endpoint), "web push request did not complete" diff --git a/fluxer_push/src/relay/mod.rs b/fluxer_push/src/relay/mod.rs index ad9695fc9..ae878057a 100644 --- a/fluxer_push/src/relay/mod.rs +++ b/fluxer_push/src/relay/mod.rs @@ -21,10 +21,11 @@ use base64::prelude::*; use envelope::Urgency; use fluxer_svc::shutdown::wait_for_shutdown; use quota::Quota; +use rand::Rng as _; use reject::{Reason, Rejection}; use sha2::{Digest as _, Sha256}; use std::net::SocketAddr; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant}; use tracing::{info, warn}; @@ -37,6 +38,13 @@ const TTL_HEADER: &str = "ttl"; const URGENCY_HEADER: &str = "urgency"; const JSON_CONTENT_TYPE: &str = "application/json"; const DIGEST_BYTES: usize = 8; +const LOG_SALT_BYTES: usize = 16; + +static LOG_SALT: LazyLock<[u8; LOG_SALT_BYTES]> = LazyLock::new(|| { + let mut salt = [0u8; LOG_SALT_BYTES]; + rand::rng().fill_bytes(&mut salt); + salt +}); const MIN_APNS_DEVICE_TOKEN_LEN: usize = 64; const MAX_APNS_DEVICE_TOKEN_LEN: usize = 256; const MAX_FCM_DEVICE_TOKEN_LEN: usize = 512; @@ -485,7 +493,10 @@ fn digest(value: &str) -> String { if value.is_empty() { return "-".to_owned(); } - BASE64_URL_SAFE_NO_PAD.encode(&Sha256::digest(value.as_bytes())[..DIGEST_BYTES]) + let mut hasher = Sha256::new(); + hasher.update(*LOG_SALT); + hasher.update(value.as_bytes()); + BASE64_URL_SAFE_NO_PAD.encode(&hasher.finalize()[..DIGEST_BYTES]) } #[cfg(test)] @@ -518,6 +529,16 @@ mod tests { } } + #[test] + fn the_log_digest_is_not_a_bare_hash_of_the_token() { + const TOKEN: &str = "3dbc5a5ef1a1c1666afc26f466e1b3ebaaf4c66d92dddeb0fd1b69c49641d4cd"; + let unsalted = + BASE64_URL_SAFE_NO_PAD.encode(&Sha256::digest(TOKEN.as_bytes())[..DIGEST_BYTES]); + assert_ne!(digest(TOKEN), unsalted); + assert_eq!(digest(TOKEN), digest(TOKEN)); + assert_eq!(digest(""), "-"); + } + #[test] fn payload_too_large_answers_413() { assert_eq!( diff --git a/fluxer_push/src/relay/quota.rs b/fluxer_push/src/relay/quota.rs index be6da86ce..d9496466e 100644 --- a/fluxer_push/src/relay/quota.rs +++ b/fluxer_push/src/relay/quota.rs @@ -7,7 +7,7 @@ use sha2::{Digest as _, Sha256}; use std::collections::HashMap; use std::net::IpAddr; use std::sync::Mutex; -use std::time::Instant; +use std::time::{Duration, Instant}; use tokio::sync::{Semaphore, SemaphorePermit}; const BUSY_RETRY_AFTER_SECONDS: i64 = 1; @@ -16,6 +16,7 @@ const SECONDS_PER_MINUTE: f64 = 60.0; const KEY_BYTES: usize = 16; const IPV4_PREFIX_BYTES: usize = 3; const IPV6_PREFIX_BYTES: usize = 8; +const MAX_GENERATION_AGE: Duration = Duration::from_secs(300); type Key = [u8; KEY_BYTES]; @@ -117,6 +118,7 @@ struct Bucket { struct Held { live: HashMap, aged: HashMap, + rotated_at: Option, } pub struct Buckets { @@ -135,6 +137,7 @@ impl Buckets { held: Mutex::new(Held { live: HashMap::new(), aged: HashMap::new(), + rotated_at: None, }), } } @@ -163,10 +166,50 @@ impl Buckets { bucket.tokens -= 1.0; } - if held.live.len() >= self.entries { + let rotated_at = *held.rotated_at.get_or_insert(now); + let stale = now.saturating_duration_since(rotated_at) >= MAX_GENERATION_AGE; + if held.live.len() >= self.entries || stale { held.aged = std::mem::take(&mut held.live); + held.rotated_at = Some(now); } held.live.insert(key, bucket); allowed } } + +#[cfg(test)] +mod tests { + use super::*; + + fn config() -> BucketConfig { + BucketConfig { + entries: 1_000_000, + per_minute: 60, + burst: 20, + } + } + + #[test] + fn a_generation_rotates_on_age_as_well_as_on_size() { + let buckets = Buckets::new(&config()); + let start = Instant::now(); + assert!(buckets.take([1u8; KEY_BYTES], start)); + assert_eq!(buckets.held.lock().unwrap().live.len(), 1); + let later = start + MAX_GENERATION_AGE; + assert!(buckets.take([2u8; KEY_BYTES], later)); + let held = buckets.held.lock().unwrap(); + assert_eq!(held.live.len(), 1); + assert_eq!(held.aged.len(), 1); + } + + #[test] + fn the_burst_is_spent_before_a_caller_is_refused() { + let buckets = Buckets::new(&config()); + let now = Instant::now(); + let key = [3u8; KEY_BYTES]; + for _ in 0..20 { + assert!(buckets.take(key, now)); + } + assert!(!buckets.take(key, now)); + } +} diff --git a/fluxer_push/src/vendor.rs b/fluxer_push/src/vendor.rs index 1c8a1f1e3..4d7b3c12c 100644 --- a/fluxer_push/src/vendor.rs +++ b/fluxer_push/src/vendor.rs @@ -184,7 +184,7 @@ async fn outcome( Err(error) => { let unreachable = Unreachable::of(&error); warn!( - error = %error, + error = %error.without_url(), kind = unreachable.label(), "vendor request did not complete" ); @@ -297,6 +297,17 @@ mod tests { assert!(Unreachable::of(&error).is_permanent()); } + #[tokio::test] + async fn a_logged_error_never_carries_the_device_token() { + const TOKEN: &str = "3dbc5a5ef1a1c1666afc26f466e1b3ebaaf4c66d92dddeb0fd1b69c49641d4cd"; + let error = error_for(&format!("https://push.invalid/3/device/{TOKEN}")).await; + assert!( + error.to_string().contains(TOKEN), + "reqwest still puts the url in Display, so the guard below is what matters" + ); + assert!(!error.without_url().to_string().contains(TOKEN)); + } + #[tokio::test] async fn a_refused_connection_stays_retryable() { let error = error_for("http://127.0.0.1:1/").await;