malice

public

Star0

Watch0

128 commits 2 branches 0 tags 0 releases4.4 MiB

use std::{
    collections::HashMap,
    path::{Path, PathBuf},
    sync::{Mutex, MutexGuard, PoisonError},
    time::Duration,
};

use axum::{
    Router,
    body::Bytes,
    extract::{DefaultBodyLimit, State},
    http::{HeaderMap, StatusCode},
    response::{IntoResponse, Response},
    routing::{get, post},
};
use macltok::{HttpPusher, PushFormat, PushNotificationStatus, PusherKey};
use macltok_push_gateway::{GatewayDevice, GatewayNotification, notify_response, parse_notify};
use malice_core::push::{notice, unread_total};
use malice_push::{Message, Outcome, Sender, Subscription, Urgency};
use malice_static::push::{NOTIFY_PATH, json_response};
use rand_core::{OsRng, RngCore};
use serde::{Deserialize, Serialize};
use tokio::{sync::Semaphore, time::timeout};

use crate::{
    error::{ApiError, json_ok, no_content},
    routes::auth,
    state::{AppState, RestoreError},
};

pub const HOSTED_APP_ID: &str = "dev.malice.hosted";
const CONCURRENT_NOTIFIES: usize = 8;
const RESOLVE_TIMEOUT: Duration = Duration::from_secs(20);

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Entry {
    sid: String,
    subscription: Subscription,
}

/// Browser subscriptions by the random pushkey each one's pusher carries,
/// kept in `push.json` beside the sessions.
pub struct HostedPush {
    sender: Sender,
    public_url: Option<String>,
    path: PathBuf,
    entries: Mutex<HashMap<String, Entry>>,
    permits: Semaphore,
}

impl HostedPush {
    pub fn new(data_dir: &Path, public_url: Option<String>, sender: Sender) -> Self {
        let path = data_dir.join("push.json");
        let entries = match std::fs::read(&path) {
            Ok(bytes) => serde_json::from_slice(&bytes).unwrap_or_else(|error| {
                tracing::warn!(%error, "push subscriptions are unreadable, starting empty");
                HashMap::new()
            }),
            Err(_) => HashMap::new(),
        };
        Self {
            sender,
            public_url,
            path,
            entries: Mutex::new(entries),
            permits: Semaphore::new(CONCURRENT_NOTIFIES),
        }
    }

    /// The VAPID key lives in `data_dir`. Subscribing needs `MALICE_PUBLIC_URL`,
    /// since the homeserver must reach the notify endpoint.
    pub fn from_env(data_dir: &Path) -> Result<Self, String> {
        let public_url = malice_push::env::public_url();
        if public_url.is_none() {
            tracing::info!("web push is off until MALICE_PUBLIC_URL is set");
        }
        let sender = malice_push::env::sender(data_dir, public_url.as_deref())?;
        Ok(Self::new(data_dir, public_url, sender))
    }

    fn entries(&self) -> MutexGuard<'_, HashMap<String, Entry>> {
        self.entries.lock().unwrap_or_else(PoisonError::into_inner)
    }

    fn save(&self, entries: &HashMap<String, Entry>) {
        let result = serde_json::to_vec(entries)
            .map_err(std::io::Error::other)
            .and_then(|bytes| {
                let temp = self.path.with_extension("json.tmp");
                let mut options = std::fs::OpenOptions::new();
                options.write(true).create(true).truncate(true);
                #[cfg(unix)]
                {
                    use std::os::unix::fs::OpenOptionsExt;
                    options.mode(0o600);
                }
                let mut file = options.open(&temp)?;
                std::io::Write::write_all(&mut file, &bytes)?;
                file.sync_all()?;
                std::fs::rename(&temp, &self.path)
            });
        if let Err(error) = result {
            tracing::warn!(%error, "saving push subscriptions failed");
        }
    }

    /// Files `subscription` under `sid`, keeping the pushkey if this session
    /// already had it. Returns the pushkey, and the session and pushkey that
    /// held the endpoint before when another session did.
    fn claim(&self, sid: &str, subscription: Subscription) -> (String, Option<(String, String)>) {
        let mut entries = self.entries();
        let held = entries
            .iter()
            .find(|(_, entry)| entry.subscription.endpoint == subscription.endpoint)
            .map(|(pushkey, entry)| (pushkey.clone(), entry.sid.clone()));
        let (pushkey, displaced) = match held {
            Some((pushkey, owner)) if owner == sid => (pushkey, None),
            Some((pushkey, owner)) => {
                entries.remove(&pushkey);
                (new_pushkey(), Some((owner, pushkey)))
            }
            None => (new_pushkey(), None),
        };
        entries.insert(
            pushkey.clone(),
            Entry {
                sid: sid.to_owned(),
                subscription,
            },
        );
        self.save(&entries);
        (pushkey, displaced)
    }

    fn get(&self, pushkey: &str) -> Option<Entry> {
        self.entries().get(pushkey).cloned()
    }

    fn find(&self, sid: &str, endpoint: &str) -> Option<String> {
        self.entries()
            .iter()
            .find(|(_, entry)| entry.sid == sid && entry.subscription.endpoint == endpoint)
            .map(|(pushkey, _)| pushkey.clone())
    }

    fn forget(&self, pushkey: &str) {
        let mut entries = self.entries();
        if entries.remove(pushkey).is_some() {
            self.save(&entries);
        }
    }

    /// The pushers a logout of `sid` should take down with it.
    pub fn pusher_keys(&self, sid: &str) -> Vec<PusherKey> {
        self.entries()
            .iter()
            .filter(|(_, entry)| entry.sid == sid)
            .map(|(pushkey, _)| PusherKey {
                app_id: HOSTED_APP_ID.to_owned(),
                pushkey: pushkey.clone(),
            })
            .collect()
    }

    pub fn forget_session(&self, sid: &str) {
        let mut entries = self.entries();
        let before = entries.len();
        entries.retain(|_, entry| entry.sid != sid);
        if entries.len() != before {
            self.save(&entries);
        }
    }
}

