hennery_kernel/
delivery.rs

1//! Web Push delivery (kernel spec §6; plan 10b-ii): drain the notice queue
2//! (`push::Notices`), apply each notice's hat policy, and send it to every
3//! subscription of the owner's, encrypted to the browser (RFC 8291) and
4//! signed with the collector's VAPID key (RFC 8292).
5//!
6//! Requests go out through a `Transport`. The collector's is the shared
7//! egress client, public addresses only (kernel spec §7.1, plan 8b): `spawn`
8//! takes it from an `Egress` itself, so no caller can hand delivery another.
9//! Tests use a fake one, or an `InternalNetwork` client to a push service on
10//! loopback.
11
12use crate::egress::{Allowance, Egress, EgressClient, EgressError, Method, Request, header};
13use crate::hosts::Hosts;
14use crate::operator::Operator;
15use crate::push::{
16    Notice, Notices, PushPolicy, Subscription, Urgency, VAPID_TOKEN_SECS, VapidKey, audience, parse_keys, subject,
17};
18use crate::secret::unix_now;
19use anyhow::Result;
20use base64::Engine;
21use base64::engine::general_purpose::URL_SAFE_NO_PAD;
22use hennery_proto::rest::PushPayload;
23use std::collections::HashMap;
24use std::future::Future;
25use std::pin::Pin;
26use std::sync::{Arc, Mutex};
27use std::time::Duration;
28
29/// How long a push service keeps a notification it could not deliver yet
30/// (kernel spec §6: one hour).
31pub const PUSH_TTL_SECS: u64 = 60 * 60;
32
33/// The plaintext is padded with spaces to a multiple of this, so its size
34/// tells the push service little of the title (plan 10a's review).
35pub const PAD_TO: usize = 512;
36
37/// The most plaintext one push carries: RFC 8291's 4096-byte body, less the
38/// 86-byte header, the 16-byte tag and the padding delimiter.
39pub const MAX_PLAINTEXT: usize = 4096 - 86 - 16 - 1;
40
41/// Subscriptions sent to at once, per notice.
42pub const MAX_IN_FLIGHT: usize = 8;
43
44/// One push, ready to send.
45#[derive(Clone)]
46pub struct PushRequest {
47    pub endpoint: url::Url,
48    /// `TTL`, `Urgency`, `Content-Encoding`, `Content-Type` and
49    /// `Authorization`. Never `Topic`: the push service would see it.
50    pub headers: Vec<(&'static str, String)>,
51    /// The encrypted payload (`aes128gcm`).
52    pub body: Vec<u8>,
53}
54
55impl std::fmt::Debug for PushRequest {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        f.debug_struct("PushRequest")
58            .field("endpoint_host", &self.endpoint.host_str())
59            .field("body_len", &self.body.len())
60            .finish_non_exhaustive()
61    }
62}
63
64/// What came of one request.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub enum Sent {
67    /// The push service answered.
68    Status {
69        code: u16,
70        /// Its `Retry-After`, when it gave one in seconds.
71        retry_after: Option<Duration>,
72    },
73    /// The egress policy refused the address: nothing was sent.
74    Refused,
75    /// No answer in time.
76    Timeout,
77    /// The connection failed.
78    Failed,
79}
80
81/// Sends one push request. The collector's sends through the egress client
82/// (`PublicOnly`); never one that reaches a private address.
83pub trait Transport: Send + Sync + 'static {
84    fn send(&self, request: PushRequest) -> Pin<Box<dyn Future<Output = Sent> + Send + '_>>;
85}
86
87/// The egress client sends a push as any other request (kernel spec §7.1):
88/// its policy refuses a non-public address before anything is sent, and an
89/// error never carries the endpoint.
90impl Transport for EgressClient {
91    fn send(&self, push: PushRequest) -> Pin<Box<dyn Future<Output = Sent> + Send + '_>> {
92        Box::pin(async move {
93            let mut request = Request::new(Method::POST, push.endpoint);
94            for (name, value) in &push.headers {
95                let (Ok(name), Ok(value)) = (
96                    header::HeaderName::from_bytes(name.as_bytes()),
97                    header::HeaderValue::from_str(value),
98                ) else {
99                    return Sent::Failed;
100                };
101                request.headers_mut().insert(name, value);
102            }
103            *request.body_mut() = Some(push.body.into());
104            match EgressClient::send(self, request).await {
105                Ok(response) => Sent::Status {
106                    code: response.status().as_u16(),
107                    retry_after: retry_after(response.headers()),
108                },
109                Err(EgressError::Refused(_)) => Sent::Refused,
110                Err(EgressError::Timeout) => Sent::Timeout,
111                Err(EgressError::Http(_)) => Sent::Failed,
112            }
113        })
114    }
115}
116
117/// `Retry-After` in seconds (RFC 9110 §10.2.3); an HTTP date, or anything
118/// else, is no hint.
119fn retry_after(headers: &header::HeaderMap) -> Option<Duration> {
120    let secs = headers
121        .get(header::RETRY_AFTER)?
122        .to_str()
123        .ok()?
124        .trim()
125        .parse::<u64>()
126        .ok()?;
127    Some(Duration::from_secs(secs))
128}
129
130/// Start the collector's delivery: a queue, and a task draining it through
131/// `egress`'s public-only client (kernel spec §6, §7.1: Web Push never
132/// reaches a private address). The queue's sending end is `AppState::push`.
133pub fn spawn(hosts: Arc<Hosts>, operator: Arc<Operator>, vapid: Arc<VapidKey>, egress: &Egress) -> crate::push::Push {
134    let (push, notices) = crate::push::Push::new();
135    let delivery = Delivery::new(
136        hosts,
137        operator,
138        vapid,
139        egress.client(Allowance::PublicOnly),
140        RetryPolicy::default(),
141    );
142    tokio::spawn(delivery.run(notices));
143    push
144}
145
146/// When to try again, and how long to wait.
147#[derive(Debug, Clone, PartialEq, Eq)]
148pub struct RetryPolicy {
149    /// The wait before each retry: as many retries as there are waits.
150    pub delays: Vec<Duration>,
151    /// The longest `Retry-After` honoured; longer is cut to it.
152    pub max_retry_after: Duration,
153    /// How long one try may take before it counts as unanswered.
154    pub attempt_timeout: Duration,
155    /// The most one subscription's tries and waits may take together: a
156    /// retry that could overrun it is not made (10b-ii's review, A1).
157    /// Notices are delivered one after another, so a slow or silent push
158    /// service must not hold the others up for minutes.
159    pub budget: Duration,
160}
161
162impl Default for RetryPolicy {
163    /// Two retries, after 1 s and 4 s, `Retry-After` up to 10 s; each try
164    /// 10 s at most, and 30 s in all.
165    fn default() -> Self {
166        Self {
167            delays: vec![Duration::from_secs(1), Duration::from_secs(4)],
168            max_retry_after: Duration::from_secs(10),
169            attempt_timeout: Duration::from_secs(10),
170            budget: Duration::from_secs(30),
171        }
172    }
173}
174
175/// What became of one subscription's push.
176#[derive(Debug, Clone, PartialEq, Eq)]
177pub enum Outcome {
178    /// The owning module withdrew this work before transport admission.
179    Suppressed,
180    /// The push service took it (2xx).
181    Delivered,
182    /// The subscription is gone (404, 410): removed.
183    Removed,
184    /// Not delivered; the reason is the subscription's `last_error`.
185    Failed(String),
186}
187
188/// The notice's payload under `policy`, or `None` for a muted hat (kernel
189/// spec §6; `Notice`'s documented rules):
190/// - the title: the notice's generic title under `generic_title`, else its
191///   title;
192/// - the body: the notice's `detail` when `details` is on and it has one,
193///   with `generic_title` too; otherwise none under `generic_title`, else
194///   the notice's body.
195pub fn payload_for(notice: &Notice, policy: PushPolicy) -> Option<PushPayload> {
196    if policy.muted {
197        return None;
198    }
199    let body = match (&notice.detail, policy.details) {
200        (Some(detail), true) => detail.clone(),
201        _ if policy.generic_title => String::new(),
202        _ => notice.body.clone(),
203    };
204    let title = if policy.generic_title {
205        notice.generic_title.clone()
206    } else {
207        notice.title.clone()
208    };
209    Some(PushPayload {
210        title,
211        body,
212        url: notice.url.clone(),
213        tag: notice.tag.clone(),
214    })
215}
216
217/// `payload` as JSON, padded with trailing spaces (still valid JSON) to the
218/// next multiple of `PAD_TO`, at most `MAX_PLAINTEXT`. `Err` for one that
219/// cannot fit, which the notices' caps rule out.
220pub fn padded(payload: &PushPayload) -> Result<Vec<u8>> {
221    let mut json = serde_json::to_vec(payload)?;
222    anyhow::ensure!(json.len() <= MAX_PLAINTEXT, "a push payload of {} bytes", json.len());
223    let target = json.len().div_ceil(PAD_TO).max(1) * PAD_TO;
224    json.resize(target.min(MAX_PLAINTEXT), b' ');
225    Ok(json)
226}
227
228/// Delivers notices: see the module docs.
229pub struct Delivery<T> {
230    hosts: Arc<Hosts>,
231    operator: Arc<Operator>,
232    vapid: Arc<VapidKey>,
233    transport: T,
234    retry: RetryPolicy,
235    /// `Authorization` headers by (audience, subject), with when each was
236    /// made: reused until an hour before it expires (plan 10a's
237    /// re-confirmation, N2).
238    tokens: Mutex<Tokens>,
239}
240
241/// `Authorization` headers by (audience, subject), each with when it was
242/// made: one token per push service, reused until an hour before it
243/// expires, and dropped once it has.
244#[derive(Default)]
245struct Tokens(HashMap<(String, Option<String>), (String, i64)>);
246
247impl Tokens {
248    fn authorization(&mut self, vapid: &VapidKey, endpoint: &url::Url, subject: Option<&str>, now: i64) -> String {
249        let key = (audience(endpoint), subject.map(str::to_string));
250        if let Some((header, made)) = self.0.get(&key)
251            && now - made < VAPID_TOKEN_SECS - 60 * 60
252        {
253            return header.clone();
254        }
255        self.0.retain(|_, (_, made)| now - *made < VAPID_TOKEN_SECS);
256        let header = vapid.authorization(endpoint, subject, now);
257        self.0.insert(key, (header.clone(), now));
258        header
259    }
260}
261
262impl<T: Transport> Delivery<T> {
263    pub fn new(
264        hosts: Arc<Hosts>,
265        operator: Arc<Operator>,
266        vapid: Arc<VapidKey>,
267        transport: T,
268        retry: RetryPolicy,
269    ) -> Arc<Self> {
270        Arc::new(Self {
271            hosts,
272            operator,
273            vapid,
274            transport,
275            retry,
276            tokens: Mutex::new(Tokens::default()),
277        })
278    }
279
280    /// Deliver every notice queued, one after another, until the process
281    /// ends. A notice that cannot be delivered is logged; the next one is
282    /// tried all the same.
283    pub async fn run(self: Arc<Self>, mut notices: Notices) {
284        loop {
285            let notice = notices.recv().await;
286            if let Err(err) = self.deliver_guarded(&notice, notices.guard()).await {
287                tracing::warn!(tag = %notice.tag, error = %err, "push not delivered");
288            }
289        }
290    }
291
292    /// Deliver `notice` to every subscription, at most `MAX_IN_FLIGHT` at
293    /// once: each subscription's outcome, by its id. Expired subscriptions
294    /// are removed first and get nothing; a muted hat gets nothing.
295    pub async fn deliver(self: &Arc<Self>, notice: &Notice) -> Result<Vec<(String, Outcome)>> {
296        self.deliver_guarded(notice, Arc::new(|_| true)).await
297    }
298
299    /// Eligibility is checked before payload construction and each transport attempt.
300    /// An attempt admitted before revocation may complete; external copies cannot be recalled.
301    pub async fn deliver_guarded(
302        self: &Arc<Self>,
303        notice: &Notice,
304        guard: crate::push::NoticeGuard,
305    ) -> Result<Vec<(String, Outcome)>> {
306        if !guard(notice) {
307            return Ok(Vec::new());
308        }
309        let now = unix_now();
310        self.hosts.prune_expired_subscriptions(now)?;
311        let Some(payload) = payload_for(notice, self.hosts.push_policy(&notice.hat_id)?) else {
312            return Ok(Vec::new());
313        };
314        let plaintext = Arc::new(padded(&payload)?);
315        let subject = match self.operator.public_url() {
316            Some(public_url) => subject(self.operator.contact()?.as_deref(), &public_url),
317            None => None,
318        };
319        let urgency = match notice.urgency {
320            Urgency::High => "high",
321            Urgency::Normal => "normal",
322        };
323        let permits = Arc::new(tokio::sync::Semaphore::new(MAX_IN_FLIGHT));
324        let mut sending = tokio::task::JoinSet::new();
325        for sub in self.hosts.subscriptions()? {
326            let this = self.clone();
327            let plaintext = plaintext.clone();
328            let subject = subject.clone();
329            let permits = permits.clone();
330            let guard = guard.clone();
331            let notice = notice.clone();
332            sending.spawn(async move {
333                let _permit = permits.acquire_owned().await;
334                let outcome = if guard(&notice) {
335                    this.send_one(&sub, &plaintext, urgency, subject.as_deref(), &notice, &guard)
336                        .await
337                } else {
338                    Outcome::Suppressed
339                };
340                (sub.id, outcome)
341            });
342        }
343        let mut outcomes = Vec::new();
344        // A task that panicked is logged; the others' outcomes are still
345        // collected and recorded (10b-ii's review, A2).
346        while let Some(done) = sending.join_next().await {
347            match done {
348                Ok(outcome) => outcomes.push(outcome),
349                // Not the panic's message, which could hold what it unwrapped.
350                Err(err) => tracing::error!(task = %err.id(), panicked = err.is_panic(), "a push task failed"),
351            }
352        }
353        Ok(outcomes)
354    }
355
356    /// One subscription's push, retried as `retry` says; its outcome is
357    /// recorded on it.
358    async fn send_one(
359        &self,
360        sub: &Subscription,
361        plaintext: &[u8],
362        urgency: &str,
363        subject: Option<&str>,
364        notice: &Notice,
365        guard: &crate::push::NoticeGuard,
366    ) -> Outcome {
367        let outcome = match self.request(sub, plaintext, urgency, subject) {
368            Ok(request) => self.send_retrying(request, notice, guard).await,
369            Err(err) => {
370                tracing::warn!(id = %sub.id, error = %err, "push not built");
371                Outcome::Failed("could not encrypt to this device".into())
372            }
373        };
374        let recorded = match &outcome {
375            Outcome::Suppressed => Ok(()),
376            Outcome::Delivered => self.hosts.record_push_success(&sub.id, &sub.endpoint, unix_now()),
377            Outcome::Removed => self.hosts.remove_gone_subscription(&sub.id, &sub.endpoint).map(|_| ()),
378            Outcome::Failed(why) => self.hosts.record_push_error(&sub.id, &sub.endpoint, why),
379        };
380        if let Err(err) = recorded {
381            tracing::warn!(id = %sub.id, error = %err, "push outcome not recorded");
382        }
383        match &outcome {
384            Outcome::Removed => {
385                tracing::info!(id = %sub.id, host = %sub.endpoint_host(), "push subscription gone; removed")
386            }
387            Outcome::Failed(why) => tracing::warn!(id = %sub.id, host = %sub.endpoint_host(), %why, "push failed"),
388            Outcome::Delivered | Outcome::Suppressed => {}
389        }
390        outcome
391    }
392
393    /// The push for `sub`: the plaintext encrypted to its keys, and the
394    /// headers.
395    fn request(
396        &self,
397        sub: &Subscription,
398        plaintext: &[u8],
399        urgency: &str,
400        subject: Option<&str>,
401    ) -> Result<PushRequest> {
402        let endpoint = url::Url::parse(&sub.endpoint)?;
403        let (p256dh, auth) = parse_keys(&sub.p256dh, &sub.auth).map_err(anyhow::Error::msg)?;
404        let ua_public = p256::PublicKey::from_sec1_bytes(&URL_SAFE_NO_PAD.decode(p256dh)?)?;
405        let auth: [u8; 16] = URL_SAFE_NO_PAD
406            .decode(auth)?
407            .try_into()
408            .map_err(|_| anyhow::anyhow!("keys.auth is not 16 bytes"))?;
409        let auth = web_push_native::Auth::from(auth);
410        let body =
411            web_push_native::encrypt(plaintext.to_vec(), &ua_public, &auth).map_err(|e| anyhow::anyhow!("{e}"))?;
412        Ok(PushRequest {
413            headers: vec![
414                ("TTL", PUSH_TTL_SECS.to_string()),
415                ("Urgency", urgency.to_string()),
416                ("Content-Encoding", "aes128gcm".to_string()),
417                ("Content-Type", "application/octet-stream".to_string()),
418                ("Authorization", self.authorization(&endpoint, subject)),
419            ],
420            endpoint,
421            body,
422        })
423    }
424
425    /// The `Authorization` header for `endpoint`'s push service, made once
426    /// per audience and subject and reused until an hour before it expires.
427    fn authorization(&self, endpoint: &url::Url, subject: Option<&str>) -> String {
428        let mut tokens = self.tokens.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
429        tokens.authorization(&self.vapid, endpoint, subject, unix_now())
430    }
431
432    /// Send `request`, retrying a 429, a 5xx, a timeout or a failed
433    /// connection after each of `retry`'s delays (or the service's
434    /// `Retry-After`, if longer, up to its cap). Any other answer is final.
435    /// Each try has `attempt_timeout`, and no retry is made that could not
436    /// finish within `budget` of the first.
437    async fn send_retrying(&self, request: PushRequest, notice: &Notice, guard: &crate::push::NoticeGuard) -> Outcome {
438        let started = tokio::time::Instant::now();
439        let mut delays = self.retry.delays.iter();
440        loop {
441            if !guard(notice) {
442                return Outcome::Suppressed;
443            }
444            let sent = tokio::time::timeout(self.retry.attempt_timeout, self.transport.send(request.clone()))
445                .await
446                .unwrap_or(Sent::Timeout);
447            let (retryable, failed, retry_after) = match sent {
448                Sent::Status { code: 200..=299, .. } => return Outcome::Delivered,
449                Sent::Status { code: 404 | 410, .. } => return Outcome::Removed,
450                Sent::Status { code, retry_after } if code == 429 || code >= 500 => {
451                    (true, format!("the push service answered {code}"), retry_after)
452                }
453                Sent::Status { code, .. } => (false, format!("the push service refused it ({code})"), None),
454                Sent::Refused => (
455                    false,
456                    "refused: the push service is not at a public address".into(),
457                    None,
458                ),
459                Sent::Timeout => (true, "the push service did not answer in time".into(), None),
460                Sent::Failed => (true, "the push service could not be reached".into(), None),
461            };
462            match delays.next() {
463                Some(delay) if retryable => {
464                    let wait = retry_after.map_or(*delay, |after| after.min(self.retry.max_retry_after).max(*delay));
465                    if started.elapsed() + wait + self.retry.attempt_timeout > self.retry.budget {
466                        return Outcome::Failed(format!("{failed}; gave up within the time allowed"));
467                    }
468                    tokio::time::sleep(wait).await;
469                }
470                _ => return Outcome::Failed(failed),
471            }
472        }
473    }
474}
475
476#[cfg(test)]
477mod tests {
478    use super::*;
479
480    fn notice() -> Notice {
481        Notice {
482            hat_id: "hat-1".into(),
483            urgency: Urgency::High,
484            title: "Fix the flaky test".into(),
485            generic_title: "Session needs your answer".into(),
486            body: "needs your answer".into(),
487            detail: Some("Run cargo test".into()),
488            url: "/sessions/s1".into(),
489            tag: "s1".into(),
490        }
491    }
492
493    /// 10b-ii's review, A3: a token is reused until an hour before it
494    /// expires, then made again with a later expiry.
495    #[test]
496    fn a_token_is_renewed_an_hour_before_it_expires() {
497        let vapid = VapidKey::generate();
498        let endpoint = url::Url::parse("https://fcm.googleapis.com/fcm/send/a").unwrap();
499        let mut tokens = Tokens::default();
500        let made = 1_800_000_000;
501        let first = tokens.authorization(&vapid, &endpoint, None, made);
502        let renew_at = made + VAPID_TOKEN_SECS - 60 * 60;
503        assert_eq!(tokens.authorization(&vapid, &endpoint, None, renew_at - 1), first);
504        let renewed = tokens.authorization(&vapid, &endpoint, None, renew_at);
505        assert_ne!(renewed, first);
506        // Another subject is another token.
507        assert_ne!(
508            tokens.authorization(&vapid, &endpoint, Some("mailto:a@b.co"), renew_at),
509            renewed
510        );
511        // Long past its expiry, an entry is dropped when another is made.
512        let other = url::Url::parse("https://web.push.apple.com/b").unwrap();
513        tokens.authorization(&vapid, &other, None, made + 3 * VAPID_TOKEN_SECS);
514        assert_eq!(tokens.0.len(), 1);
515    }
516
517    #[test]
518    fn retry_after_is_read_in_seconds_only() {
519        let mut headers = header::HeaderMap::new();
520        assert_eq!(retry_after(&headers), None);
521        headers.insert(header::RETRY_AFTER, header::HeaderValue::from_static(" 30 "));
522        assert_eq!(retry_after(&headers), Some(Duration::from_secs(30)));
523        headers.insert(
524            header::RETRY_AFTER,
525            header::HeaderValue::from_static("Wed, 21 Oct 2015 07:28:00 GMT"),
526        );
527        assert_eq!(retry_after(&headers), None);
528        headers.insert(header::RETRY_AFTER, header::HeaderValue::from_static("-1"));
529        assert_eq!(retry_after(&headers), None);
530    }
531
532    #[test]
533    fn the_policy_decides_what_a_payload_shows() {
534        let default = payload_for(&notice(), PushPolicy::default()).unwrap();
535        assert_eq!(
536            (
537                default.title.as_str(),
538                default.body.as_str(),
539                default.url.as_str(),
540                default.tag.as_str()
541            ),
542            ("Fix the flaky test", "needs your answer", "/sessions/s1", "s1")
543        );
544        let generic = PushPolicy {
545            generic_title: true,
546            ..PushPolicy::default()
547        };
548        let shown = payload_for(&notice(), generic).unwrap();
549        assert_eq!(
550            (shown.title.as_str(), shown.body.as_str()),
551            ("Session needs your answer", "")
552        );
553        let details = PushPolicy {
554            details: true,
555            ..PushPolicy::default()
556        };
557        assert_eq!(payload_for(&notice(), details).unwrap().body, "Run cargo test");
558        let both = PushPolicy {
559            details: true,
560            generic_title: true,
561            ..PushPolicy::default()
562        };
563        let shown = payload_for(&notice(), both).unwrap();
564        assert_eq!(
565            (shown.title.as_str(), shown.body.as_str()),
566            ("Session needs your answer", "Run cargo test")
567        );
568        // `details` with no detail: the body as usual.
569        let mut plain = notice();
570        plain.detail = None;
571        assert_eq!(payload_for(&plain, details).unwrap().body, "needs your answer");
572        let muted = PushPolicy {
573            muted: true,
574            details: true,
575            generic_title: false,
576        };
577        assert_eq!(payload_for(&notice(), muted), None);
578    }
579
580    #[test]
581    fn a_payload_is_padded_to_its_bucket_and_stays_json() {
582        let payload = payload_for(&notice(), PushPolicy::default()).unwrap();
583        let short = padded(&payload).unwrap();
584        assert_eq!(short.len(), PAD_TO);
585        assert_eq!(serde_json::from_slice::<PushPayload>(&short).unwrap(), payload);
586        // Titles of different lengths in one bucket look the same.
587        let mut longer = payload.clone();
588        longer.title = "x".repeat(100);
589        assert_eq!(padded(&longer).unwrap().len(), PAD_TO);
590        longer.title = "x".repeat(600);
591        assert_eq!(padded(&longer).unwrap().len(), 2 * PAD_TO);
592        longer.title = "x".repeat(3900);
593        assert_eq!(padded(&longer).unwrap().len(), MAX_PLAINTEXT);
594        longer.title = "x".repeat(4000);
595        assert!(padded(&longer).is_err());
596    }
597}
hennery_kernel/delivery.rs

hennery_kernel/
delivery.rs
