diff --git a/Cargo.lock b/Cargo.lock index 1c5950047..d4060ef16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,41 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common 0.1.6", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures 0.2.17", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + [[package]] name = "aho-corasick" version = "1.1.5" @@ -158,7 +193,7 @@ dependencies = [ "rustc-hash", "serde", "serde_derive", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -220,7 +255,7 @@ checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -695,6 +730,16 @@ dependencies = [ "windows-link", ] +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common 0.1.6", + "inout", +] + [[package]] name = "clang-sys" version = "1.9.1" @@ -708,9 +753,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.6.7" +version = "4.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa8876b300ab35ba921adea3dfd70157a46249b33f95c9084ae5709785478946" +checksum = "473c7e07f409a8d772161724aa8db6a765a2532a70f9667eeb7b49d3d02fbdca" dependencies = [ "clap_builder", "clap_derive", @@ -718,9 +763,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.6.7" +version = "4.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec0797fb7aeb1406c84efac526901f7ec3ead2124f946b494e72879d4b54704d" +checksum = "7b48fea5a88e9ae728a2dcbedbfc0e730f7d60da42e1cb049a83c9fb8b789889" dependencies = [ "anstyle", "clap_lex", @@ -728,14 +773,14 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.6.7" +version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9c751b79415d4e559e3d1fcf128e09e720eb673a06d26cf6f392d37d75b66e0" +checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" dependencies = [ "heck", "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -1010,7 +1055,7 @@ dependencies = [ "itertools 0.15.0", "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -1069,6 +1114,7 @@ dependencies = [ "governor", "hickory-resolver", "http", + "httpdate", "image", "ipaddress", "itertools 0.15.0", @@ -1078,6 +1124,7 @@ dependencies = [ "lru-cache", "nonzero_ext", "openidconnect", + "p256", "rand 0.10.2", "recaptcha-verify", "regex", @@ -1097,6 +1144,7 @@ dependencies = [ "tokio", "tracing", "url", + "web-push-native", "webpage", "yansi", ] @@ -1434,6 +1482,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" dependencies = [ "generic-array", + "rand_core 0.6.4", "typenum", ] @@ -1456,6 +1505,15 @@ dependencies = [ "linktime-proc-macro", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + [[package]] name = "ctutils" version = "0.4.2" @@ -1557,7 +1615,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -1579,7 +1637,7 @@ checksum = "2ac7135c3ef02b2f7833bbeb1be5ba7f966dcde8a87c6b87f65a778d71a02785" dependencies = [ "darling_core 0.24.1", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -1729,7 +1787,7 @@ checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -1776,6 +1834,17 @@ dependencies = [ "spki 0.7.3", ] +[[package]] +name = "ece-native" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6911cc97c62febd8cbcb2cb49ed21c58ccd418a80c43f20df4973ee08484e65d" +dependencies = [ + "aes-gcm", + "hkdf 0.12.4", + "sha2 0.10.9", +] + [[package]] name = "ed25519" version = "2.2.3" @@ -1846,7 +1915,7 @@ dependencies = [ "ff", "generic-array", "group", - "hkdf", + "hkdf 0.12.4", "pem-rfc7468", "pkcs8 0.10.2", "rand_core 0.6.4", @@ -2119,7 +2188,7 @@ checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -2207,6 +2276,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + [[package]] name = "gif" version = "0.14.2" @@ -2250,9 +2329,9 @@ dependencies = [ [[package]] name = "granit-parser" -version = "1.3.0" +version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e20f99e46474f56bd905c56e817ebddcf377a611f94c53ac4649e4d3fa3c0cd0" +checksum = "48aa83cc6ac4dc610adf5501a5392b4bcc543b2c1f24eaf77469a96e31ec9abe" dependencies = [ "arraydeque", "smallvec", @@ -2481,6 +2560,15 @@ dependencies = [ "hmac 0.12.1", ] +[[package]] +name = "hkdf" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018" +dependencies = [ + "hmac 0.13.0", +] + [[package]] name = "hmac" version = "0.12.1" @@ -2858,6 +2946,15 @@ version = "0.1.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "ipaddress" version = "0.2.11" @@ -3479,7 +3576,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -3855,6 +3952,12 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "openidconnect" version = "4.0.1" @@ -4238,6 +4341,18 @@ dependencies = [ "miniz_oxide 0.8.9", ] +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "opaque-debug", + "universal-hash", +] + [[package]] name = "portable-atomic" version = "1.15.0" @@ -4331,7 +4446,7 @@ dependencies = [ "proc-macro-error-attr3", "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -4625,7 +4740,7 @@ checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -4961,7 +5076,7 @@ dependencies = [ "quote", "ruma-identifiers-validation", "serde", - "syn 3.0.6", + "syn 3.0.5", "toml 1.1.6+spec-1.1.0", ] @@ -5014,6 +5129,7 @@ name = "ruminuwuity" version = "26.9.0" dependencies = [ "assign", + "http", "ruma", "serde", "serde_json", @@ -5442,12 +5558,12 @@ dependencies = [ [[package]] name = "serde-saphyr" -version = "1.3.0" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8050abb251097357e24aff63ba2c52a6309ecb7d23a5474023df960a02694d8" +checksum = "3afb591f9cdb6223c88ba39269aff895620c7f0716dc42b705b5733d5c7c0823" dependencies = [ "annotate-snippets", - "base64 0.21.7", + "base64 0.23.1", "encoding_rs_io", "granit-parser", "nohash-hasher", @@ -5484,7 +5600,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -5603,7 +5719,7 @@ dependencies = [ "darling 0.24.1", "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -5777,9 +5893,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.16.1" +version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" dependencies = [ "serde", ] @@ -5891,9 +6007,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.6" +version = "3.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +checksum = "12df2e0110f65b775f769bb17ef989067a1d931b2eb822bd4346631eeada89f9" dependencies = [ "proc-macro2", "quote", @@ -6021,7 +6137,7 @@ checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -6132,7 +6248,7 @@ checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] @@ -6635,6 +6751,16 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common 0.1.6", + "subtle", +] + [[package]] name = "untrusted" version = "0.9.0" @@ -6806,7 +6932,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", "wasm-bindgen-shared", ] @@ -6832,6 +6958,20 @@ dependencies = [ "web-sys", ] +[[package]] +name = "web-push-native" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7628c93e39aeee4c0bca50ae333052b4f174c653e8fca1d0a8e34f9bc4879c0c" +dependencies = [ + "aes-gcm", + "ece-native", + "hkdf 0.13.0", + "http", + "p256", + "sha2 0.11.0", +] + [[package]] name = "web-sys" version = "0.3.105" @@ -7240,7 +7380,7 @@ checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.5", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 860d40dbb..d92ec888b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -66,6 +66,15 @@ default-features = false version = "0.13.0" default-features = false +[workspace.dependencies.web-push-native] +version = "0.5.0" +default-features = false + +[workspace.dependencies.p256] +version = "0.13.2" +default-features = false +features = ["ecdsa", "std"] + # used for checking if an IP is in specific subnets / CIDR ranges easier [workspace.dependencies.ipaddress] version = "0.2.0" @@ -83,6 +92,10 @@ version = "0.1.3" [workspace.dependencies.http] version = "1.3.1" +# parses the HTTP-date form of Retry-After +[workspace.dependencies.httpdate] +version = "1.0.3" + [workspace.dependencies.regex] version = "1.11.1" diff --git a/changelog.d/+msc4174-webpush.feature.md b/changelog.d/+msc4174-webpush.feature.md new file mode 100644 index 000000000..ed2bd08b2 --- /dev/null +++ b/changelog.d/+msc4174-webpush.feature.md @@ -0,0 +1 @@ +Added encrypted Web Push (MSC4174), enabled by default. Contributed by @eleboucher. diff --git a/src/api/client/capabilities.rs b/src/api/client/capabilities.rs index 871786ab8..af5eb6ea2 100644 --- a/src/api/client/capabilities.rs +++ b/src/api/client/capabilities.rs @@ -13,6 +13,7 @@ }, assign, }; +use serde_json::json; use crate::Ruma; @@ -55,5 +56,10 @@ pub(crate) async fn get_capabilities_route( assign!(ProfileFieldsCapability::new(true), { disallowed: Some(services.oidc.restricted_profile_fields()) }), ); + capabilities.set( + "org.matrix.msc4174.webpush", + json!({ "enabled": true, "vapid": services.pusher.webpush_vapid_public_key() }), + )?; + Ok(get_capabilities::v3::Response::new(capabilities)) } diff --git a/src/api/client/push.rs b/src/api/client/push.rs index 67d9e15e0..d04db0571 100644 --- a/src/api/client/push.rs +++ b/src/api/client/push.rs @@ -1,13 +1,16 @@ +use std::sync::LazyLock; + use axum::extract::State; -use conduwuit::{Err, Error, Result, err}; -use conduwuit_service::Services; +use conduwuit::{Err, Error, Result, err, utils::IterStream}; +use conduwuit_service::{Services, pusher::ActivationOutcome}; +use futures::StreamExt; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, api::{ client::push::{ - delete_pushrule, get_pushers, get_pushrule, get_pushrule_actions, - get_pushrule_enabled, get_pushrules_all, get_pushrules_global_scope, set_pusher, - set_pushrule, set_pushrule_actions, set_pushrule_enabled, + delete_pushrule, get_pushrule, get_pushrule_actions, get_pushrule_enabled, + get_pushrules_all, get_pushrules_global_scope, set_pushrule, set_pushrule_actions, + set_pushrule_enabled, }, error::ErrorKind, }, @@ -20,6 +23,12 @@ RemovePushRuleError, Ruleset, }, }; +use ruminuwuity::pushers::{ + PusherKind, + ack::unstable as pushers_ack, + get_pushers::{self, v3::PusherWithActivation}, + set_pusher, +}; use crate::Ruma; @@ -460,7 +469,29 @@ pub(crate) async fn get_pushers_route( ) -> Result { let sender_user = body.identity.expect_sender_user()?; - Ok(get_pushers::v3::Response::new(services.pusher.get_pushers(sender_user).await)) + let pushers = services + .pusher + .get_pushers(sender_user) + .await + .into_iter() + .stream() + .then(async |pusher| { + let activated = match pusher.kind { + | PusherKind::WebPush(_) => Some( + services + .pusher + .is_webpush_activated(sender_user, pusher.ids.pushkey.as_str()) + .await, + ), + | _ => None, + }; + + PusherWithActivation { pusher, activated } + }) + .collect() + .await; + + Ok(get_pushers::v3::Response::new(pushers)) } /// # `POST /_matrix/client/r0/pushers/set` @@ -475,12 +506,50 @@ pub(crate) async fn set_pushers_route( let sender_user = body.identity.expect_sender_user()?; let sender_device = body.identity.expect_sender_device()?; - services + let needs_activation = services .pusher .set_pusher(sender_user, sender_device, &body.action) .await?; - Ok(set_pusher::v3::Response::new()) + Ok(set_pusher::v3::Response::new(needs_activation)) +} + +pub(crate) async fn pushers_ack_route( + State(services): State, + body: Ruma, +) -> Result { + let sender_user = body.identity.expect_sender_user()?; + + match services + .pusher + .activate_webpush_pusher(sender_user, &body.app_id, &body.ack_token) + .await? + { + | ActivationOutcome::Activated => Ok(pushers_ack::Response::new()), + | ActivationOutcome::NoSuchPusher => { + Err!(Request(NotFound("No pusher with this app ID exists."))) + }, + | ActivationOutcome::UnknownToken => Err(Error::Request( + UNKNOWN_ACTIVATION_TOKEN.clone(), + "Unknown activation token.".into(), + http::StatusCode::BAD_REQUEST, + )), + | ActivationOutcome::Expired => Err(Error::Request( + EXPIRED_ACTIVATION_TOKEN.clone(), + "Activation token has expired.".into(), + http::StatusCode::BAD_REQUEST, + )), + } +} + +static EXPIRED_ACTIVATION_TOKEN: LazyLock = + LazyLock::new(|| custom_error_kind("ORG.MATRIX.MSC4174_EXPIRED_ACTIVATION_TOKEN")); +static UNKNOWN_ACTIVATION_TOKEN: LazyLock = + LazyLock::new(|| custom_error_kind("ORG.MATRIX.MSC4174_UNKNOWN_ACTIVATION_TOKEN")); + +fn custom_error_kind(errcode: &str) -> ErrorKind { + serde_json::from_value(serde_json::json!({ "errcode": errcode })) + .expect("errcode deserializes into an ErrorKind") } /// user somehow has bad push rules, these must always exist per spec. diff --git a/src/api/router.rs b/src/api/router.rs index 6b972043d..9ef053a6d 100644 --- a/src/api/router.rs +++ b/src/api/router.rs @@ -171,6 +171,7 @@ pub fn build(router: Router, state: State) -> Router { .ruma_route(&client::get_key_changes_route) .ruma_route(&client::get_pushers_route) .ruma_route(&client::set_pushers_route) + .ruma_route(&client::pushers_ack_route) .ruma_route(&client::upgrade_room_route) .ruma_route(&client::get_threads_route) .ruma_route(&client::get_relating_events_with_rel_type_and_event_type_route) diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 13358d59b..84193ba1c 100644 --- a/src/core/config/mod.rs +++ b/src/core/config/mod.rs @@ -2853,6 +2853,26 @@ pub enum OidcProfileKeyImportMode { ]; impl Config { + /// Get the base domain to use for user-facing URLs. + #[must_use] + pub fn get_client_domain(&self) -> Url { + self.well_known.client.clone().unwrap_or_else(|| { + let host = self.server_name.host(); + format!("https://{host}") + .as_str() + .try_into() + .expect("server name should be a valid host") + }) + } + + /// Contact for the VAPID `sub` claim. + #[must_use] + pub fn webpush_vapid_contact(&self) -> Url { + self.get_client_domain() + .join("/_continuwuity/about") + .expect("about page is a valid path") + } + /// Pre-initialize config pub fn load(paths: &[PathBuf]) -> Result { let envs = [ diff --git a/src/core/matrix/versions.rs b/src/core/matrix/versions.rs index 79fbf82dc..42b147a40 100644 --- a/src/core/matrix/versions.rs +++ b/src/core/matrix/versions.rs @@ -44,5 +44,7 @@ pub fn unstable_features() -> BTreeMap { ("org.matrix.msc4155".to_owned(), true), // profile change propagation (https://github.com/matrix-org/matrix-spec-proposals/pull/4466) ("computer.gingershaped.msc4466".to_owned(), true), + // web push (https://github.com/matrix-org/matrix-spec-proposals/pull/4174) + ("org.matrix.msc4174".to_owned(), true), ]) } diff --git a/src/database/maps.rs b/src/database/maps.rs index 457e62764..063f88320 100644 --- a/src/database/maps.rs +++ b/src/database/maps.rs @@ -266,6 +266,14 @@ pub(super) fn open_list(db: &Arc, maps: &[Descriptor]) -> Result { name: "senderkey_pusher", ..descriptor::RANDOM_SMALL }, + Descriptor { + name: "senderkey_webpushbackoff", + ..descriptor::RANDOM_SMALL + }, + Descriptor { + name: "senderkey_webpushstate", + ..descriptor::RANDOM_SMALL + }, Descriptor { name: "server_signingkeys", ..descriptor::RANDOM diff --git a/src/ruminuwuity/Cargo.toml b/src/ruminuwuity/Cargo.toml index 2cd86e93c..0bdcb2f90 100644 --- a/src/ruminuwuity/Cargo.toml +++ b/src/ruminuwuity/Cargo.toml @@ -21,6 +21,7 @@ unstable-msc4203 = [] [dependencies] assign.workspace = true +http.workspace = true ruma.workspace = true serde = { workspace = true } serde_json = { workspace = true } diff --git a/src/ruminuwuity/mod.rs b/src/ruminuwuity/mod.rs index 9c98031db..39623e1ca 100644 --- a/src/ruminuwuity/mod.rs +++ b/src/ruminuwuity/mod.rs @@ -5,3 +5,4 @@ pub mod invite_permission_config; pub mod meowlnir_antispam; pub mod policy; +pub mod pushers; diff --git a/src/ruminuwuity/pushers/ack.rs b/src/ruminuwuity/pushers/ack.rs new file mode 100644 index 000000000..fdf93e59f --- /dev/null +++ b/src/ruminuwuity/pushers/ack.rs @@ -0,0 +1,36 @@ +pub mod unstable { + use ruma::{ + api::{auth_scheme::AccessToken, error::Error as MatrixError, request, response}, + metadata, + }; + + metadata! { + method: POST, + rate_limited: false, + authentication: AccessToken, + history: { + unstable => "/_matrix/client/unstable/org.matrix.msc4174/pushers/ack", + } + } + + #[request(error = MatrixError)] + pub struct Request { + pub app_id: String, + pub ack_token: String, + } + + #[response(error = MatrixError)] + #[derive(Default)] + #[allow(clippy::empty_structs_with_brackets, reason = "required by #[response]")] + pub struct Response {} + + impl Request { + #[must_use] + pub fn new(app_id: String, ack_token: String) -> Self { Self { app_id, ack_token } } + } + + impl Response { + #[must_use] + pub fn new() -> Self { Self {} } + } +} diff --git a/src/ruminuwuity/pushers/get_pushers.rs b/src/ruminuwuity/pushers/get_pushers.rs new file mode 100644 index 000000000..f1462899f --- /dev/null +++ b/src/ruminuwuity/pushers/get_pushers.rs @@ -0,0 +1,69 @@ +pub mod v3 { + use ruma::{ + api::{auth_scheme::AccessToken, error::Error as MatrixError, request, response}, + metadata, + serde::from_raw_json_value, + }; + use serde::{Deserialize, Serialize, de}; + use serde_json::value::RawValue as RawJsonValue; + + use crate::pushers::Pusher; + + metadata! { + method: GET, + rate_limited: false, + authentication: AccessToken, + history: { + 1.0 => "/_matrix/client/r0/pushers", + 1.1 => "/_matrix/client/v3/pushers", + } + } + + #[request(error = MatrixError)] + #[derive(Default)] + #[allow(clippy::empty_structs_with_brackets, reason = "required by #[request]")] + pub struct Request {} + + #[response(error = MatrixError)] + pub struct Response { + pub pushers: Vec, + } + + #[derive(Clone, Debug, Serialize)] + pub struct PusherWithActivation { + #[serde(flatten)] + pub pusher: Pusher, + + #[serde(skip_serializing_if = "Option::is_none")] + pub activated: Option, + } + + #[derive(Deserialize)] + struct ActivationDeHelper { + activated: Option, + } + + impl<'de> Deserialize<'de> for PusherWithActivation { + fn deserialize(deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + let json = Box::::deserialize(deserializer)?; + + let ActivationDeHelper { activated } = from_raw_json_value(&json)?; + let pusher = from_raw_json_value(&json)?; + + Ok(Self { pusher, activated }) + } + } + + impl Request { + #[must_use] + pub fn new() -> Self { Self {} } + } + + impl Response { + #[must_use] + pub fn new(pushers: Vec) -> Self { Self { pushers } } + } +} diff --git a/src/ruminuwuity/pushers/mod.rs b/src/ruminuwuity/pushers/mod.rs new file mode 100644 index 000000000..dad81c6bb --- /dev/null +++ b/src/ruminuwuity/pushers/mod.rs @@ -0,0 +1,87 @@ +pub mod ack; +pub mod get_pushers; +mod pusher_serde; +pub mod set_pusher; + +use std::fmt; + +pub use ruma::api::client::push::{EmailPusherData, PusherIds}; +use ruma::{ + push::{HttpPusherData, PushFormat}, + serde::JsonObject, +}; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Serialize)] +pub struct Pusher { + #[serde(flatten)] + pub ids: PusherIds, + + #[serde(flatten)] + pub kind: PusherKind, + + pub app_display_name: String, + + pub device_display_name: String, + + #[serde(skip_serializing_if = "Option::is_none")] + pub profile_tag: Option, + + pub lang: String, +} + +#[derive(Clone, Debug)] +pub enum PusherKind { + Http(HttpPusherData), + Email(EmailPusherData), + WebPush(WebPushPusherData), + + Custom { + kind: String, + data: JsonObject, + }, +} + +#[derive(Clone, Serialize, Deserialize)] +pub struct WebPushPusherData { + pub url: String, + + pub auth: String, + + #[serde(skip_serializing_if = "Option::is_none")] + pub format: Option, + + #[serde(flatten, default, skip_serializing_if = "JsonObject::is_empty")] + pub data: JsonObject, +} + +impl fmt::Debug for WebPushPusherData { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("WebPushPusherData") + .field("url", &self.url) + .field("auth", &"") + .field("format", &self.format) + .field("data", &self.data) + .finish() + } +} + +impl PusherKind { + #[must_use] + pub fn as_str(&self) -> &str { + match self { + | Self::Http(_) => "http", + | Self::Email(_) => "email", + | Self::WebPush(_) => "org.matrix.msc4174.webpush", + | Self::Custom { kind, .. } => kind, + } + } + + #[must_use] + pub fn as_webpush(&self) -> Option<&WebPushPusherData> { + match self { + | Self::WebPush(data) => Some(data), + | _ => None, + } + } +} diff --git a/src/ruminuwuity/pushers/pusher_serde.rs b/src/ruminuwuity/pushers/pusher_serde.rs new file mode 100644 index 000000000..fa76726e3 --- /dev/null +++ b/src/ruminuwuity/pushers/pusher_serde.rs @@ -0,0 +1,196 @@ +use ruma::serde::from_raw_json_value; +use serde::{Deserialize, Serialize, de, ser::SerializeStruct}; +use serde_json::value::RawValue as RawJsonValue; + +use super::{Pusher, PusherIds, PusherKind}; + +#[derive(Debug, Deserialize)] +struct PusherDeHelper { + #[serde(flatten)] + ids: PusherIds, + app_display_name: String, + device_display_name: String, + profile_tag: Option, + lang: String, +} + +impl<'de> Deserialize<'de> for Pusher { + fn deserialize(deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + let json = Box::::deserialize(deserializer)?; + + let PusherDeHelper { + ids, + app_display_name, + device_display_name, + profile_tag, + lang, + } = from_raw_json_value(&json)?; + let kind = from_raw_json_value(&json)?; + + Ok(Self { + ids, + kind, + app_display_name, + device_display_name, + profile_tag, + lang, + }) + } +} + +impl Serialize for PusherKind { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + let mut st = serializer.serialize_struct("PusherKind", 2)?; + st.serialize_field("kind", self.as_str())?; + + match self { + | Self::Http(data) => st.serialize_field("data", data)?, + | Self::Email(data) => st.serialize_field("data", data)?, + | Self::WebPush(data) => st.serialize_field("data", data)?, + | Self::Custom { data, .. } => st.serialize_field("data", data)?, + } + + st.end() + } +} + +#[derive(Debug, Deserialize)] +struct PusherKindDeHelper { + kind: String, + data: Box, +} + +impl<'de> Deserialize<'de> for PusherKind { + fn deserialize(deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + let json = Box::::deserialize(deserializer)?; + let PusherKindDeHelper { kind, data } = from_raw_json_value(&json)?; + + match kind.as_str() { + | "http" => from_raw_json_value(&data).map(Self::Http), + | "email" => from_raw_json_value(&data).map(Self::Email), + | "org.matrix.msc4174.webpush" => from_raw_json_value(&data).map(Self::WebPush), + | _ => Ok(Self::Custom { kind, data: from_raw_json_value(&data)? }), + } + } +} + +#[cfg(test)] +mod tests { + use serde_json::{from_value as from_json_value, json, to_value as to_json_value}; + + use super::{Pusher, PusherKind}; + + #[test] + fn roundtrip_ruma_http_pusher() { + let stored = json!({ + "pushkey": "abcdef", + "app_id": "my.matrix.app", + "kind": "http", + "data": { + "url": "https://push.example.org/_matrix/push/v1/notify", + "format": "event_id_only", + "default_payload": { "session_id": "abc" }, + }, + "app_display_name": "My Matrix App", + "device_display_name": "My Phone", + "profile_tag": "tag", + "lang": "en", + }); + + let pusher: Pusher = from_json_value(stored.clone()).unwrap(); + assert!(matches!(pusher.kind, PusherKind::Http(_))); + assert_eq!(to_json_value(&pusher).unwrap(), stored); + } + + #[test] + fn deserialize_legacy_append_field() { + let stored = json!({ + "pushkey": "abcdef", + "app_id": "my.matrix.app", + "kind": "email", + "data": {}, + "append": true, + "app_display_name": "My Matrix App", + "device_display_name": "My Phone", + "lang": "en", + }); + + let pusher: Pusher = from_json_value(stored).unwrap(); + assert!(matches!(pusher.kind, PusherKind::Email(_))); + assert_eq!(pusher.profile_tag, None); + } + + #[test] + fn roundtrip_webpush_pusher() { + let stored = json!({ + "pushkey": "BLn9b-VR0ca83knDNZ32dCHGyjJp", + "app_id": "m.webpush", + "kind": "org.matrix.msc4174.webpush", + "data": { + "url": "https://updates.push.services.mozilla.com/wpush/v2/abc", + "auth": "_ordMnz7uTCmrpBTeUV4Bw", + "format": "event_id_only", + "default_payload": { "session_id": "abc" }, + }, + "app_display_name": "WebPush", + "device_display_name": "Firefox", + "lang": "en", + }); + + let pusher: Pusher = from_json_value(stored.clone()).unwrap(); + let data = pusher.kind.as_webpush().unwrap(); + assert_eq!(data.auth, "_ordMnz7uTCmrpBTeUV4Bw"); + assert_eq!(data.data.len(), 1); + assert_eq!(to_json_value(&pusher).unwrap(), stored); + } + + #[test] + fn roundtrip_pusher_with_activation() { + use crate::pushers::get_pushers::v3::PusherWithActivation; + + let wire = json!({ + "pushkey": "BLn9b-VR0ca83knDNZ32dCHGyjJp", + "app_id": "m.webpush", + "kind": "org.matrix.msc4174.webpush", + "data": { + "url": "https://updates.push.services.mozilla.com/wpush/v2/abc", + "auth": "_ordMnz7uTCmrpBTeUV4Bw", + }, + "app_display_name": "WebPush", + "device_display_name": "Firefox", + "lang": "en", + "activated": true, + }); + + let pusher: PusherWithActivation = from_json_value(wire.clone()).unwrap(); + assert_eq!(pusher.activated, Some(true)); + assert_eq!(pusher.pusher.kind.as_str(), "org.matrix.msc4174.webpush"); + assert_eq!(to_json_value(&pusher).unwrap(), wire); + } + + #[test] + fn roundtrip_unknown_kind() { + let stored = json!({ + "pushkey": "abcdef", + "app_id": "my.matrix.app", + "kind": "my.custom.kind", + "data": { "whatever": 1 }, + "app_display_name": "App", + "device_display_name": "Device", + "lang": "en", + }); + + let pusher: Pusher = from_json_value(stored.clone()).unwrap(); + assert_eq!(pusher.kind.as_str(), "my.custom.kind"); + assert_eq!(to_json_value(&pusher).unwrap(), stored); + } +} diff --git a/src/ruminuwuity/pushers/set_pusher.rs b/src/ruminuwuity/pushers/set_pusher.rs new file mode 100644 index 000000000..9b1b4cfec --- /dev/null +++ b/src/ruminuwuity/pushers/set_pusher.rs @@ -0,0 +1,129 @@ +pub mod v3 { + use ruma::{ + JsOption, + api::{ + EmptyBody, IncomingResponse, OutgoingResponse, + auth_scheme::AccessToken, + error::{DeserializationError, Error as MatrixError, IntoHttpError}, + request, + }, + metadata, + serde::from_raw_json_value, + }; + use serde::{Deserialize, Serialize, de}; + use serde_json::value::RawValue as RawJsonValue; + + use crate::pushers::{Pusher, PusherIds}; + + metadata! { + method: POST, + rate_limited: true, + authentication: AccessToken, + history: { + 1.0 => "/_matrix/client/r0/pushers/set", + 1.1 => "/_matrix/client/v3/pushers/set", + } + } + + #[request(error = MatrixError)] + pub struct Request { + #[ruma_api(body)] + pub action: PusherAction, + } + + #[derive(Clone, Debug, Default)] + pub struct Response { + pub needs_activation: bool, + } + + impl Request { + #[must_use] + pub fn new(action: PusherAction) -> Self { Self { action } } + } + + impl Response { + #[must_use] + pub fn new(needs_activation: bool) -> Self { Self { needs_activation } } + } + + #[cfg(feature = "server")] + impl OutgoingResponse for Response { + type Body = EmptyBody; + + fn try_into_http_response_inner( + self, + ) -> Result, IntoHttpError> { + let status = if self.needs_activation { + http::StatusCode::CREATED + } else { + http::StatusCode::OK + }; + + http::Response::builder() + .status(status) + .header(http::header::CONTENT_TYPE, ruma::http_headers::APPLICATION_JSON) + .body(EmptyBody) + .map_err(Into::into) + } + } + + #[cfg(feature = "client")] + impl IncomingResponse for Response { + type EndpointError = MatrixError; + + fn try_from_http_response_inner( + response: http::Response<&[u8]>, + ) -> Result { + Ok(Self { + needs_activation: response.status() == http::StatusCode::CREATED, + }) + } + } + + #[derive(Clone, Debug)] + pub enum PusherAction { + Post(Box), + Delete(PusherIds), + } + + #[derive(Debug, Deserialize)] + struct PusherActionDeHelper { + kind: JsOption, + } + + impl<'de> Deserialize<'de> for PusherAction { + fn deserialize(deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + let json = Box::::deserialize(deserializer)?; + let PusherActionDeHelper { kind } = from_raw_json_value(&json)?; + + match kind { + | JsOption::Some(_) => Ok(Self::Post(from_raw_json_value(&json)?)), + | JsOption::Null => Ok(Self::Delete(from_raw_json_value(&json)?)), + | JsOption::Undefined => Err(de::Error::missing_field("kind")), + } + } + } + + impl Serialize for PusherAction { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + match self { + | Self::Post(pusher) => pusher.serialize(serializer), + | Self::Delete(ids) => { + use serde::ser::SerializeStruct; + + let mut st = serializer.serialize_struct("PusherAction", 3)?; + st.serialize_field("pushkey", &ids.pushkey)?; + st.serialize_field("app_id", &ids.app_id)?; + st.serialize_field("kind", &None::<&str>)?; + st.end() + }, + } + } + } +} diff --git a/src/service/Cargo.toml b/src/service/Cargo.toml index fff9f6df5..d647fa976 100644 --- a/src/service/Cargo.toml +++ b/src/service/Cargo.toml @@ -88,6 +88,7 @@ futures.workspace = true governor.workspace = true hickory-resolver.workspace = true http.workspace = true +httpdate.workspace = true image.workspace = true image.optional = true ipaddress.workspace = true @@ -96,6 +97,7 @@ log.workspace = true loole.workspace = true lru-cache.workspace = true nonzero_ext.workspace = true +p256.workspace = true regex.workspace = true reqwest.workspace = true assign.workspace = true @@ -112,6 +114,7 @@ termimad.optional = true tokio.workspace = true tracing.workspace = true url.workspace = true +web-push-native.workspace = true webpage.workspace = true webpage.optional = true recaptcha-verify = { version = "0.2.0", default-features = false } diff --git a/src/service/client/mod.rs b/src/service/client/mod.rs index 0fe55b0d5..26ee1ee51 100644 --- a/src/service/client/mod.rs +++ b/src/service/client/mod.rs @@ -24,6 +24,7 @@ pub struct Service { pub sender: reqwest::Client, pub appservice: reqwest::Client, pub pusher: reqwest::Client, + pub webpush: reqwest::Client, pub cidr_range_denylist: Vec, } @@ -141,6 +142,16 @@ fn build(args: crate::Args<'_>) -> Result> { .redirect(redirect::Policy::limited(2)) .build()?, + // MSC4174 recommends against following redirects, since the destination + // is only checked before the request is sent. + webpush: base(config)? + .connect_timeout(Duration::from_secs(config.pusher_conn_timeout)) + .timeout(Duration::from_secs(config.pusher_timeout)) + .pool_max_idle_per_host(1) + .pool_idle_timeout(Duration::from_secs(config.pusher_idle_timeout)) + .redirect(redirect::Policy::none()) + .build()?, + cidr_range_denylist: config .ip_range_denylist .iter() diff --git a/src/service/config/mod.rs b/src/service/config/mod.rs index a7effeb26..429666f15 100644 --- a/src/service/config/mod.rs +++ b/src/service/config/mod.rs @@ -6,7 +6,6 @@ config::{Config, check}, error, }; -use url::Url; use crate::registration_tokens::{ValidToken, ValidTokenSource}; @@ -53,18 +52,6 @@ pub fn get_config_file_token(&self) -> Option { .map(|token| ValidToken { token, source: ValidTokenSource::Config }) } - /// Get the base domain to use for user-facing URLs. - #[must_use] - pub fn get_client_domain(&self) -> Url { - self.well_known.client.clone().unwrap_or_else(|| { - let host = self.server_name.host(); - format!("https://{host}") - .as_str() - .try_into() - .expect("server name should be a valid host") - }) - } - fn handle_reload(&self) -> Result { if self.server.config.config_reload_signal { #[cfg(all(feature = "systemd", target_os = "linux"))] diff --git a/src/service/pusher/mod.rs b/src/service/pusher/mod.rs index 3a15fd6bc..df82d7bb4 100644 --- a/src/service/pusher/mod.rs +++ b/src/service/pusher/mod.rs @@ -1,9 +1,11 @@ +mod webpush; + use std::{fmt::Debug, mem, sync::Arc}; use bytes::BytesMut; use conduwuit::utils::response::LimitReadExt; use conduwuit_core::{ - Err, Event, Result, debug_warn, err, trace, + Err, Event, Result, Server, debug_warn, err, trace, utils::{stream::TryIgnore, string_from_bytes}, warn, }; @@ -15,7 +17,6 @@ api::{ IncomingResponseExt, OutgoingRequest, OutgoingRequestExt, auth_scheme::NoAuthentication, - client::push::{Pusher, PusherKind, set_pusher}, path_builder::SinglePath, push_gateway::send_event_notification::{ self, @@ -27,18 +28,33 @@ Action, HighlightTweakValue, PushConditionPowerLevelsCtx, PushConditionRoomCtx, PushFormat, Ruleset, Tweak, }, - serde::Raw, + serde::{JsonObject, Raw}, uint, }; +use ruminuwuity::pushers::{Pusher, PusherKind, set_pusher::v3::PusherAction}; +pub use self::webpush::ActivationOutcome; +use self::webpush::Vapid; use crate::{Dep, client, config, globals, rooms, sending, users}; +const PUSH_GATEWAY_SCHEMES: &[&str] = &["http", "https"]; +const WEBPUSH_SCHEMES: &[&str] = &["https"]; + +/// Whether the pusher opted out of badge counts (MSC4076). +pub(crate) fn badge_counts_disabled(data: &JsonObject) -> bool { + data.contains_key("org.matrix.msc4076.disable_badge_count") + || data.contains_key("disable_badge_count") +} + pub struct Service { db: Data, services: Services, + me: std::sync::Weak, + vapid: Vapid, } struct Services { + server: Arc, globals: Dep, config: Dep, client: Dep, @@ -50,17 +66,24 @@ struct Services { struct Data { senderkey_pusher: Arc, + senderkey_webpushbackoff: Arc, + senderkey_webpushstate: Arc, pushkey_deviceid: Arc, } impl crate::Service for Service { fn build(args: crate::Args<'_>) -> Result> { - Ok(Arc::new(Self { + let vapid = Vapid::init(args.db, args.server.config.webpush_vapid_contact().to_string())?; + + Ok(Arc::new_cyclic(|me| Self { db: Data { senderkey_pusher: args.db["senderkey_pusher"].clone(), + senderkey_webpushbackoff: args.db["senderkey_webpushbackoff"].clone(), + senderkey_webpushstate: args.db["senderkey_webpushstate"].clone(), pushkey_deviceid: args.db["pushkey_deviceid"].clone(), }, services: Services { + server: args.server.clone(), globals: args.depend::("globals"), client: args.depend::("client"), config: args.depend::("config"), @@ -70,6 +93,8 @@ fn build(args: crate::Args<'_>) -> Result> { users: args.depend::("users"), sending: args.depend::("sending"), }, + me: me.clone(), + vapid, })) } @@ -81,71 +106,87 @@ pub async fn set_pusher( &self, sender: &UserId, sender_device: &DeviceId, - pusher: &set_pusher::v3::PusherAction, - ) -> Result { - match pusher { - | set_pusher::v3::PusherAction::Post(data) => { - let pushkey = data.pusher.ids.pushkey.as_str(); - - if pushkey.len() > 512 { - return Err!(Request(InvalidParam( - "Push key length cannot be greater than 512 bytes." - ))); - } - - if data.pusher.ids.app_id.as_str().len() > 64 { - return Err!(Request(InvalidParam( - "App ID length cannot be greater than 64 bytes." - ))); - } - - // add some validation to the pusher URL - let pusher_kind = &data.pusher.kind; - if let PusherKind::Http(http) = pusher_kind { - let url = &http.url; - let url = url::Url::parse(&http.url).map_err(|e| { - err!(Request(InvalidParam( - warn!(%url, "HTTP pusher URL is not a valid URL: {e}") - ))) - })?; - - if ["http", "https"] - .iter() - .all(|&scheme| scheme != url.scheme().to_lowercase()) - { - return Err!(Request(InvalidParam( - warn!(%url, "HTTP pusher URL is not a valid HTTP/HTTPS URL") - ))); - } - - if let Ok(ip) = - IPAddress::parse(url.host_str().expect("URL previously validated")) - { - if !self.services.client.valid_cidr_range(&ip) { - return Err!(Request(InvalidParam( - warn!(%url, "HTTP pusher URL is a forbidden remote address") - ))); - } - } - } - - let pushkey = data.pusher.ids.pushkey.as_str(); - let key = (sender, pushkey); - self.db.senderkey_pusher.put(key, Json(pusher)); - self.db.pushkey_deviceid.insert(pushkey, sender_device); - }, - | set_pusher::v3::PusherAction::Delete(ids) => { + action: &PusherAction, + ) -> Result { + let pusher = match action { + | PusherAction::Post(pusher) => pusher, + | PusherAction::Delete(ids) => { self.delete_pusher(sender, ids.pushkey.as_str()).await; + return Ok(false); }, - | _ => return Err!(Request(InvalidParam("Unknown pusher action"))), + }; + + let pushkey = pusher.ids.pushkey.as_str(); + if pushkey.len() > 512 { + return Err!(Request(InvalidParam( + "Push key length cannot be greater than 512 bytes." + ))); } - Ok(()) + if pusher.ids.app_id.as_str().len() > 64 { + return Err!(Request(InvalidParam("App ID length cannot be greater than 64 bytes."))); + } + + match &pusher.kind { + | PusherKind::Http(http) => { + self.validate_push_url(&http.url, PUSH_GATEWAY_SCHEMES)?; + }, + | PusherKind::WebPush(data) => { + self.validate_push_url(&data.url, WEBPUSH_SCHEMES)?; + webpush::decode_pushkey(pushkey)?; + webpush::decode_auth(&data.auth)?; + + if let Some(default_payload) = data.data.get("default_payload") { + if serde_json::to_vec(default_payload)?.len() > webpush::MAX_DEFAULT_PAYLOAD { + return Err!(Request(TooLarge( + "Default payload cannot be greater than {} bytes.", + webpush::MAX_DEFAULT_PAYLOAD + ))); + } + } + }, + | PusherKind::Email(_) | PusherKind::Custom { .. } => (), + } + + let key = (sender, pushkey); + let previous: Option = self + .db + .senderkey_pusher + .qry(&key) + .await + .deserialized() + .map_err(|e| { + if !e.is_not_found() { + warn!(%pushkey, "Replacing a stored pusher that could not be read: {e}"); + } + }) + .ok(); + let needs_activation = self + .webpush_needs_activation(sender, pusher, previous.as_ref()) + .await; + + if !needs_activation && pusher.kind.as_webpush().is_none() { + self.db.senderkey_webpushstate.del(key); + self.db.senderkey_webpushbackoff.del(key); + } + + self.db.senderkey_pusher.put(key, Json(pusher)); + self.db.pushkey_deviceid.insert(pushkey, sender_device); + + // Must run after the put: it delivers straight away, and a rejection + // from the push service deletes the pusher it is validating. + if needs_activation { + self.begin_webpush_activation(sender, pusher); + } + + Ok(needs_activation) } pub async fn delete_pusher(&self, sender: &UserId, pushkey: &str) { let key = (sender, pushkey); self.db.senderkey_pusher.del(key); + self.db.senderkey_webpushbackoff.del(key); + self.db.senderkey_webpushstate.del(key); self.db.pushkey_deviceid.remove(pushkey); self.services @@ -192,6 +233,35 @@ pub fn get_pushkeys<'a>( .map(ToOwned::to_owned) } + /// Checks that a pusher URL is well-formed and not pointing at a forbidden + /// address. + fn validate_push_url(&self, url: &str, allowed: &[&str]) -> Result { + let parsed = url::Url::parse(url).map_err(|e| { + err!(Request(InvalidParam(warn!(%url, "Pusher URL is not a valid URL: {e}")))) + })?; + + let scheme = parsed.scheme().to_lowercase(); + if !allowed.contains(&scheme.as_str()) { + return Err!(Request(InvalidParam(warn!( + %url, + "Pusher URL scheme {scheme} is not allowed" + )))); + } + + if let Some(host) = parsed.host_str() { + if let Ok(ip) = IPAddress::parse(host) { + if !self.services.client.valid_cidr_range(&ip) { + return Err!(Request(InvalidParam(warn!( + %url, + "Pusher URL is a forbidden remote address" + )))); + } + } + } + + Ok(parsed) + } + #[tracing::instrument(skip(self, dest, request))] pub async fn send_request(&self, dest: &str, request: T) -> Result where @@ -211,9 +281,41 @@ pub async fn send_request(&self, dest: &str, request: T) -> Result Result>> { + if let Some(url_host) = request.url().host_str() { trace!("Checking request URL for IP"); if let Ok(ip) = IPAddress::parse(url_host) { if !self.services.client.valid_cidr_range(&ip) { @@ -222,69 +324,36 @@ pub async fn send_request(&self, dest: &str, request: T) -> Result { - // reqwest::Response -> http::Response conversion - - trace!("Checking response destination's IP"); - if let Some(remote_addr) = response.remote_addr() { - if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) { - if !self.services.client.valid_cidr_range(&ip) { - return Err!(BadServerResponse( - "Not allowed to send requests to this IP" - )); - } - } + trace!("Checking response destination's IP"); + if let Some(remote_addr) = response.remote_addr() { + if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) { + if !self.services.client.valid_cidr_range(&ip) { + return Err!(BadServerResponse("Not allowed to send requests to this IP")); } - - let status = response.status(); - let mut http_response_builder = http::Response::builder() - .status(status) - .version(response.version()); - mem::swap( - response.headers_mut(), - http_response_builder - .headers_mut() - .expect("http::response::Builder is usable"), - ); - - let body = response - .limit_read( - self.services - .config - .max_request_size - .try_into() - .expect("usize fits into u64"), - ) - .await?; - - if !status.is_success() { - debug_warn!("Push gateway response body: {:?}", string_from_bytes(&body)); - return Err!(BadServerResponse(warn!( - "Push gateway {dest} returned unsuccessful HTTP response: {status}" - ))); - } - - let (parts, body) = http_response_builder - .body(body) - .expect("reqwest body is valid http body") - .into_parts(); - let response = T::IncomingResponse::try_from_http_response( - http::Response::from_parts(parts, body.as_ref()), - ); - response.map_err(|e| { - err!(BadServerResponse(warn!( - "Push gateway {dest} returned invalid response: {e}" - ))) - }) - }, - | Err(e) => { - warn!("Could not send request to pusher {dest}: {e}"); - Err(e.into()) - }, + } } + + let mut result = http::Response::new(Vec::new()); + *result.status_mut() = response.status(); + *result.version_mut() = response.version(); + mem::swap(response.headers_mut(), result.headers_mut()); + + *result.body_mut() = response + .limit_read( + self.services + .config + .max_request_size + .try_into() + .expect("usize fits into u64"), + ) + .await?; + + Ok(result) } #[tracing::instrument(skip(self, user, unread, pusher, ruleset, event))] @@ -315,13 +384,7 @@ pub async fn send_push_notice( let serialized = event.to_format(); for action in self - .get_actions( - user, - &ruleset, - power_levels.clone(), - &serialized, - event.room_id().unwrap(), - ) + .get_actions(user, &ruleset, power_levels.clone(), &serialized, room_id) .await { let n = match action { @@ -343,7 +406,8 @@ pub async fn send_push_notice( } if notify == Some(true) { - self.send_notice(unread, pusher, tweaks, event).await?; + self.send_notice(user, unread, pusher, tweaks, event) + .await?; } // Else the event triggered no actions @@ -406,6 +470,7 @@ pub async fn get_actions<'a>( #[tracing::instrument(skip(self, unread, pusher, tweaks, event))] async fn send_notice( &self, + user: &UserId, unread: UInt, pusher: &Pusher, tweaks: Vec, @@ -417,109 +482,110 @@ async fn send_notice( // TODO: email match &pusher.kind { | PusherKind::Http(http) => { - let url = &http.url; - let url = url::Url::parse(&http.url).map_err(|e| { - err!(Request(InvalidParam( - warn!(%url, "HTTP pusher URL is not a valid URL: {e}") - ))) - })?; + self.validate_push_url(&http.url, PUSH_GATEWAY_SCHEMES)?; - if ["http", "https"] - .iter() - .all(|&scheme| scheme != url.scheme().to_lowercase()) - { - return Err!(Request(InvalidParam( - warn!(%url, "HTTP pusher URL is not a valid HTTP/HTTPS URL") - ))); - } - - if let Ok(ip) = - IPAddress::parse(url.host_str().expect("URL previously validated")) - { - if !self.services.client.valid_cidr_range(&ip) { - return Err!(Request(InvalidParam( - warn!(%url, "HTTP pusher URL is a forbidden remote address") - ))); - } - } - - // TODO (timo): can pusher/devices have conflicting formats - let event_id_only = http.format == Some(PushFormat::EventIdOnly); - - let mut device = - Device::new(pusher.ids.app_id.clone(), pusher.ids.pushkey.clone()); - device.data.data.clone_from(&http.data); - device.data.format.clone_from(&http.format); - - // Tweaks are only added if the format is NOT event_id_only - if !event_id_only { - device.tweaks.clone_from(&tweaks); - } - - let d = vec![device]; - let mut notify = Notification::new(d); - - notify.event_id = Some(event.event_id().to_owned()); - notify.room_id = Some(event.room_id().unwrap().to_owned()); - if http - .data - .get("org.matrix.msc4076.disable_badge_count") - .is_none() && http.data.get("disable_badge_count").is_none() - { - notify.counts = NotificationCounts::new(unread, uint!(0)); - } else { - // counts will not be serialised if it's the default (0, 0) - // skip_serializing_if = "NotificationCounts::is_default" - notify.counts = NotificationCounts::default(); - } - - if !event_id_only { - if *event.kind() == TimelineEventType::RoomEncrypted - || tweaks.iter().any(|t| { - matches!( - t, - Tweak::Highlight(HighlightTweakValue::Yes) | Tweak::Sound(_) - ) - }) { - notify.prio = NotificationPriority::High; - } else { - notify.prio = NotificationPriority::Low; - } - notify.sender = Some(event.sender().to_owned()); - notify.event_type = Some(event.kind().to_owned()); - notify.content = serde_json::value::to_raw_value(event.content()).ok(); - - if *event.kind() == TimelineEventType::RoomMember { - notify.user_is_target = - event.state_key() == Some(event.sender().as_str()); - } - - notify.sender_display_name = - self.services.users.displayname(event.sender()).await.ok(); - - notify.room_name = self - .services - .state_accessor - .get_name(event.room_id().unwrap()) - .await - .ok(); - - notify.room_alias = self - .services - .state_accessor - .get_canonical_alias(event.room_id().unwrap()) - .await - .ok(); - } + let notify = self + .build_notification( + pusher, + &http.data, + http.format.as_ref(), + &tweaks, + event, + unread, + ) + .await; self.send_request(&http.url, send_event_notification::v1::Request::new(notify)) .await?; Ok(()) }, - // TODO: Handle email - //PusherKind::Email(_) => Ok(()), - | _ => Ok(()), + | PusherKind::WebPush(data) => { + let notify = self + .build_notification( + pusher, + &data.data, + data.format.as_ref(), + &tweaks, + event, + unread, + ) + .await; + + self.send_webpush_notice(user, pusher, data, notify).await + }, + | PusherKind::Email(_) | PusherKind::Custom { .. } => Ok(()), } } + + async fn build_notification( + &self, + pusher: &Pusher, + data: &JsonObject, + format: Option<&PushFormat>, + tweaks: &[Tweak], + event: &E, + unread: UInt, + ) -> Notification + where + E: Event + Send + Sync, + { + // TODO (timo): can pusher/devices have conflicting formats + let event_id_only = format == Some(&PushFormat::EventIdOnly); + + let mut device = Device::new(pusher.ids.app_id.clone(), pusher.ids.pushkey.clone()); + device.data.data.clone_from(data); + device.data.format = format.cloned(); + + // Tweaks are only added if the format is NOT event_id_only + if !event_id_only { + device.tweaks = tweaks.to_owned(); + } + + let mut notify = Notification::new(vec![device]); + + notify.event_id = Some(event.event_id().to_owned()); + notify.room_id = event.room_id().map(ToOwned::to_owned); + if badge_counts_disabled(data) { + // counts will not be serialised if it's the default (0, 0) + // skip_serializing_if = "NotificationCounts::is_default" + notify.counts = NotificationCounts::default(); + } else { + notify.counts = NotificationCounts::new(unread, uint!(0)); + } + + if !event_id_only { + if *event.kind() == TimelineEventType::RoomEncrypted + || tweaks.iter().any(|t| { + matches!(t, Tweak::Highlight(HighlightTweakValue::Yes) | Tweak::Sound(_)) + }) { + notify.prio = NotificationPriority::High; + } else { + notify.prio = NotificationPriority::Low; + } + notify.sender = Some(event.sender().to_owned()); + notify.event_type = Some(event.kind().to_owned()); + notify.content = serde_json::value::to_raw_value(event.content()).ok(); + + if *event.kind() == TimelineEventType::RoomMember { + notify.user_is_target = event.state_key() == Some(event.sender().as_str()); + } + + notify.sender_display_name = + self.services.users.displayname(event.sender()).await.ok(); + + if let Some(room_id) = event.room_id() { + notify.room_name = self.services.state_accessor.get_name(room_id).await.ok(); + + notify.room_alias = self + .services + .state_accessor + .get_canonical_alias(room_id) + .await + .ok(); + } + } + + notify + } } diff --git a/src/service/pusher/webpush/helpers.rs b/src/service/pusher/webpush/helpers.rs new file mode 100644 index 000000000..7ac455e3f --- /dev/null +++ b/src/service/pusher/webpush/helpers.rs @@ -0,0 +1,208 @@ +//! Web Push key validation, token comparison, and payload shaping helpers. + +use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; +use conduwuit_core::{Err, Result, err}; +use ruma::{api::push_gateway::send_event_notification::v1::Notification, serde::JsonObject}; +use serde_json::Value as JsonValue; +use web_push_native::{Auth, p256::PublicKey}; + +use crate::pusher::badge_counts_disabled; + +const MAX_PAYLOAD: usize = 3900; +const MAX_BODY_LENGTH: usize = 1000; +const MAX_CIPHERTEXT_LENGTH: usize = 2000; + +pub(crate) fn decode_pushkey(pushkey: &str) -> Result { + let bytes = URL_SAFE_NO_PAD + .decode(pushkey) + .map_err(|e| err!(Request(InvalidParam("Push key is not URL-safe base64: {e}"))))?; + if bytes.len() != 65 || bytes.first() != Some(&0x04) { + return Err!(Request(InvalidParam( + "Push key must be a P-256 public key in the uncompressed form" + ))); + } + + PublicKey::from_sec1_bytes(&bytes) + .map_err(|e| err!(Request(InvalidParam("Push key is not a valid P-256 public key: {e}")))) +} + +pub(crate) fn decode_auth(auth: &str) -> Result { + let bytes = URL_SAFE_NO_PAD + .decode(auth) + .map_err(|e| err!(Request(InvalidParam("Auth secret is not URL-safe base64: {e}"))))?; + let bytes: [u8; 16] = bytes + .try_into() + .map_err(|_| err!(Request(InvalidParam("Auth secret must be 16 bytes"))))?; + + Ok(Auth::from(bytes)) +} + +pub(super) fn constant_time_eq(a: &str, b: &str) -> bool { + a.len() == b.len() + && a.bytes() + .zip(b.bytes()) + .fold(0_u8, |acc, (x, y)| acc | (x ^ y)) + == 0 +} + +pub(super) fn build_payload(notify: &Notification, data: &JsonObject) -> Result> { + let JsonValue::Object(mut payload) = serde_json::to_value(notify)? else { + return Err!("Notification did not serialize to an object"); + }; + payload.remove("devices"); + + if let Some(JsonValue::Object(counts)) = payload.remove("counts") { + for key in ["unread", "missed_calls"] { + if let Some(count) = counts.get(key) { + payload.insert(key.to_owned(), count.clone()); + } + } + } else if !badge_counts_disabled(data) { + payload.insert("unread".to_owned(), 0.into()); + } + if let Some(JsonValue::Object(default_payload)) = data.get("default_payload") { + for (key, value) in default_payload { + payload.insert(key.clone(), value.clone()); + } + } + if let Some(JsonValue::Object(content)) = payload.get_mut("content") { + if let Some(JsonValue::String(body)) = content.get("body") + && body.len() > MAX_BODY_LENGTH + { + content.insert( + "body".to_owned(), + format!("{}…", truncate_bytes(body, MAX_BODY_LENGTH)).into(), + ); + } + if content + .get("ciphertext") + .and_then(JsonValue::as_str) + .is_some_and(|ciphertext| ciphertext.len() > MAX_CIPHERTEXT_LENGTH) + { + content.remove("ciphertext"); + } + } + + let mut encoded = serde_json::to_vec(&payload)?; + if encoded.len() > MAX_PAYLOAD { + payload.remove("content"); + encoded = serde_json::to_vec(&payload)?; + } + if encoded.len() > MAX_PAYLOAD { + return Err!(Request(TooLarge("Web push payload is too large to deliver"))); + } + + Ok(encoded) +} + +fn truncate_bytes(text: &str, max: usize) -> String { + let mut truncated = String::with_capacity(max); + for ch in text.chars() { + if truncated.len().saturating_add(ch.len_utf8()) > max { + break; + } + truncated.push(ch); + } + + truncated +} + +#[cfg(test)] +mod tests { + use ruma::{ + api::push_gateway::send_event_notification::v1::{Notification, NotificationCounts}, + push::PushFormat, + serde::JsonObject, + uint, + }; + use serde_json::{Value as JsonValue, json}; + + use super::{MAX_BODY_LENGTH, build_payload, constant_time_eq}; + + fn payload_of(notify: &Notification, data: &JsonObject) -> JsonObject { + let encoded = build_payload(notify, data).unwrap(); + match serde_json::from_slice(&encoded).unwrap() { + | JsonValue::Object(payload) => payload, + | other => panic!("expected an object, got {other}"), + } + } + + fn notification_with_body(body: &str) -> Notification { + let mut notify = Notification::new(vec![]); + notify.content = Some(serde_json::value::to_raw_value(&json!({ "body": body })).unwrap()); + + notify + } + + #[test] + fn flattens_counts_and_drops_devices() { + let mut notify = Notification::new(vec![]); + notify.counts = NotificationCounts::new(uint!(3), uint!(1)); + let payload = payload_of(¬ify, &JsonObject::new()); + + assert!(!payload.contains_key("devices")); + assert!(!payload.contains_key("counts")); + assert_eq!(payload.get("unread"), Some(&json!(3))); + assert_eq!(payload.get("missed_calls"), Some(&json!(1))); + } + + #[test] + fn keeps_zero_unread_count() { + let payload = payload_of(&Notification::new(vec![]), &JsonObject::new()); + + assert_eq!(payload.get("unread"), Some(&json!(0))); + } + + #[test] + fn omits_count_when_badges_are_disabled() { + let mut data = JsonObject::new(); + data.insert("org.matrix.msc4076.disable_badge_count".to_owned(), json!(true)); + let payload = payload_of(&Notification::new(vec![]), &data); + + assert!(!payload.contains_key("unread")); + } + + #[test] + fn merges_default_payload() { + let mut data = JsonObject::new(); + data.insert("default_payload".to_owned(), json!({ "session_id": "abc" })); + data.insert("format".to_owned(), json!(PushFormat::EventIdOnly)); + let payload = payload_of(&Notification::new(vec![]), &data); + + assert_eq!(payload.get("session_id"), Some(&json!("abc"))); + assert!(!payload.contains_key("format")); + } + + #[test] + fn truncates_long_body() { + let payload = payload_of(¬ification_with_body(&"a".repeat(2000)), &JsonObject::new()); + let body = payload["content"]["body"].as_str().unwrap(); + + assert!(body.len() <= MAX_BODY_LENGTH + '…'.len_utf8()); + assert!(body.ends_with('…')); + } + + #[test] + fn truncates_multibyte_body_by_bytes() { + let payload = payload_of(¬ification_with_body(&"é".repeat(2000)), &JsonObject::new()); + let body = payload["content"]["body"].as_str().unwrap(); + + assert!(body.len() <= MAX_BODY_LENGTH + '…'.len_utf8()); + assert!(body.chars().count() < 2000); + } + + #[test] + fn rejects_payload_that_cannot_be_shrunk() { + let mut data = JsonObject::new(); + data.insert("default_payload".to_owned(), json!({ "pad": "x".repeat(4000) })); + + build_payload(&Notification::new(vec![]), &data).unwrap_err(); + } + + #[test] + fn token_comparison() { + assert!(constant_time_eq("abc", "abc")); + assert!(!constant_time_eq("abc", "abz")); + assert!(!constant_time_eq("abc", "abcd")); + } +} diff --git a/src/service/pusher/webpush/mod.rs b/src/service/pusher/webpush/mod.rs new file mode 100644 index 000000000..f09f65890 --- /dev/null +++ b/src/service/pusher/webpush/mod.rs @@ -0,0 +1,432 @@ +mod helpers; +mod vapid; + +use std::time::{Duration, SystemTime}; + +use conduwuit_core::{Err, Result, debug, debug_warn, err, utils, warn}; +use conduwuit_database::{Deserialized, Json}; +use futures::StreamExt; +pub(super) use helpers::{decode_auth, decode_pushkey}; +use http::StatusCode; +use ruma::{ + UserId, + api::push_gateway::send_event_notification::v1::{Notification, NotificationPriority}, + push::PushFormat, +}; +use ruminuwuity::pushers::{Pusher, WebPushPusherData}; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use tokio::time::sleep; +pub(super) use vapid::Vapid; +use web_push_native::WebPushBuilder; + +use self::helpers::{build_payload, constant_time_eq}; +use super::{Service, WEBPUSH_SCHEMES}; + +const TTL: Duration = Duration::from_hours(48); + +const ACK_TOKEN_TTL_MS: u64 = 5 * 60 * 1000; +const ACK_TOKEN_LENGTH: usize = 32; + +pub(super) const MAX_DEFAULT_PAYLOAD: usize = 1024; + +const DELIVERY_ATTEMPTS: u32 = 3; +const RETRY_BASE: Duration = Duration::from_secs(2); +const RETRY_MAX: Duration = Duration::from_secs(20); + +const BACKOFF_BASE: Duration = Duration::from_mins(1); +const BACKOFF_MAX: Duration = Duration::from_hours(24); + +#[derive(Clone, Copy, Debug)] +pub enum ActivationOutcome { + Activated, + NoSuchPusher, + UnknownToken, + Expired, +} + +enum Attempt { + Delivered, + + /// The subscription is gone; delete the pusher. + Rejected, + + /// Transient, carrying the delay the push service asked for. + Failed(Option), +} + +#[derive(Clone, Default, Deserialize, Serialize)] +pub(super) struct WebPushState { + pub(super) activated: bool, + + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) ack_token: Option, + + #[serde(default)] + pub(super) token_expires_at: u64, +} + +#[derive(Clone, Default, Deserialize, Serialize)] +struct WebPushBackoff { + #[serde(default)] + next_attempt_at: u64, + + #[serde(default)] + failure_count: u32, +} + +impl Service { + pub fn webpush_vapid_public_key(&self) -> &str { self.vapid.public_key() } + + /// Reports whether a Web Push pusher has completed its activation + /// handshake. + pub async fn is_webpush_activated(&self, sender: &UserId, pushkey: &str) -> bool { + self.webpush_state(sender, pushkey).await.activated + } + + pub(super) async fn webpush_state(&self, sender: &UserId, pushkey: &str) -> WebPushState { + self.db + .senderkey_webpushstate + .qry(&(sender, pushkey)) + .await + .deserialized() + .unwrap_or_default() + } + + fn put_webpush_state(&self, sender: &UserId, pushkey: &str, state: &WebPushState) { + self.db + .senderkey_webpushstate + .put((sender, pushkey), Json(state)); + } + + pub(super) async fn webpush_needs_activation( + &self, + sender: &UserId, + pusher: &Pusher, + previous: Option<&Pusher>, + ) -> bool { + let Some(data) = pusher.kind.as_webpush() else { + return false; + }; + + let Some(old) = previous.and_then(|previous| previous.kind.as_webpush()) else { + return true; + }; + + old.url != data.url + || old.auth != data.auth + || !self + .webpush_state(sender, pusher.ids.pushkey.as_str()) + .await + .activated + } + + pub(super) fn begin_webpush_activation(&self, sender: &UserId, pusher: &Pusher) { + // Send a one-time encrypted token before allowing notification + // delivery. + let token = utils::rand::string(ACK_TOKEN_LENGTH); + let state = WebPushState { + activated: false, + ack_token: Some(token.clone()), + token_expires_at: utils::millis_since_unix_epoch().saturating_add(ACK_TOKEN_TTL_MS), + }; + + self.put_webpush_state(sender, pusher.ids.pushkey.as_str(), &state); + self.db + .senderkey_webpushbackoff + .del((sender, pusher.ids.pushkey.as_str())); + + let Some(service) = self.me.upgrade() else { + return; + }; + let sender = sender.to_owned(); + let pusher = pusher.clone(); + let payload = json!({ "app_id": pusher.ids.app_id, "ack_token": token }); + + self.services.server.runtime().spawn(async move { + let Some(data) = pusher.kind.as_webpush() else { + return; + }; + + if let Err(e) = service + .deliver_webpush( + &sender, + &pusher, + data, + payload.to_string().into_bytes(), + NotificationPriority::High, + ) + .await + { + debug_warn!( + %sender, app_id = %pusher.ids.app_id, + "Failed to send web push validation token: {e}" + ); + } + }); + } + + #[tracing::instrument(skip(self, ack_token), level = "debug")] + /// Activates the pusher whose pending token matches the client + /// acknowledgement. + pub async fn activate_webpush_pusher( + &self, + sender: &UserId, + app_id: &str, + ack_token: &str, + ) -> Result { + let candidates: Vec = self + .get_pushkeys(sender) + .filter_map(async |pushkey| { + let pusher = self.get_pusher(sender, &pushkey).await.ok()?; + + (pusher.kind.as_webpush().is_some() && pusher.ids.app_id == app_id) + .then_some(pushkey) + }) + .collect() + .await; + + if candidates.is_empty() { + return Ok(ActivationOutcome::NoSuchPusher); + } + + for pushkey in candidates { + let mut state = self.webpush_state(sender, &pushkey).await; + if !state + .ack_token + .as_deref() + .is_some_and(|token| constant_time_eq(token, ack_token)) + { + continue; + } + + state.ack_token = None; + if state.token_expires_at < utils::millis_since_unix_epoch() { + self.put_webpush_state(sender, &pushkey, &state); + return Ok(ActivationOutcome::Expired); + } + + state.activated = true; + self.put_webpush_state(sender, &pushkey, &state); + self.db.senderkey_webpushbackoff.del((sender, &pushkey)); + + return Ok(ActivationOutcome::Activated); + } + + Ok(ActivationOutcome::UnknownToken) + } + + #[tracing::instrument(skip(self, pusher, data, notify), level = "debug")] + /// Delivers a notification only after the pusher has completed activation. + pub(super) async fn send_webpush_notice( + &self, + user: &UserId, + pusher: &Pusher, + data: &WebPushPusherData, + notify: Notification, + ) -> Result { + let pushkey = pusher.ids.pushkey.as_str(); + if !self.webpush_state(user, pushkey).await.activated { + debug!(%user, %pushkey, "Web push pusher is not activated yet, skipping"); + return Ok(()); + } + + let backoff: WebPushBackoff = self + .db + .senderkey_webpushbackoff + .qry(&(user, pushkey)) + .await + .deserialized() + .unwrap_or_default(); + + if utils::millis_since_unix_epoch() < backoff.next_attempt_at { + debug!(%user, %pushkey, "Web push pusher is backing off, dropping notification"); + return Ok(()); + } + + let prio = if data.format.as_ref() == Some(&PushFormat::EventIdOnly) { + NotificationPriority::Low + } else { + notify.prio.clone() + }; + let payload = build_payload(¬ify, &data.data)?; + + self.deliver_webpush(user, pusher, data, payload, prio) + .await + } + + #[tracing::instrument(skip(self, pusher, data, payload), level = "debug")] + /// Encrypts and sends a Web Push request, retrying transient failures and + /// deleting rejected subscriptions. + async fn deliver_webpush( + &self, + user: &UserId, + pusher: &Pusher, + data: &WebPushPusherData, + payload: Vec, + prio: NotificationPriority, + ) -> Result { + let endpoint = self.validate_push_url(&data.url, WEBPUSH_SCHEMES)?; + let authorization = self.vapid.authorization(&endpoint)?; + let pushkey = pusher.ids.pushkey.as_str(); + + let mut attempt = 1_u32; + loop { + match self + .attempt_webpush(user, pusher, data, &payload, &authorization, &prio) + .await? + { + | Attempt::Delivered => { + self.db.senderkey_webpushbackoff.del((user, pushkey)); + + return Ok(()); + }, + | Attempt::Rejected => { + self.delete_pusher(user, pushkey).await; + + return Ok(()); + }, + | Attempt::Failed(retry_after) => { + if attempt >= DELIVERY_ATTEMPTS { + self.back_off_webpush(user, pushkey, retry_after).await; + + return Err!(BadServerResponse(warn!( + %user, %pushkey, + "Push service did not accept the notification after {attempt} attempts" + ))); + } + + let delay = retry_after + .unwrap_or_else(|| RETRY_BASE.saturating_mul(attempt)) + .min(RETRY_MAX); + sleep(delay).await; + attempt = attempt.saturating_add(1); + }, + } + } + } + + /// The message is re-encrypted per attempt, since each needs its own + /// ephemeral key. + async fn attempt_webpush( + &self, + user: &UserId, + pusher: &Pusher, + data: &WebPushPusherData, + payload: &[u8], + authorization: &str, + prio: &NotificationPriority, + ) -> Result { + let pushkey = pusher.ids.pushkey.as_str(); + let mut request = WebPushBuilder::new( + data.url + .parse() + .map_err(|e| err!(Request(InvalidParam("Pusher URL is not a valid URI: {e}"))))?, + decode_pushkey(pushkey)?, + decode_auth(&data.auth)?, + ) + .with_valid_duration(TTL) + .build(payload.to_vec()) + .map_err(|e| err!("Failed to encrypt web push message: {e}"))?; + + let headers = request.headers_mut(); + headers.insert(http::header::AUTHORIZATION, authorization.parse()?); + headers.insert( + "urgency", + if *prio == NotificationPriority::High { + "high" + } else { + "normal" + } + .parse()?, + ); + + let response = match self + .execute_pusher_request( + &self.services.client.webpush, + reqwest::Request::try_from(request)?, + ) + .await + { + | Ok(response) => response, + | Err(e) => { + debug_warn!(%user, %pushkey, "Could not reach the push service: {e}"); + + return Ok(Attempt::Failed(None)); + }, + }; + + let status = response.status(); + if status.is_success() { + return Ok(Attempt::Delivered); + } + + if matches!(status, StatusCode::FORBIDDEN | StatusCode::NOT_FOUND | StatusCode::GONE) { + warn!( + %user, %pushkey, + "Push service rejected the subscription ({status}), deleting pusher" + ); + + return Ok(Attempt::Rejected); + } + + if status.is_redirection() { + warn!(%user, %pushkey, "Push service redirected ({status}); not following it"); + } + + debug_warn!(%user, %pushkey, "Push service returned unsuccessful HTTP response: {status}"); + + Ok(Attempt::Failed( + (status == StatusCode::TOO_MANY_REQUESTS) + .then(|| retry_after(response.headers())) + .flatten(), + )) + } + + async fn back_off_webpush( + &self, + user: &UserId, + pushkey: &str, + retry_after: Option, + ) { + let key = (user, pushkey); + let mut backoff: WebPushBackoff = self + .db + .senderkey_webpushbackoff + .qry(&key) + .await + .deserialized() + .unwrap_or_default(); + + backoff.failure_count = backoff.failure_count.saturating_add(1); + + let delay = retry_after.map_or_else( + || { + BACKOFF_BASE + .saturating_mul(1_u32 << backoff.failure_count.min(10)) + .min(BACKOFF_MAX) + }, + |retry_after| retry_after.min(BACKOFF_MAX), + ); + + backoff.next_attempt_at = utils::millis_since_unix_epoch() + .saturating_add(delay.as_millis().try_into().unwrap_or(u64::MAX)); + + self.db.senderkey_webpushbackoff.put(key, Json(&backoff)); + } +} + +/// RFC 9110 allows `Retry-After` to be either a delay in seconds or an HTTP +/// date. +fn retry_after(headers: &http::HeaderMap) -> Option { + let value = headers.get(http::header::RETRY_AFTER)?.to_str().ok()?; + + if let Ok(secs) = value.parse::() { + return Some(Duration::from_secs(secs)); + } + + httpdate::parse_http_date(value) + .ok()? + .duration_since(SystemTime::now()) + .ok() +} diff --git a/src/service/pusher/webpush/vapid.rs b/src/service/pusher/webpush/vapid.rs new file mode 100644 index 000000000..ad9a83cb7 --- /dev/null +++ b/src/service/pusher/webpush/vapid.rs @@ -0,0 +1,97 @@ +//! VAPID key management and authorization-header generation for Web Push. + +use std::{collections::HashMap, sync::Arc}; + +use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; +use conduwuit_core::{Result, SyncRwLock, debug_info, err, utils}; +use conduwuit_database::Database; +use p256::{ + SecretKey, + ecdsa::{Signature, SigningKey, signature::Signer}, + elliptic_curve::sec1::ToEncodedPoint, +}; +use serde_json::json; + +const VAPID_TTL_SECS: u64 = 12 * 60 * 60; +const VAPID_REFRESH_MARGIN_SECS: u64 = 60; + +/// The server's VAPID keypair and cached authorization headers. +pub(crate) struct Vapid { + secret: SecretKey, + public_key: String, + contact: String, + authorizations: SyncRwLock>, +} + +impl Vapid { + /// Loads the persistent VAPID keypair, generating one on first use. + pub(crate) fn init(db: &Arc, contact: String) -> Result { + let secret = match db["global"].get_blocking(b"vapid_secret_key") { + | Ok(ref val) => SecretKey::from_slice(val).map_err(|e| { + err!(Database(error!( + "Stored VAPID key is invalid: {e}. Refusing to replace it, as that would \ + invalidate every existing web push subscription." + ))) + })?, + | Err(e) if !e.is_not_found() => return Err(e), + | Err(_) => { + let secret = loop { + if let Ok(secret) = SecretKey::from_slice(&rand::random::<[u8; 32]>()) { + break secret; + } + }; + + db["global"].raw_put(b"vapid_secret_key", &*secret.to_bytes()); + debug_info!("Generated new VAPID keypair for web push"); + secret + }, + }; + + let public_key = + URL_SAFE_NO_PAD.encode(secret.public_key().to_encoded_point(false).as_bytes()); + + Ok(Self { + secret, + public_key, + contact, + authorizations: SyncRwLock::new(HashMap::new()), + }) + } + + pub(super) fn public_key(&self) -> &str { &self.public_key } + + /// Returns a cached or freshly signed VAPID authorization for an endpoint. + pub(super) fn authorization(&self, endpoint: &url::Url) -> Result { + let audience = endpoint.origin().ascii_serialization(); + let now = utils::millis_since_unix_epoch().saturating_div(1000); + + if let Some((header, expires_at)) = self.authorizations.read().get(&audience) + && now.saturating_add(VAPID_REFRESH_MARGIN_SECS) < *expires_at + { + return Ok(header.clone()); + } + + let expires_at = now.saturating_add(VAPID_TTL_SECS); + let header = self.sign(&audience, expires_at)?; + let mut authorizations = self.authorizations.write(); + authorizations.retain(|_, (_, expires_at)| now < *expires_at); + authorizations.insert(audience, (header.clone(), expires_at)); + + Ok(header) + } + + fn sign(&self, audience: &str, expires_at: u64) -> Result { + let header = + URL_SAFE_NO_PAD.encode(serde_json::to_vec(&json!({ "typ": "JWT", "alg": "ES256" }))?); + let claims = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&json!({ + "aud": audience, + "exp": expires_at, + "sub": self.contact, + }))?); + let signing_input = format!("{header}.{claims}"); + let signature: Signature = SigningKey::from(&self.secret).sign(signing_input.as_bytes()); + let signature = URL_SAFE_NO_PAD.encode(signature.to_bytes()); + + Ok(format!("vapid t={signing_input}.{signature},k={}", self.public_key)) + } +}