fn new_pushkey() -> String {
    let mut bytes = [0u8; 16];
    OsRng.fill_bytes(&mut bytes);
    hex::encode(bytes)
}

#[derive(Debug, Deserialize)]
struct SubscribeKeys {
    p256dh: String,
    auth: String,
}

#[derive(Debug, Deserialize)]
struct SubscribeBody {
    endpoint: String,
    keys: SubscribeKeys,
}

#[derive(Debug, Deserialize)]
struct UnsubscribeBody {
    endpoint: String,
}

pub fn api_routes() -> Router<AppState> {
    Router::new()
        .route("/push/vapid", get(vapid))
        .route("/push/subscribe", post(subscribe))
        .route("/push/unsubscribe", post(unsubscribe))
}

pub fn notify_route() -> Router<AppState> {
    Router::new()
        .route(NOTIFY_PATH, post(notify))
        .layer(DefaultBodyLimit::max(64 * 1024))
}

fn push_off() -> ApiError {
    ApiError::new(StatusCode::SERVICE_UNAVAILABLE, "web push is not set up")
}

/// The push service, the public URL to point pushers at, or why push is off.
fn configured(state: &AppState) -> Result<(&HostedPush, &str), ApiError> {
    let push = state.push().ok_or_else(push_off)?;
    let url = push.public_url.as_deref().ok_or_else(push_off)?;
    Ok((push, url))
}

async fn vapid(State(state): State<AppState>) -> Response {
    match configured(&state) {
        Ok((push, _)) => json_ok(
            &serde_json::json!({ "public_key": push.sender.vapid_public_key() }),
            None,
        ),
        Err(error) => error.into_response(),
    }
}

async fn subscribe(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: axum::Json<SubscribeBody>,
) -> Response {
    let (push, public_url) = match configured(&state) {
        Ok(configured) => configured,
        Err(error) => return error.into_response(),
    };
    let (user, sid) = match auth(&state, &headers).await {
        Ok(pair) => pair,
        Err(error) => return error.into_response(),
    };
    let subscription = Subscription {
        endpoint: body.0.endpoint,
        p256dh: body.0.keys.p256dh,
        auth: body.0.keys.auth,
    };
    if !push.sender.allows(&subscription.endpoint) {
        return ApiError::bad_request("endpoint is not a known push service").into_response();
    }
    if !subscription.keys_valid() {
        return ApiError::bad_request("subscription keys are not valid").into_response();
    }
    let (pushkey, displaced) = push.claim(&sid, subscription);
    if let Some((owner, pushkey)) = displaced {
        let state = state.clone();
        tokio::spawn(async move {
            if let Ok(owner) = state.get_or_restore(&owner).await {
                let _ = owner
                    .handle
                    .remove_pusher(PusherKey {
                        app_id: HOSTED_APP_ID.to_owned(),
                        pushkey,
                    })
                    .await;
            }
        });
    }
    let pusher = HttpPusher {
        app_id: HOSTED_APP_ID.to_owned(),
        pushkey: pushkey.clone(),
        app_display_name: "malice".to_owned(),
        device_display_name: "Browser".to_owned(),
        lang: "en".to_owned(),
        url: format!("{public_url}{NOTIFY_PATH}"),
        format: PushFormat::EventIdOnly,
        data: serde_json::Map::new(),
    };
    match user.handle.set_pusher(pusher).await {
        Ok(()) => no_content(None),
        Err(error) => {
            push.forget(&pushkey);
            ApiError::from(malice_core::error::CoreError::from(error)).into_response()
        }
    }
}

