malice

public

Star0

Watch0

127 commits 2 branches 0 tags 0 releases4.4 MiB

use std::{sync::Arc, time::Duration};

use aws_lc_rs::digest::{SHA256, digest};
use base64::{
    Engine,
    engine::general_purpose::{STANDARD_NO_PAD, URL_SAFE_NO_PAD},
};
use reqwest::{StatusCode, header};
use serde::{Deserialize, Serialize};
use url::Url;

use crate::{
    ece::{self, EceError},
    vapid::Vapid,
};

const PUSH_HOSTS: &[&str] = &[
    "*.push.apple.com",
    "fcm.googleapis.com",
    "updates.push.services.mozilla.com",
    "*.notify.windows.com",
];
const ALLOW_HOSTS_ENV: &str = "MALICE_PUSH_ALLOW_HOSTS";
const MAX_ENDPOINT_LEN: usize = 2048;
const TTL_SECS: u32 = 12 * 60 * 60;

/// A browser's `PushSubscription`, keys as the browser's `toJSON` gives them.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Subscription {
    pub endpoint: String,
    pub p256dh: String,
    pub auth: String,
}

impl Subscription {
    /// Whether the keys decode to a P-256 point and a 16-byte secret.
    pub fn keys_valid(&self) -> bool {
        matches!(
            (decode_key(&self.p256dh), decode_key(&self.auth)),
            (Some(p256dh), Some(auth)) if p256dh.len() == 65 && p256dh[0] == 4 && auth.len() == 16
        )
    }
}

fn decode_key(value: &str) -> Option<Vec<u8>> {
    let value = value.trim().trim_end_matches('=');
    URL_SAFE_NO_PAD
        .decode(value)
        .or_else(|_| STANDARD_NO_PAD.decode(value))
        .ok()
}

/// Push service hosts this server will POST to, so a pusher can't aim it at
/// an arbitrary URL.
#[derive(Debug, Clone, Default)]
pub struct Allowlist {
    /// `host:port` authorities also allowed over plain http, for local testing
    extra: Vec<String>,
}

impl Allowlist {
    /// The built-in push services, plus any `host:port` entries in
    /// `MALICE_PUSH_ALLOW_HOSTS`.
    pub fn from_env() -> Self {
        let extra = std::env::var(ALLOW_HOSTS_ENV)
            .unwrap_or_default()
            .split(',')
            .map(|host| host.trim().to_ascii_lowercase())
            .filter(|host| !host.is_empty())
            .collect::<Vec<_>>();
        if !extra.is_empty() {
            tracing::warn!(hosts = ?extra, "{ALLOW_HOSTS_ENV} allows extra push hosts, for testing only");
        }
        Self { extra }
    }

    pub fn with_extra(extra: impl IntoIterator<Item = String>) -> Self {
        Self {
            extra: extra.into_iter().collect(),
        }
    }

    pub fn check(&self, endpoint: &str) -> Option<Url> {
        if endpoint.len() > MAX_ENDPOINT_LEN {
            return None;
        }
        let url = Url::parse(endpoint).ok()?;
        if !url.username().is_empty() || url.password().is_some() {
            return None;
        }
        let port = url.port_or_known_default()?;
        let builtin = url.scheme() == "https"
            && port == 443
            && url.domain().is_some_and(|host| {
                PUSH_HOSTS
                    .iter()
                    .any(|pattern| match pattern.strip_prefix('*') {
                        Some(suffix) => host.len() > suffix.len() && host.ends_with(suffix),
                        None => host == *pattern,
                    })
            });
        let extra = matches!(url.scheme(), "http" | "https")
            && self.extra.contains(&format!("{}:{port}", url.host_str()?));
        (builtin || extra).then_some(url)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Urgency {
    Normal,
    High,
}

pub struct Message<'a> {
    pub payload: &'a [u8],
    pub urgency: Urgency,
    /// pushes sharing a topic replace each other while still queued
    pub topic: &'a str,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
    Delivered,
    /// the push service no longer knows the subscription (404 or 410)
    Gone,
    /// the endpoint is not an allowed push service, or the keys are unusable
    Refused,
    Retryable,
    Failed(String),
}

impl Outcome {
    pub fn drops_subscription(&self) -> bool {
        matches!(self, Self::Gone | Self::Refused)
    }
}

/// Encrypts and delivers pushes, signed with this server's VAPID key.
#[derive(Clone)]
pub struct Sender {
    http: reqwest::Client,
    vapid: Arc<Vapid>,
    contact: String,
    allow: Allowlist,
}

impl Sender {
    /// `contact` is the VAPID `sub`, a `mailto:` or `https:` URL.
    pub fn new(vapid: Vapid, contact: String, allow: Allowlist) -> Result<Self, String> {
        let http = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .connect_timeout(Duration::from_secs(5))
            .timeout(Duration::from_secs(10))
            .build()
            .map_err(|error| format!("building the push client: {error}"))?;
        Ok(Self {
            http,
            vapid: Arc::new(vapid),
            contact,
            allow,
        })
    }

