#![doc = include_str!("../README.md")]
#![forbid(unsafe_code)]
pub mod ece;
pub mod vapid;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use bytes::Bytes;
use cratefield_core::{
Clock, HttpClient, Notification, Priority, Push, PushError, PushOutcome, Recipient, ttl_secs,
};
use http::header::{AUTHORIZATION, CONTENT_ENCODING, CONTENT_TYPE, LOCATION};
use http::{Request, StatusCode};
use serde_json::{Map, Value};
use sha2::{Digest as _, Sha256};
use crate::ece::{Ece, EceError, SubscriptionKeys};
use crate::vapid::{Vapid, VapidError, VapidKeys};
#[allow(clippy::duration_suboptimal_units)] pub const DEFAULT_TTL: Duration = Duration::from_secs(24 * 3_600);
pub const MAX_TOPIC_LEN: usize = 32;
#[derive(Debug, thiserror::Error)]
pub enum WebPushConfigError {
#[error(transparent)]
Vapid(#[from] VapidError),
#[error(transparent)]
Ece(#[from] EceError),
}
pub struct WebPush {
inner: Inner,
}
enum Inner {
Live(Box<Live>),
NotConfigured,
}
struct Live {
http: Arc<dyn HttpClient>,
clock: Arc<dyn Clock>,
vapid: Vapid,
ece: Ece,
}
impl WebPush {
pub fn new(
http: Arc<dyn HttpClient>,
clock: Arc<dyn Clock>,
keys: VapidKeys,
) -> Result<Self, WebPushConfigError> {
Self::with_record_size(http, clock, keys, ece::DEFAULT_RECORD_SIZE)
}
pub fn with_record_size(
http: Arc<dyn HttpClient>,
clock: Arc<dyn Clock>,
keys: VapidKeys,
record_size: u32,
) -> Result<Self, WebPushConfigError> {
let vapid = Vapid::new(keys)?;
let ece = Ece::new(record_size)?;
Ok(Self {
inner: Inner::Live(Box::new(Live {
http,
clock,
vapid,
ece,
})),
})
}
#[must_use]
pub fn not_configured() -> Self {
Self {
inner: Inner::NotConfigured,
}
}
#[must_use]
pub fn public_key(&self) -> Option<&str> {
match &self.inner {
Inner::Live(live) => Some(live.vapid.public_key()),
Inner::NotConfigured => None,
}
}
#[must_use]
pub fn max_payload(&self) -> Option<usize> {
match &self.inner {
Inner::Live(live) => Some(live.ece.max_plaintext()),
Inner::NotConfigured => None,
}
}
}
impl std::fmt::Debug for WebPush {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match &self.inner {
Inner::Live(live) => f
.debug_struct("WebPush")
.field("vapid", &live.vapid)
.field("record_size", &live.ece.record_size())
.finish_non_exhaustive(),
Inner::NotConfigured => f.write_str("WebPush::NotConfigured"),
}
}
}
fn urgency_header(priority: Priority) -> &'static str {
match priority {
Priority::Immediate => "high",
Priority::Conserve => "normal",
}
}
fn ttl_header(ttl: Option<Duration>) -> String {
ttl_secs(ttl.unwrap_or(DEFAULT_TTL)).to_string()
}
fn topic_header(collapse_id: &str) -> String {
let fits = !collapse_id.is_empty()
&& collapse_id.len() <= MAX_TOPIC_LEN
&& collapse_id
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_');
if fits {
return collapse_id.to_owned();
}
let digest = Sha256::digest(collapse_id.as_bytes());
let mut encoded = URL_SAFE_NO_PAD.encode(digest);
encoded.truncate(MAX_TOPIC_LEN);
encoded
}
fn build_payload(notification: &Notification) -> Vec<u8> {
let mut root = Map::new();
root.insert(
"title".to_owned(),
Value::String(notification.title.clone()),
);
root.insert("body".to_owned(), Value::String(notification.body.clone()));
if let Some(icon) = ¬ification.icon {
root.insert("icon".to_owned(), Value::String(icon.clone()));
}
if let Some(url) = ¬ification.url {
root.insert("url".to_owned(), Value::String(url.clone()));
}
if let Some(thread_id) = ¬ification.thread_id {
root.insert("tag".to_owned(), Value::String(thread_id.clone()));
}
root.insert("silent".to_owned(), Value::Bool(notification.silent));
if !notification.data.is_null() {
root.insert("data".to_owned(), notification.data.clone());
}
Value::Object(root).to_string().into_bytes()
}
const REDACTED: &str = "[redacted]";
const DETAIL_MAX_CHARS: usize = 200;
fn is_uri_char(c: char) -> bool {
c.is_ascii_alphanumeric() || "-._~:/?#[]@!$&'()*+,;=%".contains(c)
}
fn redact_uris(text: &str) -> String {
let mut out = String::with_capacity(text.len());
let mut rest = text;
while !rest.is_empty() {
let start = rest.find(is_uri_char).unwrap_or(rest.len());
out.push_str(&rest[..start]);
rest = &rest[start..];
let end = rest.find(|c: char| !is_uri_char(c)).unwrap_or(rest.len());
let (run, tail) = rest.split_at(end);
if run.contains('/') {
out.push_str(REDACTED);
} else {
out.push_str(run);
}
rest = tail;
}
out
}
fn detail(body: &[u8], status: StatusCode) -> String {
let text: String = String::from_utf8_lossy(body)
.chars()
.filter(|c| !c.is_control())
.take(DETAIL_MAX_CHARS)
.collect();
let text = redact_uris(&text);
let text = text.trim();
if text.is_empty() {
status.as_u16().to_string()
} else {
text.to_owned()
}
}
#[async_trait]
impl Push for WebPush {
async fn send(
&self,
to: &Recipient,
notification: &Notification,
) -> Result<PushOutcome, PushError> {
let Recipient::WebPush {
endpoint,
p256dh,
auth,
} = to
else {
return Err(PushError::unsupported_recipient(to));
};
let live = match &self.inner {
Inner::NotConfigured => return Ok(PushOutcome::NotConfigured),
Inner::Live(live) => live,
};
let origin =
vapid::origin_of(endpoint).map_err(|err| PushError::Rejected(err.to_string()))?;
let keys = SubscriptionKeys::parse(p256dh, auth)
.map_err(|err| PushError::Rejected(err.to_string()))?;
let payload = build_payload(notification);
let body = live.ece.seal(&keys, &payload).map_err(|err| match &err {
EceError::Random(_) => PushError::transient(err.to_string()),
_ => PushError::Rejected(err.to_string()),
})?;
let mut builder = Request::builder()
.method("POST")
.uri(endpoint)
.header(
AUTHORIZATION,
live.vapid.authorization(live.clock.as_ref(), &origin),
)
.header(CONTENT_ENCODING, "aes128gcm")
.header(CONTENT_TYPE, "application/octet-stream")
.header("ttl", ttl_header(notification.ttl))
.header("urgency", urgency_header(notification.priority));
if let Some(collapse_id) = ¬ification.collapse_id {
builder = builder.header("topic", topic_header(collapse_id));
}
let request = builder
.body(Bytes::from(body))
.map_err(|err| PushError::Rejected(format!("could not build request: {err}")))?;
let response = live
.http
.send(request)
.await
.map_err(|err| PushError::transient(err.to_string()))?;
let status = response.status();
let message = || {
format!(
"web push {}: {}",
status.as_u16(),
detail(response.body(), status)
)
};
let retry_after = || cratefield_core::retry_after(response.headers(), live.clock.as_ref());
match status {
status if status.is_success() => Ok(PushOutcome::Delivered {
id: response
.headers()
.get(LOCATION)
.and_then(|value| value.to_str().ok())
.map(str::to_owned),
}),
StatusCode::GONE => Err(PushError::Unregistered),
StatusCode::NOT_FOUND => Err(PushError::transient_after(
format!(
"{} (a 404 is not a dead subscription; only 410 is)",
message()
),
retry_after(),
)),
StatusCode::INSUFFICIENT_STORAGE => Err(PushError::Rejected(format!(
"{} (a push service refusing storage is an operator state, not load: on ntfy \
this is `visitor-subscriber-rate-limiting`, which needs a subscriber on the \
topic or the setting turned off)",
message()
))),
StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => {
live.vapid.invalidate(&origin);
Err(PushError::transient_after(
format!(
"{} (VAPID token refused; re-signing for this origin)",
message()
),
retry_after(),
))
}
status if status.is_redirection() => Err(PushError::transient_after(
format!(
"{message} (redirects are not followed: the VAPID `aud` is signed over the original origin)",
message = message()
),
retry_after(),
)),
StatusCode::TOO_MANY_REQUESTS => {
Err(PushError::transient_after(message(), retry_after()))
}
status if status.is_server_error() => {
Err(PushError::transient_after(message(), retry_after()))
}
_ => Err(PushError::Rejected(message())),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_topic_that_already_fits_is_passed_through() {
assert_eq!(topic_header("room-42"), "room-42");
assert_eq!(topic_header("a"), "a");
let exactly_32 = "a".repeat(MAX_TOPIC_LEN);
assert_eq!(topic_header(&exactly_32), exactly_32);
}
#[test]
fn a_topic_that_does_not_fit_becomes_a_stable_32_character_hash() {
let long = "a".repeat(MAX_TOPIC_LEN + 1);
let hashed = topic_header(&long);
assert_eq!(hashed.len(), MAX_TOPIC_LEN);
assert!(
hashed
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'),
"{hashed}"
);
assert_eq!(hashed, topic_header(&long));
assert_ne!(hashed, topic_header(&"b".repeat(MAX_TOPIC_LEN + 1)));
for id in ["room/42", "user@example.test", "café", ""] {
let topic = topic_header(id);
assert_eq!(topic.len(), MAX_TOPIC_LEN, "{id}");
assert!(
topic
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'),
"{id} -> {topic}"
);
}
}
#[test]
fn the_ttl_header_is_whole_seconds_with_a_day_by_default() {
assert_eq!(ttl_header(None), "86400");
assert_eq!(ttl_header(Some(Duration::from_secs(60))), "60");
assert_eq!(ttl_header(Some(Duration::ZERO)), "0", "deliberate drop");
assert_eq!(
ttl_header(Some(Duration::from_millis(900))),
"1",
"sub-second rounds up, never down to the drop-now instruction"
);
}
#[test]
fn urgency_maps_the_two_priorities_the_port_carries() {
assert_eq!(urgency_header(Priority::Immediate), "high");
assert_eq!(urgency_header(Priority::Conserve), "normal");
}
#[test]
fn an_error_detail_is_bounded_and_stripped_of_control_characters() {
assert_eq!(
detail(b"bad request", StatusCode::BAD_REQUEST),
"bad request"
);
assert_eq!(detail(b"", StatusCode::BAD_REQUEST), "400");
assert_eq!(detail(b" \n ", StatusCode::GONE), "410");
assert_eq!(detail(b"a\nb\tc", StatusCode::BAD_REQUEST), "abc");
let long = vec![b'x'; 4_096];
assert_eq!(detail(&long, StatusCode::BAD_REQUEST).len(), 200);
}
#[test]
fn an_error_detail_never_carries_a_url_or_a_path() {
let apache =
b"The requested URL /wpush/v2/gAAAAABsecret-capability was not found on this server.";
let redacted = detail(apache, StatusCode::NOT_FOUND);
assert_eq!(
redacted,
"The requested URL [redacted] was not found on this server."
);
assert!(!redacted.contains("wpush"), "{redacted}");
assert!(!redacted.contains("gAAAAAB"), "{redacted}");
for body in [
&b"see https://updates.push.services.mozilla.com/wpush/v2/gAAAAABsecret"[..],
&b"Not Found\nhttps://push.example/wpush/v2/gAAAAABsecret"[..],
&b"{\"endpoint\":\"https://push.example/wpush/v2/gAAAAABsecret\"}"[..],
] {
let redacted = detail(body, StatusCode::NOT_FOUND);
assert!(!redacted.contains("gAAAAABsecret"), "{redacted}");
assert!(!redacted.contains("push.example"), "{redacted}");
assert!(!redacted.contains("mozilla.com"), "{redacted}");
assert!(redacted.contains(REDACTED), "{redacted}");
}
assert_eq!(
detail(
b"cannot publish to UnifiedPush topic without previously active subscriber",
StatusCode::INSUFFICIENT_STORAGE
),
"cannot publish to UnifiedPush topic without previously active subscriber"
);
assert_eq!(
detail(b"UnauthorizedRegistration", StatusCode::UNAUTHORIZED),
"UnauthorizedRegistration"
);
let long_endpoint = format!("https://push.example/wpush/v2/{}", "A".repeat(4_000));
let redacted = detail(long_endpoint.as_bytes(), StatusCode::NOT_FOUND);
assert_eq!(redacted, REDACTED);
}
#[test]
fn redaction_leaves_text_that_is_not_a_path_alone() {
assert_eq!(
redact_uris("plain prose, with punctuation!"),
"plain prose, with punctuation!"
);
assert_eq!(redact_uris(""), "");
assert_eq!(redact_uris("410"), "410");
assert_eq!(redact_uris("café /a/b naïve"), "café [redacted] naïve");
}
}