async fn unsubscribe(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: axum::Json<UnsubscribeBody>,
) -> Response {
    let Some(push) = state.push() else {
        return push_off().into_response();
    };
    let (user, sid) = match auth(&state, &headers).await {
        Ok(pair) => pair,
        Err(error) => return error.into_response(),
    };
    let Some(pushkey) = push.find(&sid, &body.0.endpoint) else {
        return no_content(None);
    };
    let removed = user
        .handle
        .remove_pusher(PusherKey {
            app_id: HOSTED_APP_ID.to_owned(),
            pushkey: pushkey.clone(),
        })
        .await;
    match removed {
        Ok(()) => {
            push.forget(&pushkey);
            no_content(None)
        }
        Err(error) => ApiError::from(malice_core::error::CoreError::from(error)).into_response(),
    }
}

async fn notify(State(state): State<AppState>, body: Bytes) -> Response {
    let Some(push) = state.push() else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let notification = match parse_notify(&body) {
        Ok(notification) => notification,
        Err(error) => {
            return ApiError::bad_request(error.to_string()).into_response();
        }
    };
    if !notification.is_event() {
        return json_response(StatusCode::OK, notify_response(&[]));
    }
    let Ok(_permit) = push.permits.acquire().await else {
        return StatusCode::SERVICE_UNAVAILABLE.into_response();
    };
    let mut rejected = Vec::new();
    for device in &notification.devices {
        if deliver(&state, push, &notification, device).await {
            push.forget(&device.pushkey);
            rejected.push(device.pushkey.clone());
        }
    }
    json_response(StatusCode::OK, notify_response(&rejected))
}

/// Resolves and sends one device's push. True when its pushkey should be rejected.
async fn deliver(
    state: &AppState,
    push: &HostedPush,
    notification: &GatewayNotification,
    device: &GatewayDevice,
) -> bool {
    let (Some(room_id), Some(event_id)) = (&notification.room_id, &notification.event_id) else {
        return false;
    };
    if device.app_id != HOSTED_APP_ID {
        return true;
    }
    let Some(entry) = push.get(&device.pushkey) else {
        return true;
    };
    let user = match state.get_or_restore(&entry.sid).await {
        Ok(user) => user,
        Err(RestoreError::Unavailable(error)) => {
            tracing::warn!(%error, "push for a session that would not restore");
            return false;
        }
        Err(
            RestoreError::NotFound | RestoreError::Stale(_) | RestoreError::SoftLoggedOut { .. },
        ) => return true,
    };
    let status = match timeout(
        RESOLVE_TIMEOUT,
        user.handle.resolve_push_notification(room_id, event_id),
    )
    .await
    {
        Ok(Ok(status)) => status,
        Ok(Err(error)) => {
            tracing::warn!(%error, "resolving a push failed");
            return false;
        }
        Err(_) => {
            tracing::warn!("resolving a push timed out");
            return false;
        }
    };
    let PushNotificationStatus::Event(resolved) = status else {
        return false;
    };
    let unread = unread_total(&user.mirror.snapshot(), &resolved.room);
    let notice = notice(&resolved, unread);
    let urgency = if notice.mention || resolved.room.direct {
        Urgency::High
    } else {
        Urgency::Normal
    };
    let outcome = push
        .sender
        .send(
            &entry.subscription,
            &Message {
                payload: &notice.to_bytes(push.public_url.as_deref().unwrap_or_default()),
                urgency,
                topic: &notice.room_id,
            },
        )
        .await;
    if matches!(outcome, Outcome::Failed(_) | Outcome::Retryable) {
        tracing::warn!(?outcome, "web push delivery failed");
    }
    if !outcome.drops_subscription() {
        return false;
    }
    // continuwuity 26.9 ignores a notify response's rejected pushkeys
    let _ = user
        .handle
        .remove_pusher(PusherKey {
            app_id: HOSTED_APP_ID.to_owned(),
            pushkey: device.pushkey.clone(),
        })
        .await;
    true
}

#[cfg(test)]
mod tests {
    use axum::{body::Body, http::Request};
    use http_body_util::BodyExt;
    use malice_push::{Allowlist, Vapid};
    use serde_json::{Value, json};
    use tower::ServiceExt;

    use super::*;

    fn hosted(dir: &tempfile::TempDir) -> HostedPush {
        let sender = Sender::new(
            Vapid::load_or_create(&dir.path().join("vapid.pk8")).unwrap(),
            "mailto:test@example.org".to_owned(),
            Allowlist::default(),
        )
        .unwrap();
        HostedPush::new(
            dir.path(),
            Some("https://malice.example".to_owned()),
            sender,
        )
    }

    fn subscription(endpoint: &str) -> Subscription {
        Subscription {
            endpoint: endpoint.to_owned(),
            p256dh: "BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4".to_owned(),
            auth: "BTBZMqHH6r4Tts7J_aSIgg".to_owned(),
        }
    }

