diff --git a/src/error.rs b/src/error.rs index 20bc598..bc19a25 100644 --- a/src/error.rs +++ b/src/error.rs @@ -58,6 +58,24 @@ pub enum Error { Hex(#[from] hex::FromHexError), } +impl Error { + /// Strip any URL embedded in an underlying HTTP transport error. + /// + /// `reqwest::Error`'s `Display` includes the request URL, and the APNs + /// request URL embeds the raw device token (`/3/device/`). Every + /// transport error that can reach a log sink or be converted downstream + /// must pass through this (or `reqwest::Error::without_url` directly) + /// first so the token can never leak into logs (issue #172). No-op for + /// non-HTTP errors. + #[must_use] + pub fn redact_transport_url(self) -> Self { + match self { + Self::Http(error) => Self::Http(error.without_url()), + other => other, + } + } +} + /// Result type alias using our Error type. pub type Result = std::result::Result; @@ -138,4 +156,28 @@ mod tests { assert!(debug_str.contains("Crypto")); assert!(debug_str.contains("test")); } + + #[tokio::test] + async fn test_redact_transport_url_strips_device_token_from_http_error() { + // The APNs URL embeds the device token in its path; redact_transport_url + // must strip it so the error Display cannot leak the token (issue #172). + let error = reqwest::Client::new() + .post("http://127.0.0.1:9/3/device/aabbccddeeff00112233") + .send() + .await + .unwrap_err(); + assert!(error.to_string().contains("aabbccddeeff00112233")); + + let redacted = Error::Http(error).redact_transport_url(); + assert!( + !redacted.to_string().contains("aabbccddeeff00112233"), + "device token leaked after redaction: {redacted}" + ); + } + + #[test] + fn test_redact_transport_url_is_noop_for_non_http_errors() { + let err = Error::Apns("bad request".to_string()).redact_transport_url(); + assert_eq!(err.to_string(), "APNs error: bad request"); + } } diff --git a/src/push/apns.rs b/src/push/apns.rs index feb8c51..63fe99b 100644 --- a/src/push/apns.rs +++ b/src/push/apns.rs @@ -2,19 +2,18 @@ //! //! Uses token-based (JWT) authentication with a .p8 key file. -use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use jsonwebtoken::{Algorithm, EncodingKey, Header, encode}; use reqwest::Client; use serde::{Deserialize, Serialize}; -use tokio::sync::RwLock; use tracing::{debug, trace, warn}; -use zeroize::{Zeroize, ZeroizeOnDrop, Zeroizing}; +use zeroize::Zeroizing; use crate::config::{ApnsConfig, ApnsPayloadMode}; use crate::error::{Error, Result}; use crate::metrics::Metrics; +use crate::push::auth::{AuthTokenGenerator, MintedToken, TokenAcquisitionError, TokenCache}; use crate::push::retry::{self, PushSendOutcome, RetryConfig, SendAttemptResult}; /// APNs JWT token lifetime (50 minutes, less than the 1 hour max). @@ -34,14 +33,6 @@ struct ApnsClaims { exp: u64, } -/// Cached JWT token. -#[derive(Zeroize, ZeroizeOnDrop)] -pub(crate) struct CachedToken { - token: Zeroizing, - #[zeroize(skip)] - expires_at: SystemTime, -} - /// APNs notification payload. #[derive(Debug, Serialize)] #[serde(untagged)] @@ -190,28 +181,102 @@ fn classify_apns_400(reason: &str) -> Apns400Classification { } } -/// Validate an APNs device token format. +/// Whether an APNs `403` `reason` indicates the provider *token* (JWT) is the +/// problem, as opposed to a configuration/environment fault that also returns +/// `403`. /// -/// MIP-05 treats APNs tokens as variable-length opaque bytes. Transponder -/// hex-encodes those bytes for the APNs device-token URL path, so only reject -/// empty, odd-length, or non-hex strings here. +/// APNs returns `403` for several unrelated conditions: `ExpiredProviderToken` +/// and `InvalidProviderToken` mean the JWT is stale/wrong and re-minting can +/// recover, but `BadCertificateEnvironment`, `Forbidden`, `TopicDisallowed`, +/// etc. are static misconfigurations. Evicting (and re-signing) the JWT on +/// every non-JWT 403 turns a static misconfiguration into a per-notification +/// ES256 re-sign stampede, so only the JWT reasons trigger eviction and a +/// fresh-token retry (issues #145, #85). #[must_use] -fn is_valid_device_token(token: &str) -> bool { - !token.is_empty() - && token.len().is_multiple_of(2) - && token.chars().all(|c| c.is_ascii_hexdigit()) +fn is_apns_jwt_reason(reason: &str) -> bool { + matches!(reason, "ExpiredProviderToken" | "InvalidProviderToken") +} + +/// APNs credential generator: local ES256 signing of a short-lived JWT. +pub(crate) struct ApnsTokenGenerator { + encoding_key: Option, + team_id: String, + key_id: String, + metrics: Option, +} + +impl ApnsTokenGenerator { + /// Whether a signing key was loaded (used by `is_configured`). + fn has_encoding_key(&self) -> bool { + self.encoding_key.is_some() + } +} + +impl AuthTokenGenerator for ApnsTokenGenerator { + async fn mint(&self) -> std::result::Result { + let encoding_key = self.encoding_key.as_ref().ok_or_else(|| { + TokenAcquisitionError::permanent(Error::Apns("No encoding key configured".to_string())) + })?; + + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|e| { + TokenAcquisitionError::permanent(Error::Apns(format!("System time error: {e}"))) + })? + .as_secs(); + + // Backdate `iat` by the clock-skew leeway so a fast host clock does not + // yield an `iat` in APNs's future (403 InvalidProviderToken); `exp` + // stays within Apple's 1-hour max measured from the backdated `iat`. + let (iat, exp) = crate::push::auth_jwt_iat_exp(now, TOKEN_EXPIRATION_SECS); + + let claims = ApnsClaims { + iss: self.team_id.clone(), + iat, + exp, + }; + + let mut header = Header::new(Algorithm::ES256); + header.kid = Some(self.key_id.clone()); + + let token = Zeroizing::new( + encode(&header, &claims, encoding_key) + .map_err(|e| TokenAcquisitionError::permanent(Error::from(e)))?, + ); + + if let Some(metrics) = &self.metrics { + metrics.record_auth_token_refresh("apns_jwt"); + } + + trace!("Generated new APNs JWT token"); + Ok(MintedToken { + token, + expires_at: SystemTime::now() + TOKEN_LIFETIME, + }) + } } /// APNs client for sending push notifications. pub struct ApnsClient { pub(crate) http_client: Client, pub(crate) config: ApnsConfig, - pub(crate) encoding_key: Option, - pub(crate) cached_token: Arc>>, + pub(crate) token_cache: TokenCache, pub(crate) metrics: Option, + /// Test-only override for the APNs base URL (scheme + host + port). + #[cfg(test)] + pub(crate) test_base_url: Option, } impl ApnsClient { + /// The APNs base URL (scheme + host), overridable in tests. + fn base_url(&self) -> &str { + #[cfg(test)] + if let Some(base) = self.test_base_url.as_deref() { + return base; + } + self.config.base_url() + } + /// Create a new APNs client. #[allow(dead_code)] pub async fn new(config: ApnsConfig) -> Result { @@ -244,84 +309,23 @@ impl ApnsClient { None }; + let token_cache = TokenCache::new(ApnsTokenGenerator { + encoding_key, + team_id: config.team_id.clone(), + key_id: config.key_id.clone(), + metrics: metrics.clone(), + }); + Ok(Self { http_client, config, - encoding_key, - cached_token: Arc::new(RwLock::new(None)), + token_cache, metrics, + #[cfg(test)] + test_base_url: None, }) } - /// Get a valid JWT token, refreshing if necessary. - async fn get_token(&self) -> Result> { - // First check with read lock (fast path) - { - let cached = self.cached_token.read().await; - if let Some(ref token) = *cached - && token.expires_at > SystemTime::now() - { - return Ok(token.token.clone()); - } - } - - // Acquire write lock and double-check to avoid TOCTOU race - let mut cached = self.cached_token.write().await; - if let Some(ref token) = *cached - && token.expires_at > SystemTime::now() - { - return Ok(token.token.clone()); - } - - // Generate and cache new token. The caller gets a short-lived zeroizing - // clone for request construction while the cache owns the reusable copy. - let token = self.generate_token()?; - let cached_token = CachedToken { - token, - expires_at: SystemTime::now() + TOKEN_LIFETIME, - }; - let outbound_token = cached_token.token.clone(); - *cached = Some(cached_token); - - Ok(outbound_token) - } - - /// Generate a new JWT token. - fn generate_token(&self) -> Result> { - let encoding_key = self - .encoding_key - .as_ref() - .ok_or_else(|| Error::Apns("No encoding key configured".to_string()))?; - - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_err(|e| Error::Apns(format!("System time error: {e}")))? - .as_secs(); - - // Backdate `iat` by the clock-skew leeway so a fast host clock does not - // yield an `iat` in APNs's future (403 InvalidProviderToken); `exp` - // stays within Apple's 1-hour max measured from the backdated `iat`. - let (iat, exp) = crate::push::auth_jwt_iat_exp(now, TOKEN_EXPIRATION_SECS); - - let claims = ApnsClaims { - iss: self.config.team_id.clone(), - iat, - exp, - }; - - let mut header = Header::new(Algorithm::ES256); - header.kid = Some(self.config.key_id.clone()); - - let token = Zeroizing::new(encode(&header, &claims, encoding_key)?); - - if let Some(metrics) = &self.metrics { - metrics.record_auth_token_refresh("apns_jwt"); - } - - trace!("Generated new APNs JWT token"); - Ok(token) - } - /// Send a silent push notification to a device. /// /// Returns [`PushSendOutcome::Sent`] if APNs accepted the notification, @@ -339,26 +343,64 @@ impl ApnsClient { device_token: Zeroizing, backoff_permit: Option<&mut crate::push::retry::BackoffPermit>, ) -> Result { - // Validate token format before sending - if !is_valid_device_token(device_token.as_str()) { - trace!( - token_len = device_token.len(), - "Invalid APNs device token format" - ); - return Ok(PushSendOutcome::InvalidToken); - } + // Device-token well-formedness is guaranteed upstream: the dispatcher + // feeds tokens produced by `TokenPayload::device_token_hex()`, which + // always yields even-length lowercase hex within + // `1..=MAX_DEVICE_TOKEN_SIZE` (bounded at decrypt time in + // `crypto/token.rs`). That is the single source of truth for token + // well-formedness (issue #199), so no re-validation happens here. + debug_assert!( + !device_token.is_empty() + && device_token.len().is_multiple_of(2) + && device_token.chars().all(|c| c.is_ascii_hexdigit()), + "device token must be even-length hex; TokenPayload is the single source of truth" + ); + + // Build the request template once, outside the retry closures, so a + // retry does not re-allocate the payload/headers (issue #198). The + // device-token URL is a zeroizing buffer (issue #126); reqwest still + // materializes the serialized body/headers into non-zeroized buffers + // it owns, which is the accepted #126 posture (see build_request). + let request_parts = ApnsRequestParts::for_mode(self.config.payload_mode); + let url = device_token_url(self.base_url(), device_token.as_str()); + + debug!( + payload_mode = %self.config.payload_mode, + push_type = self.config.payload_mode.push_type(), + priority = self.config.payload_mode.priority(), + topic = %self.config.bundle_id, + device_token = redacted_device_token_id(device_token.as_str()), + "Sending APNs notification" + ); let retry_config = RetryConfig::default(); - retry::with_retry( + // Record one duration sample per logical push, measured across all + // retries, so `push_request_duration_seconds` counts notifications + // rather than HTTP attempts (issue #168). Per-attempt HTTP status is + // still recorded in handle_response. + let start = Instant::now(); + let result = retry::with_retry( &retry_config, "APNs", - || self.send_once(&device_token), + || self.send_once(url.as_str(), &request_parts), backoff_permit, self.metrics.as_ref(), ) - .await + .await; + if let Some(metrics) = &self.metrics { + metrics.observe_push_duration("apns", start.elapsed().as_secs_f64()); + } + result } + /// Build the outbound request from the pre-built template. + /// + /// reqwest owns the serialized header/body buffers, so the `bearer` token + /// and JSON payload are materialized into non-zeroized memory it controls. + /// This is the accepted defense-in-depth posture for issue #126: the + /// upstream `Zeroizing` wrapping bounds the token's own copies, but the + /// serialized HTTP request cannot be zeroized without reimplementing the + /// client. fn build_request( &self, url: &str, @@ -374,35 +416,14 @@ impl ApnsClient { .json(&request_parts.payload) } - /// Invalidate the cached token only if it still matches the one the failing - /// request used. - /// - /// Under concurrency another task may have already refreshed the cache with - /// a fresh token between the time this request read its token and the time - /// the authentication rejection came back. Evicting unconditionally would - /// discard that valid token and force redundant JWT regeneration, so the - /// eviction is gated on the cached entry still being the failing token. - async fn invalidate_cached_token(&self, failing_token: &str) { - let mut cached = self.cached_token.write().await; - if cached - .as_ref() - .is_some_and(|token| token.token.as_str() == failing_token) - { - *cached = None; - debug!("Invalidated cached APNs JWT after authentication rejection"); - } - } - async fn handle_response( &self, - request_duration: Duration, response: reqwest::Response, auth_token: &str, ) -> SendAttemptResult { let status = response.status(); if let Some(metrics) = &self.metrics { - metrics.observe_push_duration("apns", request_duration.as_secs_f64()); metrics.record_push_response_status("apns", status.as_u16()); } @@ -447,21 +468,41 @@ impl ApnsClient { } } 403 => { - self.invalidate_cached_token(auth_token).await; + // Parse the reason BEFORE deciding whether to evict. Only a + // genuine provider-token reason (ExpiredProviderToken / + // InvalidProviderToken) means the JWT is the problem; every + // other 403 (BadCertificateEnvironment, Forbidden, ...) is a + // static misconfiguration, and evicting the freshly-minted JWT + // on those would cause a per-notification re-sign stampede + // (issue #145). let error = crate::push::parse_bounded_error_body::(response) .await .unwrap_or(ApnsErrorResponse { reason: "Unknown".to_string(), }); - debug!( - payload_mode = %self.config.payload_mode, - reason = %error.reason, - "APNs authentication error" - ); - SendAttemptResult::Permanent(Error::Apns(format!( - "Authentication error: {}", - error.reason - ))) + let apns_error = Error::Apns(format!("Authentication error: {}", error.reason)); + if is_apns_jwt_reason(&error.reason) { + // The cached JWT is stale/invalid: evict it (gated on it + // still being the failing token) and ask the retry engine + // to retry once with a freshly minted JWT (issue #85). + self.token_cache.invalidate_if_matches(auth_token).await; + debug!( + payload_mode = %self.config.payload_mode, + reason = %error.reason, + "APNs provider-token rejected; will retry once with a fresh JWT" + ); + SendAttemptResult::AuthRejected(apns_error) + } else { + // Non-JWT 403: do not touch the cache, and surface as a + // permanent error so a config fault is not mistaken for a + // recoverable auth rejection. + warn!( + payload_mode = %self.config.payload_mode, + reason = %error.reason, + "APNs rejected request (non-token 403: configuration/environment error)" + ); + SendAttemptResult::Permanent(apns_error) + } } 408 => { // Request timeout - retriable. Honor Retry-After if present. @@ -526,51 +567,41 @@ impl ApnsClient { } } - /// Send a single push notification attempt without retry. + /// Send a single push notification attempt (one `with_retry` iteration). /// - /// Returns a `SendAttemptResult` indicating success, retriable error, or permanent error. - async fn send_once(&self, device_token: &Zeroizing) -> SendAttemptResult { - self.send_once_with_transport_retry_config(device_token, &RetryConfig::transport()) + /// Borrows the request template built once in [`Self::send`]; a transport + /// retry re-sends the same borrowed template rather than rebuilding it + /// (issue #198). Returns a `SendAttemptResult` indicating success, a + /// retriable error, an auth rejection, or a permanent error. + async fn send_once(&self, url: &str, request_parts: &ApnsRequestParts) -> SendAttemptResult { + self.send_once_with_transport_retry_config(url, request_parts, &RetryConfig::transport()) .await } async fn send_once_with_transport_retry_config( &self, - device_token: &Zeroizing, + url: &str, + request_parts: &ApnsRequestParts, transport_retry: &RetryConfig, ) -> SendAttemptResult { - let url = device_token_url(self.config.base_url(), device_token.as_str()); - - debug!( - payload_mode = %self.config.payload_mode, - push_type = self.config.payload_mode.push_type(), - priority = self.config.payload_mode.priority(), - topic = %self.config.bundle_id, - device_token = redacted_device_token_id(device_token.as_str()), - "Sending APNs notification" - ); - - // Add authorization header - let token = match self.get_token().await { + // Mint/read the JWT via the shared cache. A transient acquisition + // failure (none for APNs's pure-CPU signing today, but the shared + // classification allows it) maps to a retriable attempt. + let token = match self.token_cache.get().await { Ok(t) => t, - Err(e) => return SendAttemptResult::Permanent(e), + Err(e) => return e.into_send_attempt(), }; - let request_parts = ApnsRequestParts::for_mode(self.config.payload_mode); - let (request_duration, response) = match retry::with_transport_retry( + let response = match retry::with_transport_retry( transport_retry, "APNs", || async { - // Start timing after JWT acquisition and per transport attempt - // so auth work and prior connect-retry backoff do not inflate - // provider latency. - let start = Instant::now(); - let response = self - .build_request(url.as_str(), token.as_str(), &request_parts) + self.build_request(url, token.as_str(), request_parts) .send() .await - .map_err(Error::from)?; - Ok((start.elapsed(), response)) + // Strip the URL (which embeds the device token) before the + // error can reach any log sink downstream (issue #172). + .map_err(|e| Error::from(e).redact_transport_url()) }, self.metrics.as_ref(), ) @@ -580,8 +611,7 @@ impl ApnsClient { Err(e) => return SendAttemptResult::Permanent(e), }; - self.handle_response(request_duration, response, token.as_str()) - .await + self.handle_response(response, token.as_str()).await } /// Check if the client is properly configured. @@ -591,7 +621,7 @@ impl ApnsClient { return false; } - self.encoding_key.is_some() + self.token_cache.generator().has_encoding_key() && !self.config.key_id.is_empty() && !self.config.team_id.is_empty() && !self.config.bundle_id.is_empty() @@ -602,24 +632,62 @@ impl ApnsClient { impl ApnsClient { /// Create a mock APNs client for testing. pub(crate) fn mock(config: ApnsConfig, with_encoding_key: bool) -> Self { + let encoding_key = with_encoding_key.then(|| EncodingKey::from_secret(b"fake-key")); + let token_cache = TokenCache::new(ApnsTokenGenerator { + encoding_key, + team_id: config.team_id.clone(), + key_id: config.key_id.clone(), + metrics: None, + }); Self { http_client: Client::new(), config, - encoding_key: if with_encoding_key { - Some(EncodingKey::from_secret(b"fake-key")) - } else { - None - }, - cached_token: Arc::new(RwLock::new(None)), + token_cache, metrics: None, + test_base_url: None, } } + + /// Seed the token cache with a credential (test setup). + pub(crate) async fn seed_token(&self, token: &str, expires_at: SystemTime) { + self.token_cache.seed(token, expires_at).await; + } + + /// The currently cached credential value, if any (test inspection). + pub(crate) async fn cached_token_value(&self) -> Option { + self.token_cache.cached_token_value().await + } + + /// Mint a JWT directly through the generator (test inspection). + async fn generate_token(&self) -> Result> { + self.token_cache + .generator() + .mint() + .await + .map(|minted| minted.token) + .map_err(|e| match e { + TokenAcquisitionError::Permanent(err) => err, + TokenAcquisitionError::Retriable { .. } => { + Error::Apns("transient mint failure".to_string()) + } + }) + } + + /// Get a JWT through the cache (test inspection). + async fn get_token(&self) -> Result> { + self.token_cache.get().await.map_err(|e| match e { + TokenAcquisitionError::Permanent(err) => err, + TokenAcquisitionError::Retriable { .. } => { + Error::Apns("transient mint failure".to_string()) + } + }) + } } #[cfg(test)] mod tests { use super::*; - use crate::test_metrics::{counter_value, histogram_sample_count, histogram_sample_sum}; + use crate::test_metrics::{counter_value, histogram_sample_count}; use base64::Engine; use wiremock::matchers::{header, method, path_regex}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -755,130 +823,6 @@ mod tests { ); } - #[test] - fn test_cached_token_stores_jwt_in_zeroizing_string() { - // Compile-time guard: cached credentials must stay zeroizing. - fn assert_zeroizing_string(_: &Zeroizing) {} - - let cached = CachedToken { - token: Zeroizing::new("cached-jwt".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(60), - }; - - assert_zeroizing_string(&cached.token); - } - - #[test] - fn test_valid_device_token() { - // Valid short hex token (lowercase) - assert!(is_valid_device_token("0123456789abcdef")); - - // Valid 64-character hex token (lowercase) - assert!(is_valid_device_token( - "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" - )); - - // Valid longer-than-64-character hex token (uppercase) - assert!(is_valid_device_token( - "0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF01" - )); - - // Valid mixed-case hex token - assert!(is_valid_device_token( - "0123456789AbCdEf0123456789AbCdEf0123456789AbCdEf0123456789AbCdEf" - )); - } - - #[test] - fn test_invalid_device_token_empty() { - assert!(!is_valid_device_token("")); - } - - #[test] - fn test_invalid_device_token_odd_length() { - assert!(!is_valid_device_token("abc")); - } - - #[test] - fn test_invalid_device_token_non_hex_chars() { - // Contains 'g' which is not hex - assert!(!is_valid_device_token( - "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdeg" - )); - - // Contains spaces - assert!(!is_valid_device_token( - "0123456789abcdef 123456789abcdef0123456789abcdef0123456789abcdef" - )); - - // Contains special characters - assert!(!is_valid_device_token( - "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcde!" - )); - - // Contains unicode - assert!(!is_valid_device_token( - "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdéf" - )); - } - - #[tokio::test] - async fn test_send_rejects_invalid_device_token() { - let config = ApnsConfig { - enabled: true, - key_id: "KEYID123".to_string(), - team_id: "TEAMID456".to_string(), - private_key_path: String::new(), - environment: crate::config::ApnsEnvironment::Sandbox, - bundle_id: "com.example.app".to_string(), - payload_mode: Default::default(), - }; - - let client = ApnsClient { - http_client: Client::new(), - config, - encoding_key: Some(EncodingKey::from_secret(b"fake-key")), - cached_token: Arc::new(RwLock::new(Some(CachedToken { - token: Zeroizing::new("test-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }))), - metrics: None, - }; - - // Test with a token that contains non-hex characters - let result = client - .send(zeroizing_token("tooshort"), None) - .await - .unwrap(); - assert_eq!( - result, - PushSendOutcome::InvalidToken, - "Should return invalid-token outcome for token with non-hex characters" - ); - - // Test with token that has invalid characters - let result = client - .send( - zeroizing_token("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdeg"), - None, - ) - .await - .unwrap(); - assert_eq!( - result, - PushSendOutcome::InvalidToken, - "Should return invalid-token outcome for token with invalid characters" - ); - - // Test with empty token - let result = client.send(zeroizing_token(""), None).await.unwrap(); - assert_eq!( - result, - PushSendOutcome::InvalidToken, - "Should return invalid-token outcome for empty token" - ); - } - #[tokio::test] async fn test_send_once_returns_transport_error_after_connect_retries() { let proxy_addr = { @@ -893,32 +837,34 @@ mod tests { .timeout(Duration::from_millis(50)) .build() .unwrap(); - let client = ApnsClient { - http_client, - config: test_config(), - encoding_key: None, - cached_token: Arc::new(RwLock::new(Some(CachedToken { - token: Zeroizing::new("cached-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }))), - metrics: None, - }; + let mut client = ApnsClient::mock(test_config(), false); + client.http_client = http_client; + client + .seed_token( + "cached-token", + SystemTime::now() + Duration::from_secs(3600), + ) + .await; let retry_config = RetryConfig { max_retries: 3, initial_backoff: Duration::from_millis(1), }; + let request_parts = ApnsRequestParts::for_mode(client.config.payload_mode); + let url = device_token_url(client.config.base_url(), "aabbccdd11223344"); let result = client - .send_once_with_transport_retry_config( - &zeroizing_token("aabbccdd11223344"), - &retry_config, - ) + .send_once_with_transport_retry_config(url.as_str(), &request_parts, &retry_config) .await; - assert!(matches!( - result, - SendAttemptResult::Permanent(Error::Http(_)) - )); + // The propagated transport error must not carry the device-token URL + // (issue #172): the retry engine and send path both strip it. + let SendAttemptResult::Permanent(Error::Http(ref http_error)) = result else { + panic!("expected a permanent transport error, got {result:?}"); + }; + assert!( + !http_error.to_string().contains("aabbccdd11223344"), + "device token leaked into transport error: {http_error}" + ); } #[tokio::test] @@ -951,34 +897,17 @@ mod tests { .mount(&mock_server) .await; - let mut config = test_config(); - // Override the environment to use the mock server - config.environment = crate::config::ApnsEnvironment::Sandbox; - let http_client = Client::builder() .timeout(Duration::from_secs(5)) .build() .unwrap(); // Create client with custom HTTP client and pre-populated token cache - let client = ApnsClient { - http_client, - config: ApnsConfig { - enabled: true, - key_id: "KEYID123".to_string(), - team_id: "TEAMID456".to_string(), - private_key_path: String::new(), - environment: crate::config::ApnsEnvironment::Sandbox, - bundle_id: "com.example.app".to_string(), - payload_mode: Default::default(), - }, - encoding_key: None, - cached_token: Arc::new(RwLock::new(Some(CachedToken { - token: Zeroizing::new("test-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }))), - metrics: None, - }; + let mut client = ApnsClient::mock(test_config(), false); + client.http_client = http_client; + client + .seed_token("test-token", SystemTime::now() + Duration::from_secs(3600)) + .await; // We need to override the base_url - let's test with a modified approach // by creating a custom send function that uses the mock URL @@ -1075,7 +1004,11 @@ mod tests { } #[tokio::test] - async fn test_handle_response_records_supplied_provider_duration() { + async fn test_handle_response_records_per_attempt_status_not_duration() { + // Metric layering (issue #168): handle_response runs once per HTTP + // attempt and records only the per-attempt response-status counter. + // Per-logical-push duration is now recorded once in send(), so + // handle_response must NOT observe the duration histogram. let mock_server = MockServer::start().await; Mock::given(method("POST")) .and(path_regex(r"/3/device/[a-f0-9]+")) @@ -1094,27 +1027,78 @@ mod tests { .await .unwrap(); - let provider_duration = Duration::from_millis(123); - let result = client - .handle_response(provider_duration, response, "test-token") - .await; + let result = client.handle_response(response, "test-token").await; assert!(matches!(result, SendAttemptResult::Success(true))); + assert_eq!( + counter_value( + &metrics, + "transponder_push_response_status_total", + &[("platform", "apns"), ("status", "200")] + ), + 1.0 + ); + // No duration sample from handle_response — send() owns that now. The + // HistogramVec has no children until send() observes one, so the + // family is absent from the gathered output entirely. + assert!( + !metrics + .gather() + .iter() + .any(|family| { family.name() == "transponder_push_request_duration_seconds" }), + "handle_response must not observe the per-push duration histogram" + ); + } + + #[tokio::test] + async fn test_send_records_one_duration_sample_per_logical_push_across_retries() { + // A logical push that succeeds after one 429 retry must record exactly + // one duration sample (issue #168) and two per-attempt status counts. + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path_regex(r"/3/device/[a-f0-9]+")) + .respond_with(ResponseTemplate::new(429)) + .up_to_n_times(1) + .expect(1) + .mount(&mock_server) + .await; + Mock::given(method("POST")) + .and(path_regex(r"/3/device/[a-f0-9]+")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&mock_server) + .await; + + let metrics = Metrics::new().unwrap(); + let mut client = ApnsClient::mock(test_config(), true); + client.metrics = Some(metrics.clone()); + client.test_base_url = Some(mock_server.uri()); + client + .seed_token("cached", SystemTime::now() + Duration::from_secs(3600)) + .await; + + let outcome = client + .send(zeroizing_token("aabbccdd11223344"), None) + .await + .unwrap(); + + assert_eq!(outcome, PushSendOutcome::Sent); assert_eq!( histogram_sample_count( &metrics, "transponder_push_request_duration_seconds", &[("platform", "apns")] ), - 1 + 1, + "duration must be sampled once per logical push, not per attempt" ); assert_eq!( - histogram_sample_sum( + counter_value( &metrics, - "transponder_push_request_duration_seconds", - &[("platform", "apns")] + "transponder_push_response_status_total", + &[("platform", "apns"), ("status", "429")] ), - provider_duration.as_secs_f64() + 1.0 ); assert_eq!( counter_value( @@ -1144,9 +1128,7 @@ mod tests { .await .unwrap(); - client - .handle_response(Duration::ZERO, response, "test-token") - .await + client.handle_response(response, "test-token").await } #[tokio::test] @@ -1205,9 +1187,7 @@ mod tests { .await .unwrap(); - client - .handle_response(Duration::ZERO, response, "test-token") - .await + client.handle_response(response, "test-token").await } #[tokio::test] @@ -1297,9 +1277,7 @@ mod tests { .await .unwrap(); - let result = client - .handle_response(Duration::ZERO, response, "test-token") - .await; + let result = client.handle_response(response, "test-token").await; let SendAttemptResult::Permanent(Error::Apns(message)) = result else { panic!("expected a permanent APNs error, got {result:?}"); @@ -1333,9 +1311,7 @@ mod tests { .await .unwrap(); - let result = client - .handle_response(Duration::ZERO, response, "test-token") - .await; + let result = client.handle_response(response, "test-token").await; let SendAttemptResult::Permanent(Error::Apns(message)) = result else { panic!("expected a permanent APNs error, got {result:?}"); @@ -1447,26 +1423,25 @@ mod tests { } #[tokio::test] - async fn test_auth_error_invalidates_cached_token() { + async fn test_jwt_reason_403_evicts_and_asks_for_a_fresh_token_retry() { + // A provider-token 403 (ExpiredProviderToken) means the cached JWT is + // the problem: evict it (issue #145) and return AuthRejected so the + // retry engine retries once with a fresh JWT (issue #85). let mock_server = MockServer::start().await; Mock::given(method("POST")) .and(path_regex(r"/3/device/[a-f0-9]+")) .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({ - "reason": "BadJwtToken" + "reason": "ExpiredProviderToken" }))) .expect(1) .mount(&mock_server) .await; let client = ApnsClient::mock(test_config(), false); - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("poisoned-jwt".to_string()), - expires_at: SystemTime::now() + TOKEN_LIFETIME, - }); - } + client + .seed_token("poisoned-jwt", SystemTime::now() + TOKEN_LIFETIME) + .await; let response = Client::new() .post(format!("{}/3/device/{}", mock_server.uri(), "deadbeef1234")) @@ -1474,16 +1449,55 @@ mod tests { .await .unwrap(); - let result = client - .handle_response(Duration::ZERO, response, "poisoned-jwt") + let result = client.handle_response(response, "poisoned-jwt").await; + + assert!(matches!( + result, + SendAttemptResult::AuthRejected(Error::Apns(ref message)) + if message.contains("ExpiredProviderToken") + )); + assert_eq!(client.cached_token_value().await, None); + } + + #[tokio::test] + async fn test_non_jwt_403_does_not_evict_and_is_permanent() { + // A non-token 403 (BadCertificateEnvironment) is a static + // misconfiguration: the cached JWT must survive (no re-sign stampede, + // issue #145) and the result is a permanent error (not retriable). + let mock_server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path_regex(r"/3/device/[a-f0-9]+")) + .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({ + "reason": "BadCertificateEnvironment" + }))) + .expect(1) + .mount(&mock_server) + .await; + + let client = ApnsClient::mock(test_config(), false); + client + .seed_token("valid-jwt", SystemTime::now() + TOKEN_LIFETIME) .await; + let response = Client::new() + .post(format!("{}/3/device/{}", mock_server.uri(), "deadbeef1234")) + .send() + .await + .unwrap(); + + let result = client.handle_response(response, "valid-jwt").await; + assert!(matches!( result, SendAttemptResult::Permanent(Error::Apns(ref message)) - if message.contains("BadJwtToken") + if message.contains("BadCertificateEnvironment") )); - assert!(client.cached_token.read().await.is_none()); + // The valid JWT must NOT be evicted by a non-token 403. + assert_eq!( + client.cached_token_value().await.as_deref(), + Some("valid-jwt") + ); } #[tokio::test] @@ -1503,13 +1517,9 @@ mod tests { // Simulate a concurrent task having already refreshed the cache to a // newer token after the failing request read its (now stale) token. - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("fresh-jwt".to_string()), - expires_at: SystemTime::now() + TOKEN_LIFETIME, - }); - } + client + .seed_token("fresh-jwt", SystemTime::now() + TOKEN_LIFETIME) + .await; let response = Client::new() .post(format!("{}/3/device/{}", mock_server.uri(), "deadbeef1234")) @@ -1518,20 +1528,17 @@ mod tests { .unwrap(); // The failing request used the older, now-replaced token. - let result = client - .handle_response(Duration::ZERO, response, "stale-jwt") - .await; + let result = client.handle_response(response, "stale-jwt").await; assert!(matches!( result, - SendAttemptResult::Permanent(Error::Apns(ref message)) + SendAttemptResult::AuthRejected(Error::Apns(ref message)) if message.contains("InvalidProviderToken") )); // The freshly refreshed token must survive the stale rejection. - let cached = client.cached_token.read().await; assert_eq!( - cached.as_ref().map(|token| token.token.as_str()), + client.cached_token_value().await.as_deref(), Some("fresh-jwt") ); } @@ -1648,8 +1655,8 @@ mod tests { assert_eq!(response.status(), 500); } - #[test] - fn test_generate_token_no_encoding_key() { + #[tokio::test] + async fn test_generate_token_no_encoding_key() { let config = ApnsConfig { enabled: true, key_id: "KEY123".to_string(), @@ -1661,7 +1668,7 @@ mod tests { }; let client = ApnsClient::mock(config, false); - let result = client.generate_token(); + let result = client.generate_token().await; assert!(result.is_err()); let err = result.unwrap_err(); assert!(err.to_string().contains("No encoding key")); @@ -1682,13 +1689,12 @@ mod tests { let client = ApnsClient::mock(config, true); // Pre-populate the cache - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("cached-test-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }); - } + client + .seed_token( + "cached-test-token", + SystemTime::now() + Duration::from_secs(3600), + ) + .await; // Should return cached token let token = client.get_token().await.unwrap(); @@ -1711,13 +1717,9 @@ mod tests { let client = ApnsClient::mock(config, false); // Pre-populate the cache with an expired token - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("expired-token".to_string()), - expires_at: SystemTime::now() - Duration::from_secs(1), // Already expired - }); - } + client + .seed_token("expired-token", SystemTime::now() - Duration::from_secs(1)) + .await; // Should try to generate new token but fail since no encoding key let result = client.get_token().await; @@ -1739,13 +1741,12 @@ mod tests { let client = ApnsClient::mock(config, false); // No encoding key // Pre-populate the cache with a valid (non-expired) token - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("valid-cached-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), // 1 hour in the future - }); - } + client + .seed_token( + "valid-cached-token", + SystemTime::now() + Duration::from_secs(3600), + ) + .await; // Should return cached token (no encoding key needed since we have valid cache) let token = client.get_token().await.unwrap(); @@ -1765,7 +1766,7 @@ mod tests { }; let client = ApnsClient::new(config).await.unwrap(); - assert!(client.encoding_key.is_none()); + assert!(!client.token_cache.generator().has_encoding_key()); } #[tokio::test] @@ -1811,26 +1812,14 @@ mod tests { .build() .unwrap(); - let config = ApnsConfig { - enabled: true, - key_id: "KEYID123".to_string(), - team_id: "TEAMID456".to_string(), - private_key_path: String::new(), - environment: crate::config::ApnsEnvironment::Sandbox, - bundle_id: "com.example.app".to_string(), - payload_mode: Default::default(), - }; - - let client = ApnsClient { - http_client, - config, - encoding_key: Some(EncodingKey::from_secret(b"fake-key")), - cached_token: Arc::new(RwLock::new(Some(CachedToken { - token: Zeroizing::new("test-cached-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }))), - metrics: None, - }; + let mut client = ApnsClient::mock(test_config(), true); + client.http_client = http_client; + client + .seed_token( + "test-cached-token", + SystemTime::now() + Duration::from_secs(3600), + ) + .await; // The send method uses self.config.base_url() which returns the real APNs URL, // but we can test with direct HTTP calls through the mock @@ -1905,7 +1894,7 @@ OF/2NxApJCzGCEDdfSp6VQO30hyhRANCAAQRWz+jn65BtOMvdyHKcvjBeBSDZH2r }; let client = ApnsClient::new(config).await.unwrap(); - assert!(client.encoding_key.is_some()); + assert!(client.token_cache.generator().has_encoding_key()); assert!(client.is_configured()); } @@ -1946,7 +1935,7 @@ OF/2NxApJCzGCEDdfSp6VQO30hyhRANCAAQRWz+jn65BtOMvdyHKcvjBeBSDZH2r .as_secs(); // Generate a token - let token = client.generate_token().unwrap(); + let token = client.generate_token().await.unwrap(); // Token should be a valid JWT (three dot-separated parts) let parts: Vec<&str> = token.split('.').collect(); @@ -2053,20 +2042,16 @@ OF/2NxApJCzGCEDdfSp6VQO30hyhRANCAAQRWz+jn65BtOMvdyHKcvjBeBSDZH2r let client = ApnsClient::new(config).await.unwrap(); // Cache should be empty initially - { - let cached = client.cached_token.read().await; - assert!(cached.is_none()); - } + assert_eq!(client.cached_token_value().await, None); // Get token - should generate and cache let token1 = client.get_token().await.unwrap(); // Cache should now have a token - { - let cached = client.cached_token.read().await; - assert!(cached.is_some()); - assert_eq!(cached.as_ref().unwrap().token.as_str(), token1.as_str()); - } + assert_eq!( + client.cached_token_value().await.as_deref(), + Some(token1.as_str()) + ); // Get token again - should return cached token (same value) let token2 = client.get_token().await.unwrap(); @@ -2091,26 +2076,11 @@ OF/2NxApJCzGCEDdfSp6VQO30hyhRANCAAQRWz+jn65BtOMvdyHKcvjBeBSDZH2r // Create a client that points to our mock server // We'll need to create a custom client for testing send_once - let config = ApnsConfig { - enabled: true, - key_id: "KEYID123".to_string(), - team_id: "TEAMID456".to_string(), - private_key_path: String::new(), - environment: crate::config::ApnsEnvironment::Sandbox, - bundle_id: "com.example.app".to_string(), - payload_mode: Default::default(), - }; - - let client = ApnsClient { - http_client, - config, - encoding_key: Some(EncodingKey::from_secret(b"fake-key")), - cached_token: Arc::new(RwLock::new(Some(CachedToken { - token: Zeroizing::new("test-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }))), - metrics: None, - }; + let mut client = ApnsClient::mock(test_config(), true); + client.http_client = http_client; + client + .seed_token("test-token", SystemTime::now() + Duration::from_secs(3600)) + .await; // Manually test the response handling logic by making a direct request let url = format!("{}/3/device/{}", mock_server.uri(), "aabbccdd11223344"); diff --git a/src/push/auth.rs b/src/push/auth.rs new file mode 100644 index 0000000..c9703a3 --- /dev/null +++ b/src/push/auth.rs @@ -0,0 +1,489 @@ +//! Shared cached-credential machinery for the push provider clients. +//! +//! APNs and FCM each need a short-lived provider credential (an ES256 JWT and +//! an OAuth2 access token respectively) that is expensive to mint, safe to +//! reuse until expiry, and occasionally rejected by the provider. This module +//! owns the caching/invalidation semantics once so the two clients cannot +//! drift (issue #158): +//! +//! - [`AuthTokenGenerator`] is the provider-specific mint step. +//! - [`TokenCache`] wraps a generator with a read-lock fast path and a +//! single-flight refresh that runs *outside* the cache write lock, so +//! readers keep serving a still-valid token while a refresh is in flight +//! (issue #86) and concurrent misses coalesce into a single mint. +//! - [`TokenCache::invalidate_if_matches`] evicts a provider-rejected +//! credential only while it is still the cached one. *Whether* a rejection +//! is credential-related is decided by the caller: APNs must parse the 403 +//! `reason` first and evict only for provider-token reasons (issue #145), +//! while an FCM 401 is unambiguous and always evicts. + +use std::time::{Duration, SystemTime}; + +use tokio::sync::{Mutex, RwLock}; +use tracing::debug; +use zeroize::{Zeroize, ZeroizeOnDrop, Zeroizing}; + +use crate::error::Error; +use crate::push::retry::SendAttemptResult; + +/// A freshly minted provider credential and the time it should be evicted +/// from the cache. +pub(crate) struct MintedToken { + /// The credential material. + pub(crate) token: Zeroizing, + /// Cache eviction deadline (already including any provider-reported + /// lifetime and safety margin). + pub(crate) expires_at: SystemTime, +} + +/// Outcome of a credential mint attempt, classified for push retry handling. +/// +/// Minting can fail permanently (misconfiguration, credential rejection) or +/// transiently (endpoint 429/5xx or a transport failure); the send path maps +/// the latter to a retriable attempt result instead of dropping the +/// notification (see PR #209 / issue #83). +#[derive(Debug)] +pub(crate) enum TokenAcquisitionError { + /// Configuration, credential, or non-retriable provider rejection. + Permanent(Error), + /// Transient endpoint or transport failure. + Retriable { + /// The HTTP status that triggered the failure (0 for transport + /// errors that never produced a response). + status_code: u16, + /// Optional provider-supplied backpressure hint. + retry_after: Option, + }, +} + +impl TokenAcquisitionError { + /// Build a permanent (non-retriable) acquisition failure. + pub(crate) fn permanent(err: Error) -> Self { + Self::Permanent(err) + } + + /// Convert into the equivalent send-attempt classification. + pub(crate) fn into_send_attempt(self) -> SendAttemptResult { + match self { + Self::Permanent(error) => SendAttemptResult::Permanent(error), + Self::Retriable { + status_code, + retry_after, + } => SendAttemptResult::Retriable { + status_code, + retry_after, + }, + } + } +} + +/// Provider-specific credential mint step. +pub(crate) trait AuthTokenGenerator { + /// Mint a fresh credential. + /// + /// Runs outside the cache write lock, so it may safely perform network + /// IO; [`TokenCache`] guarantees at most one mint per cache is in flight + /// at a time. + async fn mint(&self) -> Result; +} + +/// Cached provider credential. +#[derive(Zeroize, ZeroizeOnDrop)] +struct CachedToken { + token: Zeroizing, + #[zeroize(skip)] + expires_at: SystemTime, +} + +/// Cached-credential store shared by the push clients. +pub(crate) struct TokenCache { + generator: G, + cached: RwLock>, + /// Single-flight refresh guard. + /// + /// A refreshing task holds this mutex — NOT the `cached` write lock — + /// across the mint (which may be a network round-trip), and takes the + /// write lock only briefly to store the result. Readers therefore keep + /// serving a still-valid cached token during a refresh (issue #86), and + /// concurrent cache misses queue here and coalesce onto the refreshed + /// value via the double-check instead of minting redundantly. + refresh: Mutex<()>, +} + +impl TokenCache { + /// Create an empty cache around a provider-specific generator. + pub(crate) fn new(generator: G) -> Self { + Self { + generator, + cached: RwLock::new(None), + refresh: Mutex::new(()), + } + } + + /// The provider-specific generator (for configuration checks). + pub(crate) fn generator(&self) -> &G { + &self.generator + } + + /// Return a clone of the cached credential if it is still valid. + async fn read_valid(&self) -> Option> { + let cached = self.cached.read().await; + cached + .as_ref() + .filter(|token| token.expires_at > SystemTime::now()) + .map(|token| token.token.clone()) + } + + /// Invalidate the cached credential only if it still matches the one the + /// failing request used. + /// + /// Under concurrency another task may have already refreshed the cache + /// with a fresh credential between the time the failing request read its + /// token and the time the provider rejection came back. Evicting + /// unconditionally would discard that valid credential and force a + /// redundant mint, so the eviction is gated on the cached entry still + /// being the failing token. + pub(crate) async fn invalidate_if_matches(&self, failing_token: &str) { + let mut cached = self.cached.write().await; + if cached + .as_ref() + .is_some_and(|token| token.token.as_str() == failing_token) + { + *cached = None; + debug!("Invalidated cached provider credential after authentication rejection"); + } + } +} + +impl TokenCache { + /// Get a valid credential, minting a fresh one if the cache is empty or + /// expired. + /// + /// The caller receives a short-lived zeroizing clone for request + /// construction while the cache owns the reusable copy. + pub(crate) async fn get(&self) -> Result, TokenAcquisitionError> { + // Fast path: serve the cached credential under the read lock. + if let Some(token) = self.read_valid().await { + return Ok(token); + } + + // Slow path: single-flight refresh. Concurrent misses queue on the + // refresh mutex; the mint itself runs without holding the cache + // write lock so readers are never blocked behind network IO. + let _refresh_guard = self.refresh.lock().await; + + // Double-check: a concurrent task may have finished a refresh while + // this one waited for the refresh guard. + if let Some(token) = self.read_valid().await { + return Ok(token); + } + + let minted = self.generator.mint().await?; + let outbound = minted.token.clone(); + let mut cached = self.cached.write().await; + *cached = Some(CachedToken { + token: minted.token, + expires_at: minted.expires_at, + }); + Ok(outbound) + } +} + +#[cfg(test)] +impl TokenCache { + /// Mutable access to the generator for test configuration. + pub(crate) fn generator_mut(&mut self) -> &mut G { + &mut self.generator + } + + /// Seed the cache with a credential (test setup). + pub(crate) async fn seed(&self, token: &str, expires_at: SystemTime) { + let mut cached = self.cached.write().await; + *cached = Some(CachedToken { + token: Zeroizing::new(token.to_owned()), + expires_at, + }); + } + + /// The currently cached credential value, if any (test inspection). + pub(crate) async fn cached_token_value(&self) -> Option { + let cached = self.cached.read().await; + cached.as_ref().map(|token| token.token.as_str().to_owned()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Duration; + use tokio::sync::Semaphore; + + /// Stub generator producing `token-1`, `token-2`, ... with a configurable + /// TTL, an optional gate blocking the mint, and an optional number of + /// initial failures. + struct StubGenerator { + mints: Arc, + gate: Option>, + fail_first: u32, + ttl: Duration, + } + + impl StubGenerator { + fn new() -> Self { + Self { + mints: Arc::new(AtomicU32::new(0)), + gate: None, + fail_first: 0, + ttl: Duration::from_secs(3600), + } + } + } + + impl AuthTokenGenerator for StubGenerator { + async fn mint(&self) -> Result { + if let Some(gate) = &self.gate { + let permit = gate.acquire().await.expect("gate closed"); + permit.forget(); + } + let n = self.mints.fetch_add(1, Ordering::SeqCst) + 1; + if n <= self.fail_first { + return Err(TokenAcquisitionError::permanent(Error::Fcm(format!( + "mint failure {n}" + )))); + } + Ok(MintedToken { + token: Zeroizing::new(format!("token-{n}")), + expires_at: SystemTime::now() + self.ttl, + }) + } + } + + #[test] + fn test_minted_and_cached_tokens_store_credentials_in_zeroizing_strings() { + // Compile-time guard: cached credentials must stay zeroizing. + fn assert_zeroizing_string(_: &Zeroizing) {} + + let minted = MintedToken { + token: Zeroizing::new("minted-credential".to_string()), + expires_at: SystemTime::now(), + }; + let cached = CachedToken { + token: Zeroizing::new("cached-credential".to_string()), + expires_at: SystemTime::now(), + }; + + assert_zeroizing_string(&minted.token); + assert_zeroizing_string(&cached.token); + } + + #[tokio::test] + async fn test_get_mints_once_and_serves_cached_token() { + let generator = StubGenerator::new(); + let mints = generator.mints.clone(); + let cache = TokenCache::new(generator); + + let first = cache.get().await.unwrap(); + let second = cache.get().await.unwrap(); + + assert_eq!(first.as_str(), "token-1"); + assert_eq!(second.as_str(), "token-1"); + assert_eq!(mints.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_get_remints_after_expiry() { + let generator = StubGenerator::new(); + let mints = generator.mints.clone(); + let cache = TokenCache::new(generator); + + cache + .seed("expired-token", SystemTime::now() - Duration::from_secs(1)) + .await; + + let token = cache.get().await.unwrap(); + + assert_eq!(token.as_str(), "token-1"); + assert_eq!(mints.load(Ordering::SeqCst), 1); + assert_eq!(cache.cached_token_value().await.as_deref(), Some("token-1")); + } + + #[tokio::test] + async fn test_concurrent_misses_coalesce_into_a_single_mint() { + // Block the mint behind a gate, fire several concurrent gets, then + // release the gate: exactly one mint must run, and every waiter must + // resolve to that one minted token via the double-check. + let gate = Arc::new(Semaphore::new(0)); + let mut generator = StubGenerator::new(); + generator.gate = Some(gate.clone()); + let mints = generator.mints.clone(); + let cache = Arc::new(TokenCache::new(generator)); + + let mut tasks = Vec::new(); + for _ in 0..5 { + let cache = cache.clone(); + tasks.push(tokio::spawn(async move { + cache.get().await.map(|token| token.as_str().to_owned()) + })); + } + + // Let every task reach either the mint gate or the refresh mutex. + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(mints.load(Ordering::SeqCst), 0, "mint should be gated"); + + // Enough permits for every task to mint if single-flight were broken. + gate.add_permits(5); + + for task in tasks { + let token = task.await.unwrap().unwrap(); + assert_eq!(token, "token-1"); + } + assert_eq!( + mints.load(Ordering::SeqCst), + 1, + "concurrent misses must coalesce into one mint" + ); + } + + #[tokio::test] + async fn test_readers_are_not_blocked_while_a_refresh_is_in_flight() { + // Regression guard for issue #86: the mint must run WITHOUT holding + // the cache write lock. Here a refresh is blocked mid-mint, and both + // a cache write (seed) and a fast-path read must still complete. + let gate = Arc::new(Semaphore::new(0)); + let mut generator = StubGenerator::new(); + generator.gate = Some(gate.clone()); + let mints = generator.mints.clone(); + let cache = Arc::new(TokenCache::new(generator)); + + cache + .seed("expired-token", SystemTime::now() - Duration::from_secs(1)) + .await; + + // Task A misses and blocks inside mint (holding only the refresh + // mutex). + let refresher = { + let cache = cache.clone(); + tokio::spawn(async move { cache.get().await.map(|t| t.as_str().to_owned()) }) + }; + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!refresher.is_finished()); + + // If mint ran under the write lock, this seed (a write) and the + // subsequent read would deadlock until the gate opened. Bound them + // with timeouts to prove they complete during the in-flight refresh. + tokio::time::timeout( + Duration::from_millis(200), + cache.seed("valid-token", SystemTime::now() + Duration::from_secs(3600)), + ) + .await + .expect("cache write must not block behind an in-flight mint"); + + let served = tokio::time::timeout(Duration::from_millis(200), cache.get()) + .await + .expect("fast-path read must not block behind an in-flight mint") + .unwrap(); + assert_eq!(served.as_str(), "valid-token"); + + // Release the refresher; it stores its minted token afterwards. + gate.add_permits(1); + let refreshed = refresher.await.unwrap().unwrap(); + assert_eq!(refreshed, "token-1"); + assert_eq!(mints.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_mint_error_propagates_and_next_get_retries() { + let mut generator = StubGenerator::new(); + generator.fail_first = 1; + let mints = generator.mints.clone(); + let cache = TokenCache::new(generator); + + let error = cache.get().await.unwrap_err(); + assert!(matches!(error, TokenAcquisitionError::Permanent(_))); + assert_eq!(cache.cached_token_value().await, None); + + // A failed mint must not poison the cache: the next get mints again. + let token = cache.get().await.unwrap(); + assert_eq!(token.as_str(), "token-2"); + assert_eq!(mints.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_invalidate_if_matches_evicts_only_the_failing_token() { + let cache = TokenCache::new(StubGenerator::new()); + cache + .seed("fresh-token", SystemTime::now() + Duration::from_secs(3600)) + .await; + + // A stale rejection (for a token another task already replaced) must + // NOT evict the fresh credential. + cache.invalidate_if_matches("stale-token").await; + assert_eq!( + cache.cached_token_value().await.as_deref(), + Some("fresh-token") + ); + + // A rejection for the cached credential itself evicts it. + cache.invalidate_if_matches("fresh-token").await; + assert_eq!(cache.cached_token_value().await, None); + } + + #[tokio::test] + async fn test_invalidate_on_empty_cache_is_a_no_op() { + let cache = TokenCache::new(StubGenerator::new()); + cache.invalidate_if_matches("anything").await; + assert_eq!(cache.cached_token_value().await, None); + } + + #[tokio::test] + async fn test_waiter_reuses_token_minted_by_the_refresh_it_waited_on() { + // Two tasks race a miss. The second must not re-mint after the first + // stores its token: the double-check under the refresh mutex serves + // the fresh value. + let gate = Arc::new(Semaphore::new(0)); + let mut generator = StubGenerator::new(); + generator.gate = Some(gate.clone()); + let mints = generator.mints.clone(); + let cache = Arc::new(TokenCache::new(generator)); + + let first = { + let cache = cache.clone(); + tokio::spawn(async move { cache.get().await.map(|t| t.as_str().to_owned()) }) + }; + tokio::time::sleep(Duration::from_millis(20)).await; + let second = { + let cache = cache.clone(); + tokio::spawn(async move { cache.get().await.map(|t| t.as_str().to_owned()) }) + }; + tokio::time::sleep(Duration::from_millis(20)).await; + + gate.add_permits(2); + + assert_eq!(first.await.unwrap().unwrap(), "token-1"); + assert_eq!(second.await.unwrap().unwrap(), "token-1"); + assert_eq!(mints.load(Ordering::SeqCst), 1); + } + + #[test] + fn test_token_acquisition_error_converts_to_send_attempt() { + let permanent = TokenAcquisitionError::permanent(Error::Apns("bad".to_string())); + assert!(matches!( + permanent.into_send_attempt(), + SendAttemptResult::Permanent(Error::Apns(_)) + )); + + let retriable = TokenAcquisitionError::Retriable { + status_code: 503, + retry_after: Some(Duration::from_secs(5)), + }; + assert!(matches!( + retriable.into_send_attempt(), + SendAttemptResult::Retriable { + status_code: 503, + retry_after: Some(delay), + } if delay == Duration::from_secs(5) + )); + } +} diff --git a/src/push/dispatcher.rs b/src/push/dispatcher.rs index de581e1..6eafc02 100644 --- a/src/push/dispatcher.rs +++ b/src/push/dispatcher.rs @@ -35,6 +35,28 @@ use crate::push::{ApnsClient, FcmClient}; /// Maximum concurrent outbound push requests. const MAX_CONCURRENT_PUSHES: usize = 100; +/// Maximum number of simultaneously *live* send tasks (active plus +/// sleeping-in-backoff), as a multiple of [`MAX_CONCURRENT_PUSHES`]. +/// +/// A send task releases its concurrency permit during a retry backoff sleep +/// (see [`crate::push::retry::BackoffPermit`]) so an active HTTP slot is not +/// wasted while sleeping, but the task stays alive and still holds a decrypted +/// device token. Under a provider 429/5xx storm every in-flight task can enter +/// backoff and release its permit at once, letting the recv loop drain the +/// entire queue and spawn thousands of live token-holding tasks — the +/// concurrency bound silently balloons to the queue size (#160). +/// +/// A second semaphore, acquired at task *spawn* and held for the task's whole +/// life (never released during backoff), caps the live-task count +/// independently. Sizing it at 2x the active-concurrency limit leaves generous +/// headroom for tasks legitimately sleeping in backoff while still bounding +/// decrypted-token residency to a small constant instead of the 10k queue. +const LIVE_TASK_MULTIPLIER: usize = 2; + +/// Maximum number of simultaneously live send tasks. See +/// [`LIVE_TASK_MULTIPLIER`]. +const MAX_LIVE_SEND_TASKS: usize = MAX_CONCURRENT_PUSHES * LIVE_TASK_MULTIPLIER; + /// Maximum number of pending notifications in the queue. /// /// This bounds the memory used by waiting tasks. When this limit is reached, @@ -206,11 +228,36 @@ impl Drop for InFlightGuard { } } +/// Cloneable shared state handed to the queue-draining dispatcher loop and, in +/// turn, to each spawned send task. Groups the fields so the spawn path stays a +/// single argument rather than a long positional list. +#[derive(Clone)] +struct DispatchWorkerContext { + apns_client: Option>, + fcm_client: Option>, + /// Active-HTTP concurrency limit (released during backoff sleeps). + semaphore: Arc, + /// Live-task limit (held for a task's whole life; see [`MAX_LIVE_SEND_TASKS`]). + live_task_semaphore: Arc, + inflight: Arc, + /// Passive per-provider delivery-health signal (see [`DeliveryHealth`]). + delivery_health: Arc, + metrics: Option, +} + /// Push notification dispatcher. pub struct PushDispatcher { apns_client: Option>, fcm_client: Option>, semaphore: Arc, + /// Caps the number of simultaneously live send tasks (see + /// [`MAX_LIVE_SEND_TASKS`]). A permit is acquired at task spawn and held + /// for the task's whole life, so a task sleeping in backoff still counts. + /// + /// The dispatcher loop holds its own clone (via [`DispatchWorkerContext`]); + /// this retained handle lets shutdown/tests observe the live-task bound. + #[cfg_attr(not(test), allow(dead_code))] + live_task_semaphore: Arc, sender: mpsc::Sender, queue_depth: Arc, shutting_down: Arc, @@ -241,6 +288,7 @@ impl PushDispatcher { let apns_client = apns_client.map(Arc::new); let fcm_client = fcm_client.map(Arc::new); let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT_PUSHES)); + let live_task_semaphore = Arc::new(Semaphore::new(MAX_LIVE_SEND_TASKS)); let (sender, receiver) = mpsc::channel(MAX_PENDING_QUEUE_SIZE); let queue_depth = Arc::new(AtomicUsize::new(0)); let shutting_down = Arc::new(AtomicBool::new(false)); @@ -256,21 +304,23 @@ impl PushDispatcher { } // Spawn the queue dispatcher task. - let dispatcher_handle = Self::spawn_dispatcher( - receiver, - queue_depth.clone(), - apns_client.clone(), - fcm_client.clone(), - semaphore.clone(), - inflight.clone(), - delivery_health.clone(), - metrics.clone(), - ); + let worker_context = DispatchWorkerContext { + apns_client: apns_client.clone(), + fcm_client: fcm_client.clone(), + semaphore: semaphore.clone(), + live_task_semaphore: live_task_semaphore.clone(), + inflight: inflight.clone(), + delivery_health: delivery_health.clone(), + metrics: metrics.clone(), + }; + let dispatcher_handle = + Self::spawn_dispatcher(receiver, queue_depth.clone(), worker_context); Self { apns_client, fcm_client, semaphore, + live_task_semaphore, sender, queue_depth, shutting_down, @@ -384,7 +434,11 @@ impl PushDispatcher { // rejections (revoked key, expired credentials), // so they count toward the hard-failure streak. delivery_health.record_hard_failure(platform); - debug!(error = %e, "APNs send error"); + // Redact any embedded URL before logging: the APNs + // URL carries the device token (#172). The send + // path already strips it, but redact again here so + // this log sink is safe regardless. + debug!(error = %e.redact_transport_url(), "APNs send error"); if let Some(ref m) = metrics { m.record_push_failed(platform_str, "error"); } @@ -409,7 +463,10 @@ impl PushDispatcher { // rejections (revoked key, expired credentials), // so they count toward the hard-failure streak. delivery_health.record_hard_failure(platform); - debug!(error = %e, "FCM send error"); + // Uniform with APNs: strip any embedded URL before + // logging (#172). FCM's URL carries no token, but + // keeping the redaction uniform is defense-in-depth. + debug!(error = %e.redact_transport_url(), "FCM send error"); if let Some(ref m) = metrics { m.record_push_failed(platform_str, "error"); } @@ -422,45 +479,62 @@ impl PushDispatcher { } /// Spawn a single dispatcher task that drains the push queue. - // The dispatcher task needs each shared handle individually; bundling - // them into a context struct would only relocate this list. - #[allow(clippy::too_many_arguments)] fn spawn_dispatcher( mut receiver: mpsc::Receiver, queue_depth: Arc, - apns_client: Option>, - fcm_client: Option>, - semaphore: Arc, - inflight: Arc, - delivery_health: Arc, - metrics: Option, + ctx: DispatchWorkerContext, ) -> JoinHandle<()> { tokio::spawn(async move { + let metrics = ctx.metrics.clone(); loop { match receiver.recv().await { Some(PushMessage::Send { platform, token }) => { Self::decrement_queue_depth(&queue_depth); Self::update_queue_size_metric(&metrics, &queue_depth); - // Acquire a permit before spawning the send task. The semaphore, - // not a worker pool, is the outbound concurrency limit. - let permit = - match Self::acquire_push_permit(semaphore.clone(), &metrics).await { + // Bound the number of simultaneously LIVE send tasks + // (active + sleeping-in-backoff) first: this permit is + // held for the task's whole life and is NOT released + // during backoff, so a provider storm can no longer let + // released concurrency permits balloon live + // token-holding tasks to the queue size (#160). The + // recv loop blocks here once MAX_LIVE_SEND_TASKS tasks + // are alive, applying backpressure to the queue instead. + let live_task_permit = + match ctx.live_task_semaphore.clone().acquire_owned().await { Ok(p) => p, Err(_) => { - // The dispatcher never closes this semaphore today; if a - // future change does, the dequeued token is zeroed as this - // loop scope unwinds. - debug!("Push semaphore closed, dispatcher exiting"); + debug!("Live-task semaphore closed, dispatcher exiting"); break; } }; - let apns_client = apns_client.clone(); - let fcm_client = fcm_client.clone(); - let metrics = metrics.clone(); - let semaphore = semaphore.clone(); - let delivery_health = delivery_health.clone(); + // Acquire a concurrency permit before spawning the send + // task. This semaphore, not a worker pool, is the + // outbound *active-HTTP* concurrency limit; the send's + // backoff releases it during sleeps (the live-task + // permit above stays held). + let permit = match Self::acquire_push_permit( + ctx.semaphore.clone(), + &metrics, + ) + .await + { + Ok(p) => p, + Err(_) => { + // The dispatcher never closes this semaphore today; if a + // future change does, the dequeued token is zeroed as this + // loop scope unwinds. + debug!("Push semaphore closed, dispatcher exiting"); + break; + } + }; + + let apns_client = ctx.apns_client.clone(); + let fcm_client = ctx.fcm_client.clone(); + let task_metrics = metrics.clone(); + let semaphore = ctx.semaphore.clone(); + let delivery_health = ctx.delivery_health.clone(); // Register the send task as in-flight BEFORE spawning // so graceful shutdown can never observe a window where @@ -468,12 +542,16 @@ impl PushDispatcher { // the task and dropped only when the task body finishes // (after all retries/backoff), independent of whether the // task currently holds a concurrency permit. - let inflight_guard = inflight.enter(); + let inflight_guard = ctx.inflight.enter(); tokio::spawn(async move { // Decrements the in-flight count when this task // completes, signalling idle to any shutdown waiter. let _inflight_guard = inflight_guard; + // Held for the task's whole life (dropped last), + // bounding live token-holding tasks even across + // backoff sleeps (#160). + let _live_task_permit = live_task_permit; // Wrap the permit so the send's internal backoff // can release the in-flight slot during sleeps and @@ -486,7 +564,7 @@ impl PushDispatcher { token, apns_client, fcm_client, - metrics.clone(), + task_metrics.clone(), &delivery_health, Some(&mut backoff_permit), ) @@ -496,7 +574,7 @@ impl PushDispatcher { // before reading available permits for the metric. drop(backoff_permit); - Self::update_semaphore_available_metric(&metrics, &semaphore); + Self::update_semaphore_available_metric(&task_metrics, &semaphore); }); } Some(PushMessage::Shutdown) => { @@ -741,6 +819,13 @@ impl PushDispatcher { pub fn max_queue_size(&self) -> usize { MAX_PENDING_QUEUE_SIZE } + + /// Available live-task permits (test inspection for the #160 bound). + #[cfg(test)] + #[must_use] + fn available_live_task_permits(&self) -> usize { + self.live_task_semaphore.available_permits() + } } #[cfg(test)] @@ -796,19 +881,16 @@ mod tests { #[tokio::test] async fn send_push_records_invalid_token_metrics() { - use crate::config::{ApnsConfig, FcmConfig}; + // FCM still short-circuits a clearly-malformed (here empty) token to + // the InvalidToken outcome before spending an OAuth round-trip. The + // APNs client no longer re-validates token format (issue #199): the + // dispatcher only ever feeds it even-length hex from + // `TokenPayload::device_token_hex()`, the single source of truth, so + // there is no dispatch-path APNs "invalid_token" case to exercise. + use crate::config::FcmConfig; use crate::metrics::Metrics; - use crate::push::{ApnsClient, FcmClient}; + use crate::push::FcmClient; - let apns_config = ApnsConfig { - enabled: true, - key_id: "KEY123".to_string(), - team_id: "TEAM456".to_string(), - private_key_path: String::new(), - environment: crate::config::ApnsEnvironment::Sandbox, - bundle_id: "com.example.app".to_string(), - payload_mode: Default::default(), - }; let fcm_config = FcmConfig { enabled: true, service_account_path: String::new(), @@ -818,16 +900,6 @@ mod tests { let delivery_health = DeliveryHealth::default(); - PushDispatcher::send_push( - Platform::Apns, - Zeroizing::new("not-a-hex-token".to_string()), - Some(Arc::new(ApnsClient::mock(apns_config, true))), - None, - Some(metrics.clone()), - &delivery_health, - None, - ) - .await; PushDispatcher::send_push( Platform::Fcm, Zeroizing::new("".to_string()), @@ -839,10 +911,6 @@ mod tests { ) .await; - assert_eq!( - push_failed_metric_value(&metrics, "apns", "invalid_token"), - 1 - ); assert_eq!( push_failed_metric_value(&metrics, "fcm", "invalid_token"), 1 @@ -1896,4 +1964,94 @@ mod tests { assert_eq!(attempts.load(Ordering::SeqCst), 2); } + + #[test] + fn test_live_task_bound_is_twice_the_concurrency_limit() { + // The live-task ceiling must be a small constant multiple of the + // active-concurrency limit, not the queue size (#160). + const _: () = assert!(MAX_LIVE_SEND_TASKS == MAX_CONCURRENT_PUSHES * 2); + const _: () = assert!(MAX_LIVE_SEND_TASKS < MAX_PENDING_QUEUE_SIZE); + } + + #[tokio::test] + async fn test_live_task_semaphore_caps_live_tasks_under_backoff_storm() { + // Regression for #160: under a provider 429 storm every in-flight send + // enters backoff and RELEASES its concurrency permit while staying + // alive (still holding a decrypted token). The recv loop would then + // reacquire the freed concurrency permits and drain the whole queue, + // ballooning live token-holding tasks to the queue size. The live-task + // semaphore (acquired at spawn, held across backoff) must cap live + // tasks at MAX_LIVE_SEND_TASKS regardless of how many are queued. + use crate::config::ApnsConfig; + use crate::push::ApnsClient; + use wiremock::matchers::{method, path_regex}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + // Every send gets a 429 with a long Retry-After, so tasks stay in + // backoff (permit released, live-task permit held) for the whole test. + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path_regex(r"/3/device/[a-f0-9]+")) + .respond_with(ResponseTemplate::new(429).insert_header("retry-after", "30")) + .mount(&mock_server) + .await; + + let apns_config = ApnsConfig { + enabled: true, + key_id: "KEY123".to_string(), + team_id: "TEAM456".to_string(), + private_key_path: String::new(), + environment: crate::config::ApnsEnvironment::Sandbox, + bundle_id: "com.example.app".to_string(), + payload_mode: Default::default(), + }; + let mut client = ApnsClient::mock(apns_config, true); + client.test_base_url = Some(mock_server.uri()); + client + .seed_token( + "cached", + std::time::SystemTime::now() + Duration::from_secs(3600), + ) + .await; + let dispatcher = Arc::new(PushDispatcher::new(Some(client), None)); + + // Queue far more messages than the live-task ceiling. + let batch = MAX_LIVE_SEND_TASKS + 300; + assert_eq!( + dispatcher + .dispatch(repeated_apns_payloads(batch)) + .await + .unwrap(), + batch + ); + + // Wait until the live-task semaphore is exhausted (recv loop blocked on + // it), i.e. exactly MAX_LIVE_SEND_TASKS tasks are alive. + for _ in 0..200 { + if dispatcher.available_live_task_permits() == 0 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!( + dispatcher.available_live_task_permits(), + 0, + "live-task semaphore should be fully consumed under the storm" + ); + + // The cap must hold: even though tasks keep releasing their concurrency + // permits during backoff, no more than MAX_LIVE_SEND_TASKS are alive, + // so the remaining messages stay queued rather than spawned. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(dispatcher.available_live_task_permits(), 0); + assert!( + dispatcher.queue_capacity() < MAX_PENDING_QUEUE_SIZE, + "messages beyond the live-task ceiling must stay queued, not spawned" + ); + + // Point the client at nothing further; abandon the storming tasks by + // dropping the dispatcher (test process ends). We do not wait_for + // completion here because the 30s Retry-After would stall the test; + // the invariant under test (bounded live tasks) is already proven. + } } diff --git a/src/push/fcm.rs b/src/push/fcm.rs index b839b53..bc18588 100644 --- a/src/push/fcm.rs +++ b/src/push/fcm.rs @@ -3,19 +3,18 @@ //! Uses service account credentials for OAuth2 authentication. use std::fmt; -use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use jsonwebtoken::{Algorithm, EncodingKey, Header, encode}; use reqwest::Client; use serde::{Deserialize, Serialize}; -use tokio::sync::RwLock; use tracing::{debug, error, trace, warn}; use zeroize::{Zeroize, ZeroizeOnDrop, Zeroizing}; use crate::config::FcmConfig; use crate::error::{Error, Result}; use crate::metrics::Metrics; +use crate::push::auth::{AuthTokenGenerator, MintedToken, TokenAcquisitionError, TokenCache}; use crate::push::retry::{self, PushSendOutcome, RetryConfig, SendAttemptResult}; /// FCM OAuth2 token endpoint. @@ -111,14 +110,6 @@ fn token_cache_lifetime(expires_in: Option) -> Duration { .unwrap_or(TOKEN_LIFETIME) } -/// Cached access token. -#[derive(Zeroize, ZeroizeOnDrop)] -pub(crate) struct CachedToken { - token: Zeroizing, - #[zeroize(skip)] - expires_at: SystemTime, -} - /// FCM message payload. #[derive(Serialize)] struct FcmRequest<'a> { @@ -222,171 +213,104 @@ fn fcm_error_summary(error: &FcmError) -> String { summary } -/// Outcome of an OAuth token refresh attempt for push retry classification. -#[derive(Debug)] -enum TokenAcquisitionError { - /// Configuration, credential, or non-retriable OAuth rejection. - Permanent(Error), - /// Transient OAuth endpoint or transport failure. - Retriable { - status_code: u16, - retry_after: Option, - }, -} - -impl TokenAcquisitionError { - fn permanent(err: Error) -> Self { - Self::Permanent(err) - } - - fn from_http_status(status: reqwest::StatusCode, retry_after: Option) -> Self { - let status_code = status.as_u16(); - if status == 429 || status.is_server_error() { - Self::Retriable { - status_code, - retry_after, - } - } else { - Self::Permanent(Error::Fcm(oauth_failure_message(status))) - } - } - - fn from_transport(err: reqwest::Error) -> Self { - debug!(error = %err, "FCM OAuth token transport error"); - Self::Retriable { - status_code: 0, - retry_after: None, +/// Classify an OAuth token HTTP-error response for retry handling. +/// +/// A `429` or `5xx` from the token endpoint is transient (retriable); every +/// other status is a permanent OAuth rejection carrying only the status text +/// (never the body — see [`oauth_failure_message`]). +fn oauth_error_from_status( + status: reqwest::StatusCode, + retry_after: Option, +) -> TokenAcquisitionError { + if status == 429 || status.is_server_error() { + TokenAcquisitionError::Retriable { + status_code: status.as_u16(), + retry_after, } + } else { + TokenAcquisitionError::permanent(Error::Fcm(oauth_failure_message(status))) } +} - fn into_send_attempt(self) -> SendAttemptResult { - match self { - Self::Permanent(error) => SendAttemptResult::Permanent(error), - Self::Retriable { - status_code, - retry_after, - } => SendAttemptResult::Retriable { - status_code, - retry_after, - }, - } +/// Classify an OAuth token transport failure as a retriable acquisition error. +/// +/// The URL is stripped before logging: while the OAuth endpoint URL carries no +/// device token, keeping the redaction uniform avoids leaking a hijacked +/// `token_uri` and mirrors the send-path posture (issue #172). +fn oauth_error_from_transport(err: reqwest::Error) -> TokenAcquisitionError { + debug!(error = %err.without_url(), "FCM OAuth token transport error"); + TokenAcquisitionError::Retriable { + status_code: 0, + retry_after: None, } } -/// FCM client for sending push notifications. -pub struct FcmClient { - pub(crate) http_client: Client, - pub(crate) config: FcmConfig, - pub(crate) service_account: Option, - pub(crate) encoding_key: Option, - pub(crate) cached_token: Arc>>, - pub(crate) metrics: Option, +/// Maximum bytes read from the OAuth token success body before parsing. +/// +/// An OAuth token response is a small JSON object (an access token plus expiry +/// metadata); a few KiB is ample. Bounding the read keeps a hijacked or +/// misconfigured `token_uri` from streaming an unbounded `200` body into +/// memory (issue #154). Reused via [`crate::push::parse_bounded_json_body`], +/// whose intermediate buffer is zeroized on drop since this body carries a +/// bearer credential. +const MAX_OAUTH_BODY_BYTES: usize = 8 * 1024; + +/// FCM credential generator: mints an OAuth2 access token from the service +/// account via a JWT-bearer grant. +pub(crate) struct FcmTokenGenerator { + http_client: Client, + service_account: Option, + encoding_key: Option, + metrics: Option, /// Test-only override for the OAuth token endpoint URL. #[cfg(test)] - pub(crate) test_oauth_token_url: Option, - /// Test-only override for the FCM v1 API base URL (scheme + host + port). - #[cfg(test)] - pub(crate) test_fcm_api_base_url: Option, + test_oauth_token_url: Option, } -impl FcmClient { - /// Create a new FCM client. - #[allow(dead_code)] - pub async fn new(config: FcmConfig) -> Result { - Self::with_metrics(config, None).await +impl FcmTokenGenerator { + fn service_account(&self) -> Option<&ServiceAccount> { + self.service_account.as_ref() } - /// Create a new FCM client with metrics. - pub async fn with_metrics(config: FcmConfig, metrics: Option) -> Result { - let http_client = Client::builder().timeout(Duration::from_secs(30)).build()?; - - // Load service account if configured - let (service_account, encoding_key) = if !config.service_account_path.is_empty() { - let data = Zeroizing::new( - tokio::fs::read_to_string(&config.service_account_path) - .await - .map_err(|e| { - Error::Fcm(format!( - "Failed to read service account file '{}': {e}", - config.service_account_path - )) - })?, - ); - - let sa: ServiceAccount = serde_json::from_str(data.as_str()) - .map_err(|e| Error::Fcm(format!("Failed to parse service account JSON: {e}")))?; - - let key = EncodingKey::from_rsa_pem(sa.private_key.as_bytes()) - .map_err(|e| Error::Fcm(format!("Failed to parse service account key: {e}")))?; + fn is_configured(&self) -> bool { + self.service_account.is_some() && self.encoding_key.is_some() + } - (Some(sa), Some(key)) + /// The OAuth token endpoint the request is POSTed to. + /// + /// The signed OAuth audience for `sa`: its `token_uri`, falling back to the + /// well-known constant only when `token_uri` is empty. + /// + /// This is the single source of truth for the token endpoint. BOTH the + /// signed assertion `aud` claim and the request target derive from it, so + /// they can never diverge (issue #153) — including the degenerate empty + /// `token_uri` case, where signing `aud=""` while POSTing to the constant + /// would otherwise be an audience mismatch. + fn signed_audience(sa: &ServiceAccount) -> &str { + if sa.token_uri.is_empty() { + OAUTH_TOKEN_URL } else { - (None, None) - }; - - Ok(Self { - http_client, - config, - service_account, - encoding_key, - cached_token: Arc::new(RwLock::new(None)), - metrics, - #[cfg(test)] - test_oauth_token_url: None, - #[cfg(test)] - test_fcm_api_base_url: None, - }) + sa.token_uri.as_str() + } } - fn oauth_token_url(&self) -> &str { + /// The URL the OAuth token request is POSTed to. + /// + /// In production this is always [`Self::signed_audience`], so the request + /// target matches the signed `aud`. A test override redirects only the + /// request target (to a mock server) while `aud` keeps reflecting the real + /// `token_uri`, which is the intended test shape. + fn oauth_token_url<'a>(&'a self, sa: &'a ServiceAccount) -> &'a str { #[cfg(test)] if let Some(url) = self.test_oauth_token_url.as_deref() { return url; } - OAUTH_TOKEN_URL - } - - fn fcm_messages_send_url(&self, project_id: &str) -> String { - #[cfg(test)] - if let Some(base) = self.test_fcm_api_base_url.as_deref() { - return format!("{base}/v1/projects/{project_id}/messages:send"); - } - format!("https://fcm.googleapis.com/v1/projects/{project_id}/messages:send") - } - - /// Get a valid access token, refreshing if necessary. - async fn get_access_token( - &self, - ) -> std::result::Result, TokenAcquisitionError> { - // First check with read lock (fast path) - { - let cached = self.cached_token.read().await; - if let Some(ref token) = *cached - && token.expires_at > SystemTime::now() - { - return Ok(token.token.clone()); - } - } - - // Acquire write lock and double-check to avoid TOCTOU race - let mut cached = self.cached_token.write().await; - if let Some(ref token) = *cached - && token.expires_at > SystemTime::now() - { - return Ok(token.token.clone()); - } - - // Generate new token while holding write lock - let token = self.refresh_token_inner(&mut cached).await?; - - Ok(token) + Self::signed_audience(sa) } +} - /// Refresh the OAuth2 access token, caching it in the provided guard. - async fn refresh_token_inner( - &self, - cached: &mut tokio::sync::RwLockWriteGuard<'_, Option>, - ) -> std::result::Result, TokenAcquisitionError> { +impl AuthTokenGenerator for FcmTokenGenerator { + async fn mint(&self) -> std::result::Result { let sa = self.service_account.as_ref().ok_or_else(|| { TokenAcquisitionError::permanent(Error::Fcm( "No service account configured".to_string(), @@ -412,7 +336,10 @@ impl FcmClient { let claims = OAuthClaims { iss: sa.client_email.clone(), scope: FCM_SCOPE.to_string(), - aud: sa.token_uri.clone(), + // Derive the signed audience from the same helper the request + // target uses, so aud and endpoint can never diverge — even for an + // empty token_uri (issue #153). + aud: Self::signed_audience(sa).to_string(), iat, exp, }; @@ -428,43 +355,125 @@ impl FcmClient { assertion: jwt, }; + // POST to the signed audience (token_uri) so aud and endpoint agree + // (issue #153). let response = self .http_client - .post(self.oauth_token_url()) + .post(self.oauth_token_url(sa)) .form(&request) .send() .await - .map_err(TokenAcquisitionError::from_transport)?; + .map_err(oauth_error_from_transport)?; if !response.status().is_success() { let status = response.status(); let retry_after = retry::retry_after_from_headers(response.headers()); debug!(status = %status, "FCM OAuth token request failed"); - return Err(TokenAcquisitionError::from_http_status(status, retry_after)); + return Err(oauth_error_from_status(status, retry_after)); } - let token_response: TokenResponse = response - .json() - .await - .map_err(|e| TokenAcquisitionError::permanent(Error::from(e)))?; + // Bounded, zeroizing read of the success body (issue #154). + let token_response: TokenResponse = + crate::push::parse_bounded_json_body(response, MAX_OAUTH_BODY_BYTES) + .await + .ok_or_else(|| { + TokenAcquisitionError::permanent(Error::Fcm( + "OAuth token response was unreadable or exceeded the size cap".to_string(), + )) + })?; - // Cache the token while still holding the write lock. The caller gets a - // short-lived zeroizing clone for request construction while the cache - // owns the reusable copy. let ttl = token_cache_lifetime(token_response.expires_in); - let cached_token = CachedToken { - token: token_response.access_token, - expires_at: SystemTime::now() + ttl, - }; - let outbound_token = cached_token.token.clone(); - **cached = Some(cached_token); if let Some(metrics) = &self.metrics { metrics.record_auth_token_refresh("fcm_oauth"); } trace!("Refreshed FCM access token"); - Ok(outbound_token) + Ok(MintedToken { + token: token_response.access_token, + expires_at: SystemTime::now() + ttl, + }) + } +} + +/// FCM client for sending push notifications. +pub struct FcmClient { + pub(crate) config: FcmConfig, + pub(crate) http_client: Client, + pub(crate) token_cache: TokenCache, + pub(crate) metrics: Option, + /// Test-only override for the FCM v1 API base URL (scheme + host + port). + #[cfg(test)] + pub(crate) test_fcm_api_base_url: Option, +} + +impl FcmClient { + /// Create a new FCM client. + #[allow(dead_code)] + pub async fn new(config: FcmConfig) -> Result { + Self::with_metrics(config, None).await + } + + /// Create a new FCM client with metrics. + pub async fn with_metrics(config: FcmConfig, metrics: Option) -> Result { + let http_client = Client::builder().timeout(Duration::from_secs(30)).build()?; + + // Load service account if configured + let (service_account, encoding_key) = if !config.service_account_path.is_empty() { + let data = Zeroizing::new( + tokio::fs::read_to_string(&config.service_account_path) + .await + .map_err(|e| { + Error::Fcm(format!( + "Failed to read service account file '{}': {e}", + config.service_account_path + )) + })?, + ); + + let sa: ServiceAccount = serde_json::from_str(data.as_str()) + .map_err(|e| Error::Fcm(format!("Failed to parse service account JSON: {e}")))?; + + let key = EncodingKey::from_rsa_pem(sa.private_key.as_bytes()) + .map_err(|e| Error::Fcm(format!("Failed to parse service account key: {e}")))?; + + (Some(sa), Some(key)) + } else { + (None, None) + }; + + let token_cache = TokenCache::new(FcmTokenGenerator { + http_client: http_client.clone(), + service_account, + encoding_key, + metrics: metrics.clone(), + #[cfg(test)] + test_oauth_token_url: None, + }); + + Ok(Self { + config, + http_client, + token_cache, + metrics, + #[cfg(test)] + test_fcm_api_base_url: None, + }) + } + + fn fcm_messages_send_url(&self, project_id: &str) -> String { + #[cfg(test)] + if let Some(base) = self.test_fcm_api_base_url.as_deref() { + return format!("{base}/v1/projects/{project_id}/messages:send"); + } + format!("https://fcm.googleapis.com/v1/projects/{project_id}/messages:send") + } + + /// Get a valid access token via the shared cache, refreshing if necessary. + async fn get_access_token( + &self, + ) -> std::result::Result, TokenAcquisitionError> { + self.token_cache.get().await } /// Get the project ID, from config or service account. @@ -472,8 +481,9 @@ impl FcmClient { let project_id = if !self.config.project_id.is_empty() { self.config.project_id.as_str() } else { - self.service_account - .as_ref() + self.token_cache + .generator() + .service_account() .map(|sa| sa.project_id.as_str()) .ok_or_else(|| Error::Fcm("No project ID configured".to_string()))? }; @@ -508,27 +518,53 @@ impl FcmClient { return Ok(PushSendOutcome::InvalidToken); } + // Record one duration sample per logical push, across all retries, so + // push_request_duration_seconds counts notifications rather than HTTP + // attempts (issue #168). Per-attempt HTTP status stays in + // handle_response. let retry_config = RetryConfig::default(); - retry::with_retry( + let start = Instant::now(); + let result = retry::with_retry( &retry_config, "FCM", || self.send_once(&device_token), backoff_permit, self.metrics.as_ref(), ) - .await + .await; + if let Some(metrics) = &self.metrics { + metrics.observe_push_duration("fcm", start.elapsed().as_secs_f64()); + } + result } + /// Build the outbound request from a pre-built `FcmRequest` template. + /// + /// The template (built once in `send_once`, outside the transport-retry + /// closure) borrows the device token from its zeroizing buffer via + /// `FcmMessage<'a>` rather than copying it into an owned `String` + /// (issue #126/#198); a transport retry reuses the same template instead of + /// rebuilding the map/struct/token copy per attempt. reqwest still + /// serializes the token into a non-zeroized body buffer it owns; that + /// residual copy is the accepted #126 posture — the token cannot be + /// zeroized in the serialized HTTP body without reimplementing the client. fn build_request( &self, url: &str, access_token: &str, - device_token: &str, + request: &FcmRequest<'_>, ) -> reqwest::RequestBuilder { + self.http_client + .post(url) + .header("authorization", format!("Bearer {access_token}")) + .json(request) + } + + /// Build the FCM request body template. + fn build_message(device_token: &str) -> FcmRequest<'_> { let mut data = std::collections::HashMap::new(); data.insert("content_available".to_string(), "true".to_string()); - - let request = FcmRequest { + FcmRequest { message: FcmMessage { token: device_token, android: Some(AndroidConfig { @@ -536,30 +572,6 @@ impl FcmClient { }), data: Some(data), }, - }; - - self.http_client - .post(url) - .header("authorization", format!("Bearer {access_token}")) - .json(&request) - } - - /// Invalidate the cached token only if it still matches the one the failing - /// request used. - /// - /// Under concurrency another task may have already refreshed the cache with - /// a fresh token between the time this request read its token and the time - /// the authentication rejection came back. Evicting unconditionally would - /// discard that valid token and force a redundant OAuth round-trip, so the - /// eviction is gated on the cached entry still being the failing token. - async fn invalidate_cached_token(&self, failing_token: &str) { - let mut cached = self.cached_token.write().await; - if cached - .as_ref() - .is_some_and(|token| token.token.as_str() == failing_token) - { - *cached = None; - debug!("Invalidated cached FCM OAuth token after authentication rejection"); } } @@ -622,14 +634,12 @@ impl FcmClient { async fn handle_response( &self, - request_duration: Duration, response: reqwest::Response, access_token: &str, ) -> SendAttemptResult { let status = response.status(); if let Some(metrics) = &self.metrics { - metrics.observe_push_duration("fcm", request_duration.as_secs_f64()); metrics.record_push_response_status("fcm", status.as_u16()); } @@ -643,10 +653,13 @@ impl FcmClient { .await } 401 => { - // Auth error - permanent for this request, refresh on the next one. - self.invalidate_cached_token(access_token).await; - debug!("FCM authentication error"); - SendAttemptResult::Permanent(Error::Fcm("Authentication error".to_string())) + // Auth error. FCM 401 is unambiguously an auth failure (unlike + // APNs 403, which overloads auth and config), so evict the + // cached token unconditionally and ask the retry engine to + // retry once with a freshly minted token (issue #85). + self.token_cache.invalidate_if_matches(access_token).await; + debug!("FCM authentication error; will retry once with a fresh token"); + SendAttemptResult::AuthRejected(Error::Fcm("Authentication error".to_string())) } 403 => { // Project or service-account authorization failure. This is a @@ -706,9 +719,13 @@ impl FcmClient { } } - /// Send a single push notification attempt without retry. + /// Send a single push notification attempt (one `with_retry` iteration). /// - /// Returns a `SendAttemptResult` indicating success, retriable error, or permanent error. + /// Builds the request template once, before the transport-retry closure, + /// so a transport retry reuses it rather than rebuilding the map/struct and + /// re-copying the device token per attempt (issue #198). Returns a + /// `SendAttemptResult` indicating success, a retriable error, an auth + /// rejection, or a permanent error. async fn send_once(&self, device_token: &Zeroizing) -> SendAttemptResult { let project_id = match self.project_id() { Ok(id) => id, @@ -721,21 +738,23 @@ impl FcmClient { Err(e) => return e.into_send_attempt(), }; + // Build the request body template once, outside the transport-retry + // closure (issue #198). It borrows the device token from its zeroizing + // buffer (issue #126). + let request = Self::build_message(device_token.as_str()); + let transport_retry = RetryConfig::transport(); - let (request_duration, response) = match retry::with_transport_retry( + let response = match retry::with_transport_retry( &transport_retry, "FCM", || async { - // Start timing after OAuth acquisition and per transport - // attempt so auth refreshes and prior connect-retry backoff do - // not inflate provider latency. - let start = Instant::now(); - let response = self - .build_request(&url, access_token.as_str(), device_token.as_str()) + self.build_request(&url, access_token.as_str(), &request) .send() .await - .map_err(Error::from)?; - Ok((start.elapsed(), response)) + // FCM's URL does not embed the device token (it lives in + // the JSON body), but strip any URL uniformly with APNs so + // transport errors never carry a target into logs (#172). + .map_err(|e| Error::from(e).redact_transport_url()) }, self.metrics.as_ref(), ) @@ -745,8 +764,7 @@ impl FcmClient { Err(e) => return SendAttemptResult::Permanent(e), }; - self.handle_response(request_duration, response, access_token.as_str()) - .await + self.handle_response(response, access_token.as_str()).await } /// Check if the client is properly configured. @@ -756,7 +774,7 @@ impl FcmClient { return false; } - self.service_account.is_some() && self.encoding_key.is_some() + self.token_cache.generator().is_configured() } } @@ -822,23 +840,54 @@ impl FcmClient { (None, None) }; - Self { - http_client: Client::new(), - config, + let http_client = Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("http client"); + let token_cache = TokenCache::new(FcmTokenGenerator { + http_client: http_client.clone(), service_account, encoding_key, - cached_token: Arc::new(RwLock::new(None)), metrics: None, test_oauth_token_url: None, + }); + + Self { + config, + http_client, + token_cache, + metrics: None, test_fcm_api_base_url: None, } } + + /// Set the mock service account's project id (test setup). + pub(crate) fn set_service_account_project_id(&mut self, project_id: &str) { + if let Some(sa) = self.token_cache.generator_mut().service_account.as_mut() { + sa.project_id = project_id.to_string(); + } + } + + /// Set the OAuth token endpoint override (test setup). + pub(crate) fn set_test_oauth_token_url(&mut self, url: String) { + self.token_cache.generator_mut().test_oauth_token_url = Some(url); + } + + /// Seed the token cache with a credential (test setup). + pub(crate) async fn seed_token(&self, token: &str, expires_at: SystemTime) { + self.token_cache.seed(token, expires_at).await; + } + + /// The currently cached credential value, if any (test inspection). + pub(crate) async fn cached_token_value(&self) -> Option { + self.token_cache.cached_token_value().await + } } #[cfg(test)] mod tests { use super::*; - use crate::test_metrics::{counter_value, histogram_sample_count, histogram_sample_sum}; + use crate::test_metrics::{counter_value, histogram_sample_count}; use wiremock::matchers::{body_partial_json, header, method, path_regex}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -847,16 +896,17 @@ mod tests { } #[test] - fn test_cached_token_stores_access_token_in_zeroizing_string() { - // Compile-time guard: cached credentials must stay zeroizing. + fn test_minted_token_stores_access_token_in_zeroizing_string() { + // Compile-time guard: minted credentials handed to the shared cache + // must stay zeroizing. fn assert_zeroizing_string(_: &Zeroizing) {} - let cached = CachedToken { + let minted = MintedToken { token: Zeroizing::new("cached-access-token".to_string()), expires_at: SystemTime::now() + Duration::from_secs(60), }; - assert_zeroizing_string(&cached.token); + assert_zeroizing_string(&minted.token); } #[test] @@ -1260,7 +1310,10 @@ mod tests { } #[tokio::test] - async fn test_handle_response_records_supplied_provider_duration() { + async fn test_handle_response_records_per_attempt_status_not_duration() { + // Metric layering (issue #168): handle_response records only the + // per-attempt response-status counter; the per-logical-push duration + // is recorded once in send(), so no duration sample comes from here. let mock_server = MockServer::start().await; Mock::given(method("POST")) .and(path_regex(r"/v1/projects/.+/messages:send")) @@ -1287,27 +1340,82 @@ mod tests { .await .unwrap(); - let provider_duration = Duration::from_millis(123); - let result = client - .handle_response(provider_duration, response, "test-access-token") - .await; + let result = client.handle_response(response, "test-access-token").await; assert!(matches!(result, SendAttemptResult::Success(true))); + assert_eq!( + counter_value( + &metrics, + "transponder_push_response_status_total", + &[("platform", "fcm"), ("status", "200")] + ), + 1.0 + ); + assert!( + !metrics + .gather() + .iter() + .any(|family| family.name() == "transponder_push_request_duration_seconds"), + "handle_response must not observe the per-push duration histogram" + ); + } + + #[tokio::test] + async fn test_send_records_one_duration_sample_per_logical_push_across_retries() { + // A logical push that succeeds after one 429 retry records exactly one + // duration sample (issue #168) and two per-attempt status counts. + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path_regex(r"/v1/projects/.+/messages:send")) + .respond_with(ResponseTemplate::new(429).set_body_json(serde_json::json!({ + "error": { "code": 429, "message": "quota", "status": "RESOURCE_EXHAUSTED" } + }))) + .up_to_n_times(1) + .expect(1) + .mount(&mock_server) + .await; + Mock::given(method("POST")) + .and(path_regex(r"/v1/projects/.+/messages:send")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&mock_server) + .await; + + let metrics = Metrics::new().unwrap(); + let config = FcmConfig { + enabled: true, + service_account_path: String::new(), + project_id: "test-project".to_string(), + }; + let mut client = FcmClient::mock(config, true); + client.metrics = Some(metrics.clone()); + client.test_fcm_api_base_url = Some(mock_server.uri()); + client + .seed_token("cached", SystemTime::now() + Duration::from_secs(3600)) + .await; + + let outcome = client + .send(zeroizing_token("device-token-123"), None) + .await + .unwrap(); + + assert_eq!(outcome, PushSendOutcome::Sent); assert_eq!( histogram_sample_count( &metrics, "transponder_push_request_duration_seconds", &[("platform", "fcm")] ), - 1 + 1, + "duration must be sampled once per logical push, not per attempt" ); assert_eq!( - histogram_sample_sum( + counter_value( &metrics, - "transponder_push_request_duration_seconds", - &[("platform", "fcm")] + "transponder_push_response_status_total", + &[("platform", "fcm"), ("status", "429")] ), - provider_duration.as_secs_f64() + 1.0 ); assert_eq!( counter_value( @@ -1349,9 +1457,7 @@ mod tests { .await .unwrap(); - client - .handle_response(Duration::ZERO, response, "test-access-token") - .await + client.handle_response(response, "test-access-token").await } /// Drive `handle_response` against a mock server returning `status_code` @@ -1402,9 +1508,7 @@ mod tests { .await .unwrap(); - client - .handle_response(Duration::ZERO, response, "test-access-token") - .await + client.handle_response(response, "test-access-token").await } #[tokio::test] @@ -1534,9 +1638,7 @@ mod tests { .await .unwrap(); - client - .handle_response(Duration::ZERO, response, "test-access-token") - .await + client.handle_response(response, "test-access-token").await } #[tokio::test] @@ -1678,13 +1780,9 @@ mod tests { project_id: "test-project".to_string(), }; let client = FcmClient::mock(config, false); - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("poisoned-access-token".to_string()), - expires_at: SystemTime::now() + TOKEN_LIFETIME, - }); - } + client + .seed_token("poisoned-access-token", SystemTime::now() + TOKEN_LIFETIME) + .await; let response = Client::new() .post(format!( @@ -1696,15 +1794,17 @@ mod tests { .unwrap(); let result = client - .handle_response(Duration::ZERO, response, "poisoned-access-token") + .handle_response(response, "poisoned-access-token") .await; + // FCM 401 is unambiguous auth failure: evict the token and ask for a + // fresh-token retry (issue #85). assert!(matches!( result, - SendAttemptResult::Permanent(Error::Fcm(ref message)) + SendAttemptResult::AuthRejected(Error::Fcm(ref message)) if message.contains("Authentication error") )); - assert!(client.cached_token.read().await.is_none()); + assert_eq!(client.cached_token_value().await, None); } #[tokio::test] @@ -1733,13 +1833,9 @@ mod tests { // Simulate a concurrent task having already refreshed the cache to a // newer token after the failing request read its (now stale) token. - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("fresh-access-token".to_string()), - expires_at: SystemTime::now() + TOKEN_LIFETIME, - }); - } + client + .seed_token("fresh-access-token", SystemTime::now() + TOKEN_LIFETIME) + .await; let response = Client::new() .post(format!( @@ -1751,20 +1847,17 @@ mod tests { .unwrap(); // The failing request used the older, now-replaced token. - let result = client - .handle_response(Duration::ZERO, response, "stale-access-token") - .await; + let result = client.handle_response(response, "stale-access-token").await; assert!(matches!( result, - SendAttemptResult::Permanent(Error::Fcm(ref message)) + SendAttemptResult::AuthRejected(Error::Fcm(ref message)) if message.contains("Authentication error") )); // The freshly refreshed token must survive the stale rejection. - let cached = client.cached_token.read().await; assert_eq!( - cached.as_ref().map(|token| token.token.as_str()), + client.cached_token_value().await.as_deref(), Some("fresh-access-token") ); } @@ -1860,7 +1953,7 @@ mod tests { }; let mut client = FcmClient::mock(config, true); - client.service_account.as_mut().unwrap().project_id = "service-project".to_string(); + client.set_service_account_project_id("service-project"); assert_eq!(client.project_id().unwrap(), "service-project"); } @@ -1926,8 +2019,9 @@ mod tests { }; let client = FcmClient::new(config).await.unwrap(); - assert!(client.service_account.is_none()); - assert!(client.encoding_key.is_none()); + // No service account / encoding key loaded means the client is not + // configured to send. + assert!(!client.is_configured()); } #[tokio::test] @@ -1957,13 +2051,12 @@ mod tests { let client = FcmClient::mock(config, true); // Pre-populate the cache - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("cached-access-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }); - } + client + .seed_token( + "cached-access-token", + SystemTime::now() + Duration::from_secs(3600), + ) + .await; // Should return cached token let token = client.get_access_token().await.unwrap(); @@ -2114,13 +2207,9 @@ mod tests { let client = FcmClient::mock(config, false); // Pre-populate with expired token - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("expired-token".to_string()), - expires_at: SystemTime::now() - Duration::from_secs(1), - }); - } + client + .seed_token("expired-token", SystemTime::now() - Duration::from_secs(1)) + .await; // Should try to refresh but fail since no service account let result = client.get_access_token().await; @@ -2158,13 +2247,9 @@ mod tests { let client = FcmClient::mock(config, false); // Pre-populate cache so it doesn't fail on get_access_token first - { - let mut cached = client.cached_token.write().await; - *cached = Some(CachedToken { - token: Zeroizing::new("test-token".to_string()), - expires_at: SystemTime::now() + Duration::from_secs(3600), - }); - } + client + .seed_token("test-token", SystemTime::now() + Duration::from_secs(3600)) + .await; let result = client .send(zeroizing_token("test-device-token"), None) @@ -2204,7 +2289,7 @@ mod tests { }; let mut client = FcmClient::mock(config, true); - client.service_account.as_mut().unwrap().project_id = "x/../../admin/v1".to_string(); + client.set_service_account_project_id("x/../../admin/v1"); let result = client .send(zeroizing_token("test-device-token"), None) @@ -2249,17 +2334,8 @@ mod tests { project_id: "test-project".to_string(), }; - // Create client with no service account and no encoding key - let client = FcmClient { - http_client: Client::new(), - config, - service_account: None, - encoding_key: None, - cached_token: Arc::new(RwLock::new(None)), - metrics: None, - test_oauth_token_url: None, - test_fcm_api_base_url: None, - }; + // Create client with no service account and no encoding key. + let client = FcmClient::mock(config, false); // Should not be configured - no encoding key assert!(!client.is_configured()); @@ -2604,17 +2680,23 @@ LTP/MQIxLydQxT4+jx2NBu0= let encoding_key = EncodingKey::from_rsa_pem(TEST_RSA_PRIVATE_KEY.as_bytes()).expect("test RSA key"); - FcmClient { - http_client: Client::builder() - .timeout(Duration::from_secs(5)) - .build() - .expect("http client"), - config, + let http_client = Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("http client"); + let token_cache = TokenCache::new(FcmTokenGenerator { + http_client: http_client.clone(), service_account: Some(sa), encoding_key: Some(encoding_key), - cached_token: Arc::new(RwLock::new(None)), metrics: None, test_oauth_token_url: None, + }); + + FcmClient { + config, + http_client, + token_cache, + metrics: None, test_fcm_api_base_url: None, } } @@ -2623,7 +2705,7 @@ LTP/MQIxLydQxT4+jx2NBu0= fn test_oauth_token_http_status_classifies_transient_failures_as_retriable() { for status_code in [429_u16, 500, 503] { let status = reqwest::StatusCode::from_u16(status_code).unwrap(); - let failure = TokenAcquisitionError::from_http_status(status, None); + let failure = oauth_error_from_status(status, None); assert!( matches!( failure, @@ -2640,7 +2722,7 @@ LTP/MQIxLydQxT4+jx2NBu0= #[test] fn test_oauth_token_http_status_classifies_permanent_oauth_failures() { let status = reqwest::StatusCode::BAD_REQUEST; - let failure = TokenAcquisitionError::from_http_status(status, None); + let failure = oauth_error_from_status(status, None); assert!(matches!( failure, TokenAcquisitionError::Permanent(Error::Fcm(ref message)) @@ -2651,8 +2733,7 @@ LTP/MQIxLydQxT4+jx2NBu0= #[test] fn test_oauth_token_http_status_honors_retry_after_for_503() { let status = reqwest::StatusCode::SERVICE_UNAVAILABLE; - let failure = - TokenAcquisitionError::from_http_status(status, Some(Duration::from_secs(30))); + let failure = oauth_error_from_status(status, Some(Duration::from_secs(30))); assert!(matches!( failure, TokenAcquisitionError::Retriable { @@ -2673,7 +2754,7 @@ LTP/MQIxLydQxT4+jx2NBu0= .await; let mut client = mock_client_with_rsa_service_account("test-project"); - client.test_oauth_token_url = Some(mock_server.uri()); + client.set_test_oauth_token_url(mock_server.uri()); let result = client.get_access_token().await; assert!(matches!( @@ -2685,6 +2766,160 @@ LTP/MQIxLydQxT4+jx2NBu0= )); } + /// Build an FCM token generator whose service-account `token_uri` is + /// `token_uri` and with no OAuth-URL test override, so `mint()` POSTs to + /// the signed audience (issue #153). + fn generator_with_token_uri(token_uri: &str) -> FcmTokenGenerator { + let sa = ServiceAccount { + account_type: "service_account".to_string(), + project_id: "test-project".to_string(), + private_key: Zeroizing::new(TEST_RSA_PRIVATE_KEY.to_string()), + client_email: "test@test.iam.gserviceaccount.com".to_string(), + token_uri: token_uri.to_string(), + }; + let encoding_key = + EncodingKey::from_rsa_pem(TEST_RSA_PRIVATE_KEY.as_bytes()).expect("test RSA key"); + FcmTokenGenerator { + http_client: Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("http client"), + service_account: Some(sa), + encoding_key: Some(encoding_key), + metrics: None, + test_oauth_token_url: None, + } + } + + #[tokio::test] + async fn test_mint_posts_to_service_account_token_uri() { + // The OAuth request must target the service-account token_uri so the + // signed assertion `aud` and the endpoint agree (issue #153). The mock + // only answers on /custom/token; a POST to the constant would 404. + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path_regex(r"^/custom/token$")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "minted-token", + "expires_in": 3600 + }))) + .expect(1) + .mount(&mock_server) + .await; + + let generator = generator_with_token_uri(&format!("{}/custom/token", mock_server.uri())); + let minted = generator + .mint() + .await + .expect("mint should target token_uri"); + assert_eq!(minted.token.as_str(), "minted-token"); + } + + #[tokio::test] + async fn test_mint_falls_back_to_constant_when_token_uri_empty() { + // With an empty token_uri the generator must fall back to the well-known + // constant rather than POSTing to an empty URL (issue #153). + let generator = generator_with_token_uri(""); + let sa = generator.service_account().unwrap(); + assert_eq!(generator.oauth_token_url(sa), OAUTH_TOKEN_URL); + } + + #[test] + fn test_signed_aud_and_post_target_agree_for_standard_and_empty_token_uri() { + // The signed assertion `aud` and the production POST target must be the + // SAME URL, or Google rejects the token request as an audience mismatch + // (issue #153). Assert the invariant directly (no test-URL override, so + // oauth_token_url resolves to the production target) for both a + // non-standard token_uri and the degenerate empty case. + for token_uri in ["https://oauth2.example.test/token", ""] { + let generator = generator_with_token_uri(token_uri); + let sa = generator.service_account().unwrap(); + let aud = FcmTokenGenerator::signed_audience(sa); + let post_target = generator.oauth_token_url(sa); + assert_eq!( + aud, post_target, + "signed aud must equal the POST target for token_uri {token_uri:?}" + ); + } + + // And the empty case resolves both to the well-known constant. + let empty = generator_with_token_uri(""); + let sa = empty.service_account().unwrap(); + assert_eq!(FcmTokenGenerator::signed_audience(sa), OAUTH_TOKEN_URL); + } + + #[tokio::test] + async fn test_mint_signs_aud_matching_post_target_for_empty_token_uri() { + // End-to-end proof for the degenerate empty-token_uri case: with an + // empty token_uri and NO test override, mint() POSTs to OAUTH_TOKEN_URL + // and must sign aud=OAUTH_TOKEN_URL (not aud=""). We cannot hit the real + // Google endpoint here, so assert the two derivations agree — the mint + // path reads aud and target from the same helper, so they are identical + // by construction, and this locks that in against regressions. + let generator = generator_with_token_uri(""); + let sa = generator.service_account().unwrap(); + + let aud = FcmTokenGenerator::signed_audience(sa); + let post_target = generator.oauth_token_url(sa); + assert_eq!(aud, OAUTH_TOKEN_URL); + assert_eq!(post_target, OAUTH_TOKEN_URL); + assert_eq!(aud, post_target); + } + + #[tokio::test] + async fn test_mint_honors_provider_expires_in() { + // The cached lifetime must derive from the provider's expires_in (minus + // the safety margin), not the hardcoded fallback (issue #88 groundwork, + // preserved through the refactor). + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "minted-token", + "expires_in": 120 + }))) + .expect(1) + .mount(&mock_server) + .await; + + let generator = generator_with_token_uri(&mock_server.uri()); + let minted = generator.mint().await.expect("mint"); + // expires_at is set to now()+ttl inside mint, where ttl is the provider + // 120s expiry minus the 60s safety margin (= 60s). Measuring the + // remaining lifetime just after mint returns should land near 60s (a + // little less, by the elapsed time since expires_at was stamped), and + // well under the 120s the provider reported. + let remaining = minted + .expires_at + .duration_since(SystemTime::now()) + .expect("expiry in the future"); + assert!(remaining <= Duration::from_secs(60)); + assert!(remaining >= Duration::from_secs(50)); + } + + #[tokio::test] + async fn test_mint_rejects_oversized_oauth_success_body() { + // A hijacked/misconfigured token_uri returning a multi-megabyte 200 + // body must be refused by the bounded read rather than buffered whole + // (issue #154). + let mock_server = MockServer::start().await; + let padding = "A".repeat(MAX_OAUTH_BODY_BYTES + 1024); + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_string(format!( + r#"{{"access_token":"{padding}","expires_in":3600}}"# + ))) + .expect(1) + .mount(&mock_server) + .await; + + let generator = generator_with_token_uri(&mock_server.uri()); + let result = generator.mint().await; + let Err(TokenAcquisitionError::Permanent(Error::Fcm(message))) = result else { + panic!("expected a permanent OAuth error for the oversized body"); + }; + assert!(message.contains("unreadable or exceeded the size cap")); + assert!(!message.contains(&padding)); + } + #[tokio::test] async fn test_get_access_token_oauth_transport_error_is_retriable() { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); @@ -2692,7 +2927,7 @@ LTP/MQIxLydQxT4+jx2NBu0= drop(listener); let mut client = mock_client_with_rsa_service_account("test-project"); - client.test_oauth_token_url = Some(format!("http://{addr}/token")); + client.set_test_oauth_token_url(format!("http://{addr}/token")); let result = client.get_access_token().await; assert!(matches!( @@ -2718,7 +2953,7 @@ LTP/MQIxLydQxT4+jx2NBu0= .await; let mut client = mock_client_with_rsa_service_account("test-project"); - client.test_oauth_token_url = Some(mock_server.uri()); + client.set_test_oauth_token_url(mock_server.uri()); let result = client.get_access_token().await; let TokenAcquisitionError::Permanent(Error::Fcm(message)) = result.unwrap_err() else { @@ -2760,7 +2995,7 @@ LTP/MQIxLydQxT4+jx2NBu0= .await; let mut client = mock_client_with_rsa_service_account("test-project"); - client.test_oauth_token_url = Some(mock_server.uri()); + client.set_test_oauth_token_url(mock_server.uri()); client.test_fcm_api_base_url = Some(mock_server.uri()); let result = client.send(zeroizing_token("device-token-123"), None).await; @@ -2785,7 +3020,7 @@ LTP/MQIxLydQxT4+jx2NBu0= .await; let mut client = mock_client_with_rsa_service_account("test-project"); - client.test_oauth_token_url = Some(mock_server.uri()); + client.set_test_oauth_token_url(mock_server.uri()); client.test_fcm_api_base_url = Some(mock_server.uri()); let result = client.send(zeroizing_token("device-token-123"), None).await; diff --git a/src/push/mod.rs b/src/push/mod.rs index 8d2e670..31803db 100644 --- a/src/push/mod.rs +++ b/src/push/mod.rs @@ -1,6 +1,7 @@ //! Push notification clients and dispatching. pub mod apns; +pub mod auth; pub mod dispatcher; pub mod fcm; pub mod retry; @@ -32,33 +33,37 @@ const MAX_ERROR_BODY_BYTES: usize = 8 * 1024; /// usable token lifetime. const AUTH_JWT_CLOCK_SKEW_LEEWAY_SECS: u64 = 30; -/// Read at most [`MAX_ERROR_BODY_BYTES`] of a provider error response and -/// deserialize the bounded prefix as JSON. +/// Read at most `max_bytes` of a response body and deserialize the bounded +/// prefix as JSON. /// /// If the server declares a body larger than the cap, the helper returns before /// reading it. Otherwise the body is streamed chunk by chunk and reading stops /// as soon as the cap is reached, so the full body is never buffered. Returns /// `None` when the body is unreadable, exceeds the cap, or does not parse (e.g. /// an empty or non-JSON body, or a JSON document truncated by the cap), letting -/// callers fall back to their default status-derived classification. -async fn parse_bounded_error_body( +/// callers fall back to their default classification. +/// +/// The intermediate buffer is zeroized on drop because some callers parse +/// credential-bearing bodies through it (the FCM OAuth token success body). +async fn parse_bounded_json_body( mut response: reqwest::Response, + max_bytes: usize, ) -> Option { if response .content_length() - .is_some_and(|len| len > MAX_ERROR_BODY_BYTES as u64) + .is_some_and(|len| len > max_bytes as u64) { return None; } - let mut body = Vec::new(); + let mut body = zeroize::Zeroizing::new(Vec::new()); loop { match response.chunk().await { Ok(Some(chunk)) => { - if body.len() + chunk.len() > MAX_ERROR_BODY_BYTES { - // The body is larger than any legitimate provider error - // payload; stop reading and fall back to the default. + if body.len() + chunk.len() > max_bytes { + // The body is larger than any legitimate payload for this + // call site; stop reading and fall back to the default. return None; } body.extend_from_slice(&chunk); @@ -71,6 +76,12 @@ async fn parse_bounded_error_body( serde_json::from_slice(&body).ok() } +/// Read at most [`MAX_ERROR_BODY_BYTES`] of a provider error response and +/// deserialize the bounded prefix as JSON. See [`parse_bounded_json_body`]. +async fn parse_bounded_error_body(response: reqwest::Response) -> Option { + parse_bounded_json_body(response, MAX_ERROR_BODY_BYTES).await +} + /// Compute the `iat`/`exp` claims for a provider auth JWT from the current Unix /// time `now_secs`. /// diff --git a/src/push/retry.rs b/src/push/retry.rs index b2bcd49..cb35406 100644 --- a/src/push/retry.rs +++ b/src/push/retry.rs @@ -74,6 +74,18 @@ pub const DEFAULT_INITIAL_BACKOFF: Duration = Duration::from_millis(100); /// Maximum backoff duration cap. pub const MAX_BACKOFF: Duration = Duration::from_secs(10); +/// Upper bound on an honored provider-supplied `Retry-After` value. +/// +/// The header is provider/network-controlled input. Without a ceiling, a +/// misbehaving or compromised endpoint could pin a send task — which holds a +/// decrypted device token and an in-flight guard that blocks graceful-shutdown +/// drain — in a backoff sleep for an arbitrary duration (issue #162). 60 +/// seconds comfortably covers genuine provider backpressure (APNs/FCM hints +/// are typically single-digit seconds) while bounding how long an untrusted +/// header can pin a task. Deliberately larger than [`MAX_BACKOFF`], which only +/// caps locally-computed exponential backoff. +pub const MAX_RETRY_AFTER: Duration = Duration::from_secs(60); + /// Minimum server-requested backoff duration. const MIN_RETRY_BACKOFF: Duration = DEFAULT_INITIAL_BACKOFF; @@ -137,6 +149,16 @@ pub enum SendAttemptResult { /// Optional Retry-After duration from the response header. retry_after: Option, }, + /// The provider rejected the request's auth credential (an APNs `403` + /// with a provider-token `reason`, or an FCM `401`). + /// + /// The client has already invalidated the cached credential, so the next + /// attempt will mint a fresh one. [`with_retry`] allows exactly one + /// immediate retry per logical push for this variant (issue #85): a + /// transient rejection (clock skew, key rotation window) recovers with + /// the fresh credential, while a genuinely bad key fails again and + /// surfaces the carried error instead of looping. + AuthRejected(crate::error::Error), /// Permanent error that should not be retried. Permanent(crate::error::Error), } @@ -166,6 +188,11 @@ where { let mut retries = 0; let mut backoff = config.initial_backoff; + // One-shot budget for retrying after a provider auth rejection with a + // freshly minted credential (issue #85). Tracked separately from the + // transient-retry budget so an auth recovery does not consume (or get + // starved by) 429/5xx retries. + let mut auth_retry_used = false; // Reborrow across loop iterations: `Option<&mut T>` is not `Copy`. let mut backoff_permit = backoff_permit; @@ -173,6 +200,29 @@ where match operation().await { SendAttemptResult::Success(true) => return Ok(PushSendOutcome::Sent), SendAttemptResult::Success(false) => return Ok(PushSendOutcome::InvalidToken), + SendAttemptResult::AuthRejected(error) if !auth_retry_used => { + auth_retry_used = true; + + if let Some(metrics) = metrics { + metrics.record_push_retry(service_name.to_lowercase().as_str()); + } + + // Retry immediately: the rejection is not backpressure, and + // the client already evicted the rejected credential, so the + // next attempt mints a fresh one. + warn!( + service = service_name, + error = %error, + "Provider rejected auth credential; retrying once with a freshly minted token" + ); + } + SendAttemptResult::AuthRejected(error) => { + warn!( + service = service_name, + "Provider rejected a freshly minted auth credential; not retrying again" + ); + return Err(error); + } SendAttemptResult::Retriable { status_code, retry_after, @@ -184,9 +234,10 @@ where } // Honor provider-supplied Retry-After values, but floor zero - // or tiny values and jitter every sleep so concurrent failures - // do not retry in synchronized waves. Only cap locally-computed - // exponential backoff to avoid runaway delays. + // or tiny values, cap at MAX_RETRY_AFTER (untrusted input), + // and jitter every sleep so concurrent failures do not retry + // in synchronized waves. Locally-computed exponential backoff + // is capped separately at MAX_BACKOFF. let wait_duration = retry_sleep_duration(retry_after, backoff); warn!( @@ -236,7 +287,11 @@ fn retry_sleep_base(retry_after: Option, backoff: Duration) -> (Durati let fallback = backoff.min(MAX_BACKOFF); match retry_after { Some(Duration::ZERO) => (MIN_RETRY_BACKOFF.max(fallback), true), - Some(server_delay) => (server_delay.max(MIN_RETRY_BACKOFF), true), + // Honor the provider hint, but clamp it: floored so a tiny value + // cannot produce hot-loop retries, and capped so an untrusted header + // cannot pin the task (and its decrypted token) arbitrarily long + // (issue #162). + Some(server_delay) => (server_delay.clamp(MIN_RETRY_BACKOFF, MAX_RETRY_AFTER), true), None => (fallback, false), } } @@ -318,6 +373,12 @@ where metrics.record_push_retry(service_name.to_lowercase().as_str()); } + // A reqwest error can embed the request URL, and the APNs URL + // contains the raw device token; strip it before the error can + // reach any log sink (issue #172). The push clients already + // strip at the conversion site, but this keeps the retry + // engine safe regardless of caller discipline. + let error = error.without_url(); warn!( service = service_name, error = %error, @@ -331,6 +392,9 @@ where backoff = (backoff * 2).min(MAX_BACKOFF); } Err(Error::Http(error)) if should_retry_transport(&error) => { + // Strip any URL (which may embed a device token) before both + // logging and propagating the exhausted error (issue #172). + let error = error.without_url(); warn!( service = service_name, error = %error, @@ -341,8 +405,10 @@ where } // Non-connect transport errors are not retriable: the provider may // already have accepted the (non-idempotent) request, so return - // immediately without retrying. - Err(error) => return Err(error), + // immediately without retrying. The URL is stripped so downstream + // logging of the propagated error cannot leak a device token + // (issue #172). + Err(error) => return Err(error.redact_transport_url()), } } } @@ -590,6 +656,213 @@ mod tests { assert_duration_between(sleep, MAX_BACKOFF - MAX_RETRY_JITTER, MAX_BACKOFF); } + #[test] + fn test_retry_sleep_duration_caps_untrusted_retry_after() { + // A hostile/misbehaving provider sends `Retry-After: 86400`. The + // honored delay must clamp to MAX_RETRY_AFTER (plus at most one + // additive jitter window), not pin the task for a day (issue #162). + let sleep = retry_sleep_duration( + Some(Duration::from_secs(86_400)), + Duration::from_millis(100), + ); + + assert_duration_between(sleep, MAX_RETRY_AFTER, MAX_RETRY_AFTER + MAX_RETRY_JITTER); + } + + #[test] + fn test_retry_sleep_duration_honors_retry_after_at_cap_boundary() { + // A Retry-After exactly at the cap is honored unchanged. + let sleep = retry_sleep_duration(Some(MAX_RETRY_AFTER), Duration::from_millis(100)); + + assert_duration_between(sleep, MAX_RETRY_AFTER, MAX_RETRY_AFTER + MAX_RETRY_JITTER); + } + + #[tokio::test] + async fn test_with_retry_auth_rejected_retries_once_then_succeeds() { + // First attempt: provider rejects the auth credential. The retry + // engine must immediately retry once (fresh credential), and the + // second attempt's success must resolve the push as Sent (issue #85). + let config = RetryConfig::default(); + let attempt_count = Arc::new(AtomicU32::new(0)); + let attempt_count_clone = attempt_count.clone(); + + let start = std::time::Instant::now(); + let result = with_retry( + &config, + "test", + || { + let count = attempt_count_clone.clone(); + async move { + let attempts = count.fetch_add(1, Ordering::SeqCst) + 1; + if attempts == 1 { + SendAttemptResult::AuthRejected(crate::error::Error::Apns( + "Authentication error: ExpiredProviderToken".to_string(), + )) + } else { + SendAttemptResult::Success(true) + } + } + }, + None, + None, + ) + .await; + + assert_eq!(result.unwrap(), PushSendOutcome::Sent); + assert_eq!(attempt_count.load(Ordering::SeqCst), 2); + // The auth retry is immediate: no backoff sleep is inserted. + assert!(start.elapsed() < Duration::from_secs(1)); + } + + #[tokio::test] + async fn test_with_retry_auth_rejected_twice_returns_error() { + // A second rejection means the freshly minted credential was also + // rejected (genuinely bad key): surface the error instead of looping. + let config = RetryConfig::default(); + let attempt_count = Arc::new(AtomicU32::new(0)); + let attempt_count_clone = attempt_count.clone(); + + let result = with_retry( + &config, + "test", + || { + let count = attempt_count_clone.clone(); + async move { + count.fetch_add(1, Ordering::SeqCst); + SendAttemptResult::AuthRejected(crate::error::Error::Fcm( + "Authentication error".to_string(), + )) + } + }, + None, + None, + ) + .await; + + assert!(matches!(result, Err(Error::Fcm(ref message)) + if message.contains("Authentication error"))); + assert_eq!(attempt_count.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_with_retry_auth_retry_budget_is_separate_from_transient_budget() { + // An auth retry must not consume the transient 429/5xx budget: after + // the one-shot auth recovery, the full transient budget still applies. + let config = RetryConfig { + max_retries: 2, + initial_backoff: Duration::from_millis(1), + }; + let attempt_count = Arc::new(AtomicU32::new(0)); + let attempt_count_clone = attempt_count.clone(); + + let result = with_retry( + &config, + "test", + || { + let count = attempt_count_clone.clone(); + async move { + let attempts = count.fetch_add(1, Ordering::SeqCst) + 1; + match attempts { + 1 => SendAttemptResult::AuthRejected(crate::error::Error::Apns( + "Authentication error: ExpiredProviderToken".to_string(), + )), + 2 | 3 => SendAttemptResult::Retriable { + status_code: 503, + retry_after: None, + }, + _ => SendAttemptResult::Success(true), + } + } + }, + None, + None, + ) + .await; + + // 1 auth-rejected attempt + 1 fresh attempt + 2 transient retries. + assert_eq!(result.unwrap(), PushSendOutcome::Sent); + assert_eq!(attempt_count.load(Ordering::SeqCst), 4); + } + + #[tokio::test] + async fn test_with_transport_retry_redacts_url_from_exhausted_connect_error() { + // The APNs request URL embeds the raw device token. When transport + // retries exhaust, the propagated (and logged) error must not carry + // the URL (issue #172). + let config = RetryConfig { + max_retries: 1, + initial_backoff: Duration::from_millis(1), + }; + let client = Client::new(); + + let result: crate::error::Result = with_transport_retry( + &config, + "test", + || { + let client = client.clone(); + async move { + let error = client + .post("http://127.0.0.1:9/3/device/aabbccddeeff00112233") + .send() + .await + .unwrap_err(); + assert!(error.is_connect()); + // Deliberately convert WITHOUT stripping the URL to prove + // the retry engine redacts even without caller discipline. + Err(Error::from(error)) + } + }, + None, + ) + .await; + + let error = result.unwrap_err(); + let rendered = error.to_string(); + assert!(matches!(error, Error::Http(_))); + assert!( + !rendered.contains("aabbccddeeff00112233"), + "device token leaked into transport error display: {rendered}" + ); + } + + #[tokio::test] + async fn test_with_transport_retry_redacts_url_from_non_retriable_error() { + // Non-connect transport errors return through the terminal arm; the + // URL (which may embed a device token) must be stripped there too. + let config = RetryConfig { + max_retries: 3, + initial_backoff: Duration::from_millis(1), + }; + let client = Client::new(); + + let result: crate::error::Result = with_transport_retry( + &config, + "test", + || { + let client = client.clone(); + async move { + let error = client + .post("ftp://example.invalid/3/device/aabbccddeeff00112233") + .send() + .await + .unwrap_err(); + assert!(!error.is_connect()); + Err(Error::from(error)) + } + }, + None, + ) + .await; + + let error = result.unwrap_err(); + let rendered = error.to_string(); + assert!(matches!(error, Error::Http(_))); + assert!( + !rendered.contains("aabbccddeeff00112233"), + "device token leaked into transport error display: {rendered}" + ); + } + #[tokio::test] async fn test_with_retry_success_first_attempt() { let config = RetryConfig::default();