    pub fn vapid_public_key(&self) -> &str {
        self.vapid.public_key()
    }

    pub fn allows(&self, endpoint: &str) -> bool {
        self.allow.check(endpoint).is_some()
    }

    /// Sends once, and once more after a second if the push service asks to retry.
    pub async fn send(&self, subscription: &Subscription, message: &Message<'_>) -> Outcome {
        let Some(url) = self.allow.check(&subscription.endpoint) else {
            return Outcome::Refused;
        };
        let (Some(p256dh), Some(auth)) = (
            decode_key(&subscription.p256dh),
            decode_key(&subscription.auth),
        ) else {
            return Outcome::Refused;
        };
        let body = match ece::encrypt(&p256dh, &auth, message.payload) {
            Ok(body) => body,
            Err(EceError::BadKey) => return Outcome::Refused,
            Err(error) => return Outcome::Failed(error.to_string()),
        };
        let authorization = match self
            .vapid
            .authorization(&url.origin().ascii_serialization(), &self.contact)
        {
            Ok(authorization) => authorization,
            Err(error) => return Outcome::Failed(error),
        };
        let topic = topic(message.topic);
        let urgency = match message.urgency {
            Urgency::Normal => "normal",
            Urgency::High => "high",
        };
        for attempt in 0..2 {
            if attempt > 0 {
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
            let response = self
                .http
                .post(url.clone())
                .header(header::AUTHORIZATION, &authorization)
                .header(header::CONTENT_ENCODING, "aes128gcm")
                .header(header::CONTENT_TYPE, "application/octet-stream")
                .header("TTL", TTL_SECS)
                .header("Urgency", urgency)
                .header("Topic", &topic)
                .body(body.clone())
                .send()
                .await;
            let outcome = match response {
                Ok(response) => classify(response.status()),
                Err(error) => {
                    tracing::debug!(%error, "push service unreachable");
                    Outcome::Retryable
                }
            };
            if outcome != Outcome::Retryable {
                return outcome;
            }
        }
        Outcome::Retryable
    }
}

fn classify(status: StatusCode) -> Outcome {
    match status {
        status if status.is_success() => Outcome::Delivered,
        StatusCode::NOT_FOUND | StatusCode::GONE => Outcome::Gone,
        StatusCode::TOO_MANY_REQUESTS => Outcome::Retryable,
        status if status.is_server_error() => Outcome::Retryable,
        status => Outcome::Failed(format!("push service answered {status}")),
    }
}

/// RFC 8030 limits a topic to 32 characters of the base64url alphabet.
fn topic(raw: &str) -> String {
    let mut topic = URL_SAFE_NO_PAD.encode(digest(&SHA256, raw.as_bytes()));
    topic.truncate(32);
    topic
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn allowlist_admits_push_services_only() {
        let allow = Allowlist::default();
        assert!(
            allow
                .check("https://fcm.googleapis.com/fcm/send/abc")
                .is_some()
        );
        assert!(allow.check("https://web.push.apple.com/QGx").is_some());
        assert!(
            allow
                .check("https://wns2-par02p.notify.windows.com/w/?token=x")
                .is_some()
        );
        assert!(
            allow
                .check("https://updates.push.services.mozilla.com/wpush/v2/x")
                .is_some()
        );
        assert!(allow.check("http://fcm.googleapis.com/x").is_none());
        assert!(allow.check("https://fcm.googleapis.com:8443/x").is_none());
        assert!(allow.check("https://push.apple.com/x").is_none());
        assert!(allow.check("https://evilpush.apple.com/x").is_none());
        assert!(
            allow
                .check("https://fcm.googleapis.com.evil.org/x")
                .is_none()
        );
        assert!(allow.check("https://user@fcm.googleapis.com/x").is_none());
        assert!(allow.check("https://127.0.0.1/x").is_none());
        assert!(allow.check("http://127.0.0.1:9000/x").is_none());
        let allow = Allowlist::with_extra(["localhost:9000".to_owned()]);
        assert!(allow.check("http://localhost:9000/x").is_some());
        assert!(allow.check("http://localhost:9001/x").is_none());
        let allow = Allowlist::with_extra(["127.0.0.1:9000".to_owned()]);
        assert!(allow.check("http://127.0.0.1:9000/x").is_some());
    }

    #[test]
    fn topics_fit_rfc8030() {
        let topic = topic("!room:example.org");
        assert_eq!(topic.len(), 32);
        assert!(
            topic
                .chars()
                .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
        );
    }

    #[test]
    fn keys_validate() {
        let good = Subscription {
            endpoint: String::new(),
            p256dh: "BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4".to_owned(),
            auth: "BTBZMqHH6r4Tts7J_aSIgg==".to_owned(),
        };
        assert!(good.keys_valid());
        assert!(
            !Subscription {
                auth: "short".to_owned(),
                ..good
            }
            .keys_valid()
        );
    }
}

1	use std::{sync::Arc, time::Duration};
2
3	use aws_lc_rs::digest::{SHA256, digest};
4	use base64::{
5	Engine,
6	engine::general_purpose::{STANDARD_NO_PAD, URL_SAFE_NO_PAD},
7	};
8	use reqwest::{StatusCode, header};
9	use serde::{Deserialize, Serialize};
10	use url::Url;
11
12	use crate::{
13	ece::{self, EceError},
14	vapid::Vapid,
15	};
16
17	const PUSH_HOSTS: &[&str] = &[
18	"*.push.apple.com",
19	"fcm.googleapis.com",
20	"updates.push.services.mozilla.com",
21	"*.notify.windows.com",
22	];
23	const ALLOW_HOSTS_ENV: &str = "MALICE_PUSH_ALLOW_HOSTS";
24	const MAX_ENDPOINT_LEN: usize = 2048;
25	const TTL_SECS: u32 = 12 * 60 * 60;
26
27	/// A browser's `PushSubscription`, keys as the browser's `toJSON` gives them.
28	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
29	pub struct Subscription {
30	pub endpoint: String,
31	pub p256dh: String,
32	pub auth: String,
33	}
34
35	impl Subscription {
36	/// Whether the keys decode to a P-256 point and a 16-byte secret.
37	pub fn keys_valid(&self) -> bool {
38	matches!(
39	(decode_key(&self.p256dh), decode_key(&self.auth)),
40	(Some(p256dh), Some(auth)) if p256dh.len() == 65 && p256dh[0] == 4 && auth.len() == 16
41	)
42	}
43	}
44
45	fn decode_key(value: &str) -> Option<Vec<u8>> {
46	let value = value.trim().trim_end_matches('=');
47	URL_SAFE_NO_PAD
48	.decode(value)
49	.or_else(\|_\| STANDARD_NO_PAD.decode(value))
50	.ok()
51	}
52
53	/// Push service hosts this server will POST to, so a pusher can't aim it at
54	/// an arbitrary URL.
55	#[derive(Debug, Clone, Default)]
56	pub struct Allowlist {
57	/// `host:port` authorities also allowed over plain http, for local testing
58	extra: Vec<String>,
59	}
60
61	impl Allowlist {
62	/// The built-in push services, plus any `host:port` entries in
63	/// `MALICE_PUSH_ALLOW_HOSTS`.
64	pub fn from_env() -> Self {
65	let extra = std::env::var(ALLOW_HOSTS_ENV)
66	.unwrap_or_default()
67	.split(',')
68	.map(\|host\| host.trim().to_ascii_lowercase())
69	.filter(\|host\| !host.is_empty())
70	.collect::<Vec<_>>();
71	if !extra.is_empty() {
72	tracing::warn!(hosts = ?extra, "{ALLOW_HOSTS_ENV} allows extra push hosts, for testing only");
73	}
74	Self { extra }
75	}
76
77	pub fn with_extra(extra: impl IntoIterator<Item = String>) -> Self {
78	Self {
79	extra: extra.into_iter().collect(),
80	}
81	}
82
83	pub fn check(&self, endpoint: &str) -> Option<Url> {
84	if endpoint.len() > MAX_ENDPOINT_LEN {
85	return None;
86	}
87	let url = Url::parse(endpoint).ok()?;
88	if !url.username().is_empty() \|\| url.password().is_some() {
89	return None;
90	}
91	let port = url.port_or_known_default()?;
92	let builtin = url.scheme() == "https"
93	&& port == 443
94	&& url.domain().is_some_and(\|host\| {
95	PUSH_HOSTS
96	.iter()
97	.any(\|pattern\| match pattern.strip_prefix('*') {
98	Some(suffix) => host.len() > suffix.len() && host.ends_with(suffix),
99	None => host == *pattern,
100	})
101	});
102	let extra = matches!(url.scheme(), "http" \| "https")
103	&& self.extra.contains(&format!("{}:{port}", url.host_str()?));
104	(builtin \|\| extra).then_some(url)
105	}
106	}
107
108	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
109	pub enum Urgency {
110	Normal,
111	High,
112	}
113
114	pub struct Message<'a> {
115	pub payload: &'a [u8],
116	pub urgency: Urgency,
117	/// pushes sharing a topic replace each other while still queued
118	pub topic: &'a str,
119	}
120
121	#[derive(Debug, Clone, PartialEq, Eq)]
122	pub enum Outcome {
123	Delivered,
124	/// the push service no longer knows the subscription (404 or 410)
125	Gone,
126	/// the endpoint is not an allowed push service, or the keys are unusable
127	Refused,
128	Retryable,
129	Failed(String),
130	}
131
132	impl Outcome {
133	pub fn drops_subscription(&self) -> bool {
134	matches!(self, Self::Gone \| Self::Refused)
135	}
136	}
137
138	/// Encrypts and delivers pushes, signed with this server's VAPID key.
139	#[derive(Clone)]
140	pub struct Sender {
141	http: reqwest::Client,
142	vapid: Arc<Vapid>,
143	contact: String,
144	allow: Allowlist,
145	}
146
147	impl Sender {
148	/// `contact` is the VAPID `sub`, a `mailto:` or `https:` URL.
149	pub fn new(vapid: Vapid, contact: String, allow: Allowlist) -> Result<Self, String> {
150	let http = reqwest::Client::builder()
151	.redirect(reqwest::redirect::Policy::none())
152	.connect_timeout(Duration::from_secs(5))
153	.timeout(Duration::from_secs(10))
154	.build()
155	.map_err(\|error\| format!("building the push client: {error}"))?;
156	Ok(Self {
157	http,
158	vapid: Arc::new(vapid),
159	contact,
160	allow,
161	})
162	}
163
164	pub fn vapid_public_key(&self) -> &str {
165	self.vapid.public_key()
166	}
167
168	pub fn allows(&self, endpoint: &str) -> bool {
169	self.allow.check(endpoint).is_some()
170	}
171
172	/// Sends once, and once more after a second if the push service asks to retry.
173	pub async fn send(&self, subscription: &Subscription, message: &Message<'_>) -> Outcome {
174	let Some(url) = self.allow.check(&subscription.endpoint) else {
175	return Outcome::Refused;
176	};
177	let (Some(p256dh), Some(auth)) = (
178	decode_key(&subscription.p256dh),
179	decode_key(&subscription.auth),
180	) else {
181	return Outcome::Refused;
182	};
183	let body = match ece::encrypt(&p256dh, &auth, message.payload) {
184	Ok(body) => body,
185	Err(EceError::BadKey) => return Outcome::Refused,
186	Err(error) => return Outcome::Failed(error.to_string()),
187	};
188	let authorization = match self
189	.vapid
190	.authorization(&url.origin().ascii_serialization(), &self.contact)
191	{
192	Ok(authorization) => authorization,
193	Err(error) => return Outcome::Failed(error),
194	};
195	let topic = topic(message.topic);
196	let urgency = match message.urgency {
197	Urgency::Normal => "normal",
198	Urgency::High => "high",
199	};
200	for attempt in 0..2 {
201	if attempt > 0 {
202	tokio::time::sleep(Duration::from_secs(1)).await;
203	}
204	let response = self
205	.http
206	.post(url.clone())
207	.header(header::AUTHORIZATION, &authorization)
208	.header(header::CONTENT_ENCODING, "aes128gcm")
209	.header(header::CONTENT_TYPE, "application/octet-stream")
210	.header("TTL", TTL_SECS)
211	.header("Urgency", urgency)
212	.header("Topic", &topic)
213	.body(body.clone())
214	.send()
215	.await;
216	let outcome = match response {
217	Ok(response) => classify(response.status()),
218	Err(error) => {
219	tracing::debug!(%error, "push service unreachable");
220	Outcome::Retryable
221	}
222	};
223	if outcome != Outcome::Retryable {
224	return outcome;
225	}
226	}
227	Outcome::Retryable
228	}
229	}
230
231	fn classify(status: StatusCode) -> Outcome {
232	match status {
233	status if status.is_success() => Outcome::Delivered,
234	StatusCode::NOT_FOUND \| StatusCode::GONE => Outcome::Gone,
235	StatusCode::TOO_MANY_REQUESTS => Outcome::Retryable,
236	status if status.is_server_error() => Outcome::Retryable,
237	status => Outcome::Failed(format!("push service answered {status}")),
238	}
239	}
240
241	/// RFC 8030 limits a topic to 32 characters of the base64url alphabet.
242	fn topic(raw: &str) -> String {
243	let mut topic = URL_SAFE_NO_PAD.encode(digest(&SHA256, raw.as_bytes()));
244	topic.truncate(32);
245	topic
246	}
247
248	#[cfg(test)]
249	mod tests {
250	use super::*;
251
252	#[test]
253	fn allowlist_admits_push_services_only() {
254	let allow = Allowlist::default();
255	assert!(
256	allow
257	.check("https://fcm.googleapis.com/fcm/send/abc")
258	.is_some()
259	);
260	assert!(allow.check("https://web.push.apple.com/QGx").is_some());
261	assert!(
262	allow
263	.check("https://wns2-par02p.notify.windows.com/w/?token=x")
264	.is_some()
265	);
266	assert!(
267	allow
268	.check("https://updates.push.services.mozilla.com/wpush/v2/x")
269	.is_some()
270	);
271	assert!(allow.check("http://fcm.googleapis.com/x").is_none());
272	assert!(allow.check("https://fcm.googleapis.com:8443/x").is_none());
273	assert!(allow.check("https://push.apple.com/x").is_none());
274	assert!(allow.check("https://evilpush.apple.com/x").is_none());
275	assert!(
276	allow
277	.check("https://fcm.googleapis.com.evil.org/x")
278	.is_none()
279	);
280	assert!(allow.check("https://user@fcm.googleapis.com/x").is_none());
281	assert!(allow.check("https://127.0.0.1/x").is_none());
282	assert!(allow.check("http://127.0.0.1:9000/x").is_none());
283	let allow = Allowlist::with_extra(["localhost:9000".to_owned()]);
284	assert!(allow.check("http://localhost:9000/x").is_some());
285	assert!(allow.check("http://localhost:9001/x").is_none());
286	let allow = Allowlist::with_extra(["127.0.0.1:9000".to_owned()]);
287	assert!(allow.check("http://127.0.0.1:9000/x").is_some());
288	}
289
290	#[test]
291	fn topics_fit_rfc8030() {
292	let topic = topic("!room:example.org");
293	assert_eq!(topic.len(), 32);
294	assert!(
295	topic
296	.chars()
297	.all(\|c\| c.is_ascii_alphanumeric() \|\| c == '-' \|\| c == '_')
298	);
299	}
300
301	#[test]
302	fn keys_validate() {
303	let good = Subscription {
304	endpoint: String::new(),
305	p256dh: "BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4".to_owned(),
306	auth: "BTBZMqHH6r4Tts7J_aSIgg==".to_owned(),
307	};
308	assert!(good.keys_valid());
309	assert!(
310	!Subscription {
311	auth: "short".to_owned(),
312	..good
313	}
314	.keys_valid()
315	);
316	}
317	}