    #[test]
    fn registry_persists_and_moves_endpoints_between_sessions() {
        let dir = tempfile::tempdir().unwrap();
        let push = hosted(&dir);
        let (first, displaced) = push.claim("alpha", subscription("https://fcm.googleapis.com/a"));
        assert!(displaced.is_none());
        let (again, _) = push.claim("alpha", subscription("https://fcm.googleapis.com/a"));
        assert_eq!(first, again);
        let (moved, displaced) = push.claim("beta", subscription("https://fcm.googleapis.com/a"));
        assert_ne!(moved, first);
        assert_eq!(displaced, Some(("alpha".to_owned(), first.clone())));
        push.claim("beta", subscription("https://fcm.googleapis.com/b"));

        let reloaded = hosted(&dir);
        assert_eq!(reloaded.pusher_keys("beta").len(), 2);
        assert!(reloaded.pusher_keys("alpha").is_empty());
        assert_eq!(
            reloaded.find("beta", "https://fcm.googleapis.com/a"),
            Some(moved)
        );
        reloaded.forget_session("beta");
        assert!(hosted(&dir).pusher_keys("beta").is_empty());
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mode = std::fs::metadata(dir.path().join("push.json"))
                .unwrap()
                .permissions()
                .mode();
            assert_eq!(mode & 0o777, 0o600);
        }
    }

    async fn notify(state: &AppState, notification: Value) -> (StatusCode, Value) {
        let response = crate::routes::router(state.clone(), None)
            .oneshot(
                Request::post(NOTIFY_PATH)
                    .body(Body::from(
                        json!({ "notification": notification }).to_string(),
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        let status = response.status();
        let body = response.into_body().collect().await.unwrap().to_bytes();
        (status, serde_json::from_slice(&body).unwrap_or(Value::Null))
    }

    #[tokio::test]
    async fn notify_drops_count_only_pushes_and_rejects_unknown_pushers() {
        let dir = tempfile::tempdir().unwrap();
        let push = hosted(&dir);
        let (orphan, _) = push.claim(
            "nosuchsession",
            subscription("https://fcm.googleapis.com/o"),
        );
        let state = AppState::new(dir.path().to_owned(), [7; 32], false, false).with_push(push);
        let device = |app_id: &str, pushkey: &str| json!({"app_id": app_id, "pushkey": pushkey});

        let (status, body) = notify(
            &state,
            json!({"counts": {"unread": 1}, "devices": [device(HOSTED_APP_ID, &orphan)]}),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, json!({"rejected": []}));
        assert_eq!(state.push().unwrap().pusher_keys("nosuchsession").len(), 1);

        let (status, body) = notify(
            &state,
            json!({
                "event_id": "$e:x",
                "room_id": "!r:x",
                "devices": [
                    device(HOSTED_APP_ID, "unknown"),
                    device("dev.malice.device", "foreign"),
                    device(HOSTED_APP_ID, &orphan),
                ],
            }),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, json!({"rejected": ["unknown", "foreign", orphan]}));
        assert!(
            state
                .push()
                .unwrap()
                .pusher_keys("nosuchsession")
                .is_empty()
        );

        let (status, _) = notify(&state, json!("not a notification")).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn vapid_needs_a_public_url() {
        let dir = tempfile::tempdir().unwrap();
        let mut push = hosted(&dir);
        let key = push.sender.vapid_public_key().to_owned();
        let state =
            AppState::new(dir.path().to_owned(), [7; 32], false, false).with_push(hosted(&dir));
        let get = |state: &AppState| {
            crate::routes::router(state.clone(), None)
                .oneshot(Request::get("/api/push/vapid").body(Body::empty()).unwrap())
        };
        let response = get(&state).await.unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = response.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(
            serde_json::from_slice::<Value>(&body).unwrap()["public_key"],
            key
        );
        push.public_url = None;
        let state = AppState::new(dir.path().to_owned(), [7; 32], false, false).with_push(push);
        assert_eq!(
            get(&state).await.unwrap().status(),
            StatusCode::SERVICE_UNAVAILABLE
        );
    }
}

1	use std::{
2	collections::HashMap,
3	path::{Path, PathBuf},
4	sync::{Mutex, MutexGuard, PoisonError},
5	time::Duration,
6	};
7
8	use axum::{
9	Router,
10	body::Bytes,
11	extract::{DefaultBodyLimit, State},
12	http::{HeaderMap, StatusCode},
13	response::{IntoResponse, Response},
14	routing::{get, post},
15	};
16	use macltok::{HttpPusher, PushFormat, PushNotificationStatus, PusherKey};
17	use macltok_push_gateway::{GatewayDevice, GatewayNotification, notify_response, parse_notify};
18	use malice_core::push::{notice, unread_total};
19	use malice_push::{Message, Outcome, Sender, Subscription, Urgency};
20	use malice_static::push::{NOTIFY_PATH, json_response};
21	use rand_core::{OsRng, RngCore};
22	use serde::{Deserialize, Serialize};
23	use tokio::{sync::Semaphore, time::timeout};
24
25	use crate::{
26	error::{ApiError, json_ok, no_content},
27	routes::auth,
28	state::{AppState, RestoreError},
29	};
30
31	pub const HOSTED_APP_ID: &str = "dev.malice.hosted";
32	const CONCURRENT_NOTIFIES: usize = 8;
33	const RESOLVE_TIMEOUT: Duration = Duration::from_secs(20);
34
35	#[derive(Debug, Clone, Serialize, Deserialize)]
36	struct Entry {
37	sid: String,
38	subscription: Subscription,
39	}
40
41	/// Browser subscriptions by the random pushkey each one's pusher carries,
42	/// kept in `push.json` beside the sessions.
43	pub struct HostedPush {
44	sender: Sender,
45	public_url: Option<String>,
46	path: PathBuf,
47	entries: Mutex<HashMap<String, Entry>>,
48	permits: Semaphore,
49	}
50
51	impl HostedPush {
52	pub fn new(data_dir: &Path, public_url: Option<String>, sender: Sender) -> Self {
53	let path = data_dir.join("push.json");
54	let entries = match std::fs::read(&path) {
55	Ok(bytes) => serde_json::from_slice(&bytes).unwrap_or_else(\|error\| {
56	tracing::warn!(%error, "push subscriptions are unreadable, starting empty");
57	HashMap::new()
58	}),
59	Err(_) => HashMap::new(),
60	};
61	Self {
62	sender,
63	public_url,
64	path,
65	entries: Mutex::new(entries),
66	permits: Semaphore::new(CONCURRENT_NOTIFIES),
67	}
68	}
69
70	/// The VAPID key lives in `data_dir`. Subscribing needs `MALICE_PUBLIC_URL`,
71	/// since the homeserver must reach the notify endpoint.
72	pub fn from_env(data_dir: &Path) -> Result<Self, String> {
73	let public_url = malice_push::env::public_url();
74	if public_url.is_none() {
75	tracing::info!("web push is off until MALICE_PUBLIC_URL is set");
76	}
77	let sender = malice_push::env::sender(data_dir, public_url.as_deref())?;
78	Ok(Self::new(data_dir, public_url, sender))
79	}
80
81	fn entries(&self) -> MutexGuard<'_, HashMap<String, Entry>> {
82	self.entries.lock().unwrap_or_else(PoisonError::into_inner)
83	}
84
85	fn save(&self, entries: &HashMap<String, Entry>) {
86	let result = serde_json::to_vec(entries)
87	.map_err(std::io::Error::other)
88	.and_then(\|bytes\| {
89	let temp = self.path.with_extension("json.tmp");
90	let mut options = std::fs::OpenOptions::new();
91	options.write(true).create(true).truncate(true);
92	#[cfg(unix)]
93	{
94	use std::os::unix::fs::OpenOptionsExt;
95	options.mode(0o600);
96	}
97	let mut file = options.open(&temp)?;
98	std::io::Write::write_all(&mut file, &bytes)?;
99	file.sync_all()?;
100	std::fs::rename(&temp, &self.path)
101	});
102	if let Err(error) = result {
103	tracing::warn!(%error, "saving push subscriptions failed");
104	}
105	}
106
107	/// Files `subscription` under `sid`, keeping the pushkey if this session
108	/// already had it. Returns the pushkey, and the session and pushkey that
109	/// held the endpoint before when another session did.
110	fn claim(&self, sid: &str, subscription: Subscription) -> (String, Option<(String, String)>) {
111	let mut entries = self.entries();
112	let held = entries
113	.iter()
114	.find(\|(_, entry)\| entry.subscription.endpoint == subscription.endpoint)
115	.map(\|(pushkey, entry)\| (pushkey.clone(), entry.sid.clone()));
116	let (pushkey, displaced) = match held {
117	Some((pushkey, owner)) if owner == sid => (pushkey, None),
118	Some((pushkey, owner)) => {
119	entries.remove(&pushkey);
120	(new_pushkey(), Some((owner, pushkey)))
121	}
122	None => (new_pushkey(), None),
123	};
124	entries.insert(
125	pushkey.clone(),
126	Entry {
127	sid: sid.to_owned(),
128	subscription,
129	},
130	);
131	self.save(&entries);
132	(pushkey, displaced)
133	}
134
135	fn get(&self, pushkey: &str) -> Option<Entry> {
136	self.entries().get(pushkey).cloned()
137	}
138
139	fn find(&self, sid: &str, endpoint: &str) -> Option<String> {
140	self.entries()
141	.iter()
142	.find(\|(_, entry)\| entry.sid == sid && entry.subscription.endpoint == endpoint)
143	.map(\|(pushkey, _)\| pushkey.clone())
144	}
145
146	fn forget(&self, pushkey: &str) {
147	let mut entries = self.entries();
148	if entries.remove(pushkey).is_some() {
149	self.save(&entries);
150	}
151	}
152
153	/// The pushers a logout of `sid` should take down with it.
154	pub fn pusher_keys(&self, sid: &str) -> Vec<PusherKey> {
155	self.entries()
156	.iter()
157	.filter(\|(_, entry)\| entry.sid == sid)
158	.map(\|(pushkey, _)\| PusherKey {
159	app_id: HOSTED_APP_ID.to_owned(),
160	pushkey: pushkey.clone(),
161	})
162	.collect()
163	}
164
165	pub fn forget_session(&self, sid: &str) {
166	let mut entries = self.entries();
167	let before = entries.len();
168	entries.retain(\|_, entry\| entry.sid != sid);
169	if entries.len() != before {
170	self.save(&entries);
171	}
172	}
173	}
174
175	fn new_pushkey() -> String {
176	let mut bytes = [0u8; 16];
177	OsRng.fill_bytes(&mut bytes);
178	hex::encode(bytes)
179	}
180
181	#[derive(Debug, Deserialize)]
182	struct SubscribeKeys {
183	p256dh: String,
184	auth: String,
185	}
186
187	#[derive(Debug, Deserialize)]
188	struct SubscribeBody {
189	endpoint: String,
190	keys: SubscribeKeys,
191	}
192
193	#[derive(Debug, Deserialize)]
194	struct UnsubscribeBody {
195	endpoint: String,
196	}
197
198	pub fn api_routes() -> Router<AppState> {
199	Router::new()
200	.route("/push/vapid", get(vapid))
201	.route("/push/subscribe", post(subscribe))
202	.route("/push/unsubscribe", post(unsubscribe))
203	}
204
205	pub fn notify_route() -> Router<AppState> {
206	Router::new()
207	.route(NOTIFY_PATH, post(notify))
208	.layer(DefaultBodyLimit::max(64 * 1024))
209	}
210
211	fn push_off() -> ApiError {
212	ApiError::new(StatusCode::SERVICE_UNAVAILABLE, "web push is not set up")
213	}
214
215	/// The push service, the public URL to point pushers at, or why push is off.
216	fn configured(state: &AppState) -> Result<(&HostedPush, &str), ApiError> {
217	let push = state.push().ok_or_else(push_off)?;
218	let url = push.public_url.as_deref().ok_or_else(push_off)?;
219	Ok((push, url))
220	}
221
222	async fn vapid(State(state): State<AppState>) -> Response {
223	match configured(&state) {
224	Ok((push, _)) => json_ok(
225	&serde_json::json!({ "public_key": push.sender.vapid_public_key() }),
226	None,
227	),
228	Err(error) => error.into_response(),
229	}
230	}
231
232	async fn subscribe(
233	State(state): State<AppState>,
234	headers: HeaderMap,
235	body: axum::Json<SubscribeBody>,
236	) -> Response {
237	let (push, public_url) = match configured(&state) {
238	Ok(configured) => configured,
239	Err(error) => return error.into_response(),
240	};
241	let (user, sid) = match auth(&state, &headers).await {
242	Ok(pair) => pair,
243	Err(error) => return error.into_response(),
244	};
245	let subscription = Subscription {
246	endpoint: body.0.endpoint,
247	p256dh: body.0.keys.p256dh,
248	auth: body.0.keys.auth,
249	};
250	if !push.sender.allows(&subscription.endpoint) {
251	return ApiError::bad_request("endpoint is not a known push service").into_response();
252	}
253	if !subscription.keys_valid() {
254	return ApiError::bad_request("subscription keys are not valid").into_response();
255	}
256	let (pushkey, displaced) = push.claim(&sid, subscription);
257	if let Some((owner, pushkey)) = displaced {
258	let state = state.clone();
259	tokio::spawn(async move {
260	if let Ok(owner) = state.get_or_restore(&owner).await {
261	let _ = owner
262	.handle
263	.remove_pusher(PusherKey {
264	app_id: HOSTED_APP_ID.to_owned(),
265	pushkey,
266	})
267	.await;
268	}
269	});
270	}
271	let pusher = HttpPusher {
272	app_id: HOSTED_APP_ID.to_owned(),
273	pushkey: pushkey.clone(),
274	app_display_name: "malice".to_owned(),
275	device_display_name: "Browser".to_owned(),
276	lang: "en".to_owned(),
277	url: format!("{public_url}{NOTIFY_PATH}"),
278	format: PushFormat::EventIdOnly,
279	data: serde_json::Map::new(),
280	};
281	match user.handle.set_pusher(pusher).await {
282	Ok(()) => no_content(None),
283	Err(error) => {
284	push.forget(&pushkey);
285	ApiError::from(malice_core::error::CoreError::from(error)).into_response()
286	}
287	}
288	}
289
290	async fn unsubscribe(
291	State(state): State<AppState>,
292	headers: HeaderMap,
293	body: axum::Json<UnsubscribeBody>,
294	) -> Response {
295	let Some(push) = state.push() else {
296	return push_off().into_response();
297	};
298	let (user, sid) = match auth(&state, &headers).await {
299	Ok(pair) => pair,
300	Err(error) => return error.into_response(),
301	};
302	let Some(pushkey) = push.find(&sid, &body.0.endpoint) else {
303	return no_content(None);
304	};
305	let removed = user
306	.handle
307	.remove_pusher(PusherKey {
308	app_id: HOSTED_APP_ID.to_owned(),
309	pushkey: pushkey.clone(),
310	})
311	.await;
312	match removed {
313	Ok(()) => {
314	push.forget(&pushkey);
315	no_content(None)
316	}
317	Err(error) => ApiError::from(malice_core::error::CoreError::from(error)).into_response(),
318	}
319	}
320
321	async fn notify(State(state): State<AppState>, body: Bytes) -> Response {
322	let Some(push) = state.push() else {
323	return StatusCode::NOT_FOUND.into_response();
324	};
325	let notification = match parse_notify(&body) {
326	Ok(notification) => notification,
327	Err(error) => {
328	return ApiError::bad_request(error.to_string()).into_response();
329	}
330	};
331	if !notification.is_event() {
332	return json_response(StatusCode::OK, notify_response(&[]));
333	}
334	let Ok(_permit) = push.permits.acquire().await else {
335	return StatusCode::SERVICE_UNAVAILABLE.into_response();
336	};
337	let mut rejected = Vec::new();
338	for device in &notification.devices {
339	if deliver(&state, push, &notification, device).await {
340	push.forget(&device.pushkey);
341	rejected.push(device.pushkey.clone());
342	}
343	}
344	json_response(StatusCode::OK, notify_response(&rejected))
345	}
346
347	/// Resolves and sends one device's push. True when its pushkey should be rejected.
348	async fn deliver(
349	state: &AppState,
350	push: &HostedPush,
351	notification: &GatewayNotification,
352	device: &GatewayDevice,
353	) -> bool {
354	let (Some(room_id), Some(event_id)) = (&notification.room_id, &notification.event_id) else {
355	return false;
356	};
357	if device.app_id != HOSTED_APP_ID {
358	return true;
359	}
360	let Some(entry) = push.get(&device.pushkey) else {
361	return true;
362	};
363	let user = match state.get_or_restore(&entry.sid).await {
364	Ok(user) => user,
365	Err(RestoreError::Unavailable(error)) => {
366	tracing::warn!(%error, "push for a session that would not restore");
367	return false;
368	}
369	Err(
370	RestoreError::NotFound \| RestoreError::Stale(_) \| RestoreError::SoftLoggedOut { .. },
371	) => return true,
372	};
373	let status = match timeout(
374	RESOLVE_TIMEOUT,
375	user.handle.resolve_push_notification(room_id, event_id),
376	)
377	.await
378	{
379	Ok(Ok(status)) => status,
380	Ok(Err(error)) => {
381	tracing::warn!(%error, "resolving a push failed");
382	return false;
383	}
384	Err(_) => {
385	tracing::warn!("resolving a push timed out");
386	return false;
387	}
388	};
389	let PushNotificationStatus::Event(resolved) = status else {
390	return false;
391	};
392	let unread = unread_total(&user.mirror.snapshot(), &resolved.room);
393	let notice = notice(&resolved, unread);
394	let urgency = if notice.mention \|\| resolved.room.direct {
395	Urgency::High
396	} else {
397	Urgency::Normal
398	};
399	let outcome = push
400	.sender
401	.send(
402	&entry.subscription,
403	&Message {
404	payload: &notice.to_bytes(push.public_url.as_deref().unwrap_or_default()),
405	urgency,
406	topic: &notice.room_id,
407	},
408	)
409	.await;
410	if matches!(outcome, Outcome::Failed(_) \| Outcome::Retryable) {
411	tracing::warn!(?outcome, "web push delivery failed");
412	}
413	if !outcome.drops_subscription() {
414	return false;
415	}
416	// continuwuity 26.9 ignores a notify response's rejected pushkeys
417	let _ = user
418	.handle
419	.remove_pusher(PusherKey {
420	app_id: HOSTED_APP_ID.to_owned(),
421	pushkey: device.pushkey.clone(),
422	})
423	.await;
424	true
425	}
426
427	#[cfg(test)]
428	mod tests {
429	use axum::{body::Body, http::Request};
430	use http_body_util::BodyExt;
431	use malice_push::{Allowlist, Vapid};
432	use serde_json::{Value, json};
433	use tower::ServiceExt;
434
435	use super::*;
436
437	fn hosted(dir: &tempfile::TempDir) -> HostedPush {
438	let sender = Sender::new(
439	Vapid::load_or_create(&dir.path().join("vapid.pk8")).unwrap(),
440	"mailto:test@example.org".to_owned(),
441	Allowlist::default(),
442	)
443	.unwrap();
444	HostedPush::new(
445	dir.path(),
446	Some("https://malice.example".to_owned()),
447	sender,
448	)
449	}
450
451	fn subscription(endpoint: &str) -> Subscription {
452	Subscription {
453	endpoint: endpoint.to_owned(),
454	p256dh: "BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4".to_owned(),
455	auth: "BTBZMqHH6r4Tts7J_aSIgg".to_owned(),
456	}
457	}
458
459	#[test]
460	fn registry_persists_and_moves_endpoints_between_sessions() {
461	let dir = tempfile::tempdir().unwrap();
462	let push = hosted(&dir);
463	let (first, displaced) = push.claim("alpha", subscription("https://fcm.googleapis.com/a"));
464	assert!(displaced.is_none());
465	let (again, _) = push.claim("alpha", subscription("https://fcm.googleapis.com/a"));
466	assert_eq!(first, again);
467	let (moved, displaced) = push.claim("beta", subscription("https://fcm.googleapis.com/a"));
468	assert_ne!(moved, first);
469	assert_eq!(displaced, Some(("alpha".to_owned(), first.clone())));
470	push.claim("beta", subscription("https://fcm.googleapis.com/b"));
471
472	let reloaded = hosted(&dir);
473	assert_eq!(reloaded.pusher_keys("beta").len(), 2);
474	assert!(reloaded.pusher_keys("alpha").is_empty());
475	assert_eq!(
476	reloaded.find("beta", "https://fcm.googleapis.com/a"),
477	Some(moved)
478	);
479	reloaded.forget_session("beta");
480	assert!(hosted(&dir).pusher_keys("beta").is_empty());
481	#[cfg(unix)]
482	{
483	use std::os::unix::fs::PermissionsExt;
484	let mode = std::fs::metadata(dir.path().join("push.json"))
485	.unwrap()
486	.permissions()
487	.mode();
488	assert_eq!(mode & 0o777, 0o600);
489	}
490	}
491
492	async fn notify(state: &AppState, notification: Value) -> (StatusCode, Value) {
493	let response = crate::routes::router(state.clone(), None)
494	.oneshot(
495	Request::post(NOTIFY_PATH)
496	.body(Body::from(
497	json!({ "notification": notification }).to_string(),
498	))
499	.unwrap(),
500	)
501	.await
502	.unwrap();
503	let status = response.status();
504	let body = response.into_body().collect().await.unwrap().to_bytes();
505	(status, serde_json::from_slice(&body).unwrap_or(Value::Null))
506	}
507
508	#[tokio::test]
509	async fn notify_drops_count_only_pushes_and_rejects_unknown_pushers() {
510	let dir = tempfile::tempdir().unwrap();
511	let push = hosted(&dir);
512	let (orphan, _) = push.claim(
513	"nosuchsession",
514	subscription("https://fcm.googleapis.com/o"),
515	);
516	let state = AppState::new(dir.path().to_owned(), [7; 32], false, false).with_push(push);
517	let device = \|app_id: &str, pushkey: &str\| json!({"app_id": app_id, "pushkey": pushkey});
518
519	let (status, body) = notify(
520	&state,
521	json!({"counts": {"unread": 1}, "devices": [device(HOSTED_APP_ID, &orphan)]}),
522	)
523	.await;
524	assert_eq!(status, StatusCode::OK);
525	assert_eq!(body, json!({"rejected": []}));
526	assert_eq!(state.push().unwrap().pusher_keys("nosuchsession").len(), 1);
527
528	let (status, body) = notify(
529	&state,
530	json!({
531	"event_id": "$e:x",
532	"room_id": "!r:x",
533	"devices": [
534	device(HOSTED_APP_ID, "unknown"),
535	device("dev.malice.device", "foreign"),
536	device(HOSTED_APP_ID, &orphan),
537	],
538	}),
539	)
540	.await;
541	assert_eq!(status, StatusCode::OK);
542	assert_eq!(body, json!({"rejected": ["unknown", "foreign", orphan]}));
543	assert!(
544	state
545	.push()
546	.unwrap()
547	.pusher_keys("nosuchsession")
548	.is_empty()
549	);
550
551	let (status, _) = notify(&state, json!("not a notification")).await;
552	assert_eq!(status, StatusCode::BAD_REQUEST);
553	}
554
555	#[tokio::test]
556	async fn vapid_needs_a_public_url() {
557	let dir = tempfile::tempdir().unwrap();
558	let mut push = hosted(&dir);
559	let key = push.sender.vapid_public_key().to_owned();
560	let state =
561	AppState::new(dir.path().to_owned(), [7; 32], false, false).with_push(hosted(&dir));
562	let get = \|state: &AppState\| {
563	crate::routes::router(state.clone(), None)
564	.oneshot(Request::get("/api/push/vapid").body(Body::empty()).unwrap())
565	};
566	let response = get(&state).await.unwrap();
567	assert_eq!(response.status(), StatusCode::OK);
568	let body = response.into_body().collect().await.unwrap().to_bytes();
569	assert_eq!(
570	serde_json::from_slice::<Value>(&body).unwrap()["public_key"],
571	key
572	);
573	push.public_url = None;
574	let state = AppState::new(dir.path().to_owned(), [7; 32], false, false).with_push(push);
575	assert_eq!(
576	get(&state).await.unwrap().status(),
577	StatusCode::SERVICE_UNAVAILABLE
578	);
579	}
580	}