rustical_dav_push/
lib.rs

1#![warn(clippy::all, clippy::pedantic, clippy::nursery)]
2#![allow(clippy::missing_errors_doc)]
3mod extension;
4mod prop;
5pub mod register;
6use base64::Engine;
7use chrono::Utc;
8use derive_more::Constructor;
9pub use extension::*;
10use http::{HeaderValue, Method, header};
11pub use prop::*;
12use reqwest::{Body, Url};
13use rustical_store::{CollectionOperation, CollectionOperationInfo};
14use rustical_xml::{XmlRootTag, XmlSerialize, XmlSerializeRoot};
15use std::{collections::HashMap, sync::Arc, time::Duration};
16use tokio::sync::mpsc::Receiver;
17use tracing::{error, info, warn};
18
19mod endpoints;
20pub use endpoints::subscription_service;
21
22mod store;
23pub use store::*;
24
25mod subscription;
26pub use subscription::*;
27
28#[derive(XmlSerialize, Debug)]
29pub struct ContentUpdate {
30    #[xml(ns = "rustical_dav::namespace::NS_DAV")]
31    sync_token: Option<String>,
32}
33
34#[derive(XmlSerialize, XmlRootTag, Debug)]
35#[xml(root = "push-message", ns = "rustical_dav::namespace::NS_DAVPUSH")]
36#[xml(ns_prefix(
37    rustical_dav::namespace::NS_DAVPUSH = "",
38    rustical_dav::namespace::NS_DAV = "D",
39))]
40struct PushMessage {
41    #[xml(ns = "rustical_dav::namespace::NS_DAVPUSH")]
42    topic: String,
43    #[xml(ns = "rustical_dav::namespace::NS_DAVPUSH")]
44    content_update: Option<ContentUpdate>,
45}
46
47#[derive(Debug, Constructor)]
48pub struct DavPushController<S: SubscriptionStore> {
49    allowed_push_servers: Option<Vec<String>>,
50    sub_store: Arc<S>,
51}
52
53impl<S: SubscriptionStore> DavPushController<S> {
54    pub async fn notifier(&self, mut recv: Receiver<CollectionOperation>) {
55        loop {
56            // Make sure we don't flood the subscribers
57            tokio::time::sleep(Duration::from_secs(10)).await;
58            let mut messages = vec![];
59            recv.recv_many(&mut messages, 100).await;
60
61            // Right now we just have to show the latest content update by topic
62            // This might become more complicated in the future depending on what kind of updates
63            // we add
64            let mut latest_messages = HashMap::new();
65            for message in messages {
66                if matches!(message.data, CollectionOperationInfo::Content { .. }) {
67                    latest_messages.insert(message.topic.clone(), message);
68                }
69            }
70            let messages = latest_messages.into_values();
71
72            for message in messages {
73                self.send_message(message).await;
74            }
75        }
76    }
77
78    #[allow(clippy::cognitive_complexity)]
79    async fn send_message(&self, message: CollectionOperation) {
80        let subscriptions = match self.sub_store.get_subscriptions(&message.topic).await {
81            Ok(subs) => subs,
82            Err(err) => {
83                error!("{err}");
84                return;
85            }
86        };
87
88        if subscriptions.is_empty() {
89            return;
90        }
91
92        if matches!(message.data, CollectionOperationInfo::Delete) {
93            // Collection has been deleted, but we cannot handle that
94            return;
95        }
96
97        let content_update = if let CollectionOperationInfo::Content { sync_token } = message.data {
98            Some(ContentUpdate {
99                sync_token: Some(sync_token),
100            })
101        } else {
102            None
103        };
104
105        let push_message = PushMessage {
106            topic: message.topic,
107            content_update,
108        };
109
110        let payload = match push_message.serialize_to_string() {
111            Ok(payload) => payload,
112            Err(err) => {
113                error!("Could not serialize push message: {}", err);
114                return;
115            }
116        };
117
118        for subsciption in subscriptions {
119            if subsciption.is_expired(&Utc::now()) {
120                info!(
121                    "Deleting subscription {} on topic {} because it is expired",
122                    subsciption.id, subsciption.topic
123                );
124                self.try_delete_subscription(&subsciption.id).await;
125                continue;
126            }
127
128            if let Some(allowed_push_servers) = &self.allowed_push_servers {
129                if let Ok(url) = Url::parse(&subsciption.push_resource) {
130                    let origin = url.origin().unicode_serialization();
131                    if !allowed_push_servers.contains(&origin) {
132                        warn!(
133                            "Deleting subscription {} on topic {} because the endpoint is not in the list of allowed push servers",
134                            subsciption.id, subsciption.topic
135                        );
136                        self.try_delete_subscription(&subsciption.id).await;
137                        continue;
138                    }
139                } else {
140                    warn!(
141                        "Deleting subscription {} on topic {} because of invalid URL",
142                        subsciption.id, subsciption.topic
143                    );
144                    self.try_delete_subscription(&subsciption.id).await;
145                    continue;
146                }
147            }
148
149            if let Err(err) = send_payload(&payload, &subsciption).await {
150                error!("An error occured sending out a push notification: {err}");
151                if err.is_permament_error() {
152                    warn!(
153                        "Deleting subscription {} on topic {}",
154                        subsciption.id, subsciption.topic
155                    );
156                    self.try_delete_subscription(&subsciption.id).await;
157                }
158            }
159        }
160    }
161
162    async fn try_delete_subscription(&self, sub_id: &str) {
163        if let Err(err) = self.sub_store.delete_subscription(sub_id).await {
164            error!("Error deleting subsciption: {err}");
165        }
166    }
167}
168
169async fn send_payload(payload: &str, subsciption: &Subscription) -> Result<(), NotifierError> {
170    if subsciption.public_key_type != "p256dh" {
171        return Err(NotifierError::InvalidPublicKeyType(
172            subsciption.public_key_type.clone(),
173        ));
174    }
175    let endpoint = subsciption
176        .push_resource
177        .parse()
178        .map_err(|_| NotifierError::InvalidEndpointUrl(subsciption.push_resource.clone()))?;
179    let ua_public = base64::engine::general_purpose::URL_SAFE_NO_PAD
180        .decode(&subsciption.public_key)
181        .map_err(|_| NotifierError::InvalidKeyEncoding)?;
182    let auth_secret = base64::engine::general_purpose::URL_SAFE_NO_PAD
183        .decode(&subsciption.auth_secret)
184        .map_err(|_| NotifierError::InvalidKeyEncoding)?;
185
186    let client = reqwest::ClientBuilder::new()
187        .build()
188        .map_err(NotifierError::from)?;
189
190    let payload = ece::encrypt(&ua_public, &auth_secret, payload.as_bytes())?;
191
192    let mut request = reqwest::Request::new(Method::POST, endpoint);
193    *request.body_mut() = Some(Body::from(payload));
194    let hdrs = request.headers_mut();
195    hdrs.insert(
196        header::CONTENT_ENCODING,
197        HeaderValue::from_static("aes128gcm"),
198    );
199    hdrs.insert(
200        header::CONTENT_TYPE,
201        HeaderValue::from_static("application/octet-stream"),
202    );
203    hdrs.insert("TTL", HeaderValue::from(60));
204    client.execute(request).await?;
205
206    Ok(())
207}
208
209#[derive(Debug, thiserror::Error)]
210enum NotifierError {
211    #[error("Invalid public key type: {0}")]
212    InvalidPublicKeyType(String),
213    #[error("Invalid endpoint URL: {0}")]
214    InvalidEndpointUrl(String),
215    #[error("Invalid key encoding")]
216    InvalidKeyEncoding,
217    #[error(transparent)]
218    EceError(#[from] ece::Error),
219    #[error(transparent)]
220    ReqwestError(#[from] reqwest::Error),
221}
222
223impl NotifierError {
224    // Decide whether the error should cause the subscription to be removed
225    pub const fn is_permament_error(&self) -> bool {
226        match self {
227            Self::InvalidPublicKeyType(_)
228            | Self::InvalidEndpointUrl(_)
229            | Self::InvalidKeyEncoding => true,
230            Self::EceError(err) => matches!(
231                err,
232                ece::Error::InvalidAuthSecret | ece::Error::InvalidKeyLength
233            ),
234            Self::ReqwestError(_) => false,
235        }
236    }
237}
238
239#[cfg(test)]
240mod tests {
241    use crate::{Subscription, send_payload};
242    use base64::Engine;
243    use chrono::NaiveDateTime;
244    use ece::generate_keypair_and_auth_secret;
245
246    #[tokio::test]
247    async fn test_ntfy_request() {
248        let (keypair, auth_secret) = generate_keypair_and_auth_secret().unwrap();
249        let auth_secret = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(auth_secret);
250        let public_key =
251            base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(keypair.pub_as_raw().unwrap());
252
253        send_payload(
254            "hello",
255            &Subscription {
256                id: "asd".to_string(),
257                topic: "asd".to_string(),
258                expiration: NaiveDateTime::MAX,
259                push_resource: "https://ntfy.sh/upL00-v4L3SGM2".to_string(),
260                public_key,
261                public_key_type: "p256dh".to_string(),
262                auth_secret,
263            },
264        )
265        .await
266        .unwrap();
267    }
268}
rustical_dav_push/lib.rs

rustical_dav_push/
lib.rs
