continuwuity/src/service/pusher/mod.rs

mod webpush;

use std::{fmt::Debug, mem, net::IpAddr, sync::Arc};

use bytes::BytesMut;
use conduwuit::utils::response::LimitReadExt;
use conduwuit_core::{
	Err, Event, Result, Server, debug_warn, err, trace,
	utils::{stream::TryIgnore, string_from_bytes},
	warn,
};
use conduwuit_database::{Deserialized, Ignore, Interfix, Json, Map};
use futures::{Stream, StreamExt};
use ipaddress::IPAddress;
use ruma::{
	DeviceId, OwnedDeviceId, RoomId, UInt, UserId,
	api::{
		IncomingResponseExt, OutgoingRequest, OutgoingRequestExt,
		auth_scheme::NoAuthentication,
		path_builder::SinglePath,
		push_gateway::send_event_notification::{
			self,
			v1::{Device, Notification, NotificationCounts, NotificationPriority},
		},
	},
	events::{AnySyncTimelineEvent, TimelineEventType, room::power_levels::RoomPowerLevels},
	push::{
		Action, HighlightTweakValue, PushConditionPowerLevelsCtx, PushConditionRoomCtx,
		PushFormat, Ruleset, Tweak,
	},
	serde::{JsonObject, Raw},
	uint,
};
use ruminuwuity::pushers::{Pusher, PusherKind, set_pusher::v3::PusherAction};
use url::Host;

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<Self>,
	vapid: Vapid,
}

struct Services {
	server: Arc<Server>,
	globals: Dep<globals::Service>,
	config: Dep<config::Service>,
	client: Dep<client::Service>,
	state_accessor: Dep<rooms::state_accessor::Service>,
	state_cache: Dep<rooms::state_cache::Service>,
	users: Dep<users::Service>,
	sending: Dep<sending::Service>,
}

struct Data {
	senderkey_pusher: Arc<Map>,
	senderkey_webpushbackoff: Arc<Map>,
	senderkey_webpushstate: Arc<Map>,
	pushkey_deviceid: Arc<Map>,
}

impl crate::Service for Service {
	fn build(args: crate::Args<'_>) -> Result<Arc<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::Service>("globals"),
				client: args.depend::<client::Service>("client"),
				config: args.depend::<config::Service>("config"),
				state_accessor: args
					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
				users: args.depend::<users::Service>("users"),
				sending: args.depend::<sending::Service>("sending"),
			},
			me: me.clone(),
			vapid,
		}))
	}

	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
}

impl Service {
	pub async fn set_pusher(
		&self,
		sender: &UserId,
		sender_device: &DeviceId,
		action: &PusherAction,
	) -> Result<bool> {
		let pusher = match action {
			| PusherAction::Post(pusher) => pusher,
			| PusherAction::Delete(ids) => {
				self.delete_pusher(sender, ids.pushkey.as_str()).await;
				return Ok(false);
			},
		};

		let pushkey = pusher.ids.pushkey.as_str();
		if pushkey.len() > 512 {
			return Err!(Request(InvalidParam(
				"Push key length cannot be greater than 512 bytes."
			)));
		}

		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)
					.await?;
			},
			| PusherKind::WebPush(data) => {
				self.validate_push_url(&data.url, WEBPUSH_SCHEMES).await?;
				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<Pusher> = 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
			.sending
			.cleanup_events(None, Some(sender), Some(pushkey))
			.await
			.ok();
	}

	pub async fn get_pusher_device(&self, pushkey: &str) -> Result<OwnedDeviceId> {
		self.db.pushkey_deviceid.get(pushkey).await.deserialized()
	}

	pub async fn get_pusher(&self, sender: &UserId, pushkey: &str) -> Result<Pusher> {
		let senderkey = (sender, pushkey);
		self.db
			.senderkey_pusher
			.qry(&senderkey)
			.await
			.deserialized()
	}

	pub async fn get_pushers(&self, sender: &UserId) -> Vec<Pusher> {
		let prefix = (sender, Interfix);
		self.db
			.senderkey_pusher
			.stream_prefix(&prefix)
			.ignore_err()
			.map(|(_, pusher): (Ignore, Pusher)| pusher)
			.collect()
			.await
	}

	pub fn get_pushkeys<'a>(
		&'a self,
		sender: &'a UserId,
	) -> impl Stream<Item = String> + Send + 'a {
		let prefix = (sender, Interfix);
		self.db
			.senderkey_pusher
			.keys_prefix(&prefix)
			.ignore_err()
			.map(|(_, pushkey): (Ignore, &str)| pushkey)
			.map(ToOwned::to_owned)
	}

	/// Checks that a pusher URL is well-formed and not pointing at a forbidden
	/// address.
	async fn validate_push_url(&self, url: &str, allowed: &[&str]) -> Result<url::Url> {
		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"
			))));
		}

		// Proxies that resolve names remotely must enforce the denylist
		// themselves.
		let resolves_remotely =
			matches!(self.proxy_scheme(&parsed), Some("http" | "https" | "socks4a" | "socks5h"));

		let forbidden = match parsed.host() {
			| Some(Host::Ipv4(ip)) => !self.allowed_ip(ip.into()),
			| Some(Host::Ipv6(ip)) => !self.allowed_ip(ip.into()),
			| Some(Host::Domain(host)) if !resolves_remotely => self
				.services
				.client
				.dns_resolver
				.lookup_ip(host)
				.await
				.map_err(|e| {
					err!(Request(InvalidParam(warn!(
						%url,
						"Pusher URL host could not be resolved: {e}"
					))))
				})?
				.iter()
				.any(|ip| !self.allowed_ip(ip)),
			| _ => false,
		};
		if forbidden {
			return Err!(Request(InvalidParam(warn!(
				%url,
				"Pusher URL is a forbidden remote address"
			))));
		}

		Ok(parsed)
	}

	fn proxy_scheme(&self, url: &url::Url) -> Option<&str> {
		self.services
			.config
			.proxy
			.url_for(url)
			.map(url::Url::scheme)
	}

	fn allowed_ip(&self, ip: IpAddr) -> bool {
		IPAddress::parse(ip.to_string())
			.is_ok_and(|ip| self.services.client.valid_cidr_range(&ip))
	}

	#[tracing::instrument(skip(self, dest, request))]
	pub async fn send_request<T>(&self, dest: &str, request: T) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest<Authentication = NoAuthentication, PathBuilder = SinglePath>
			+ Debug
			+ Send,
	{
		let dest = dest
			.strip_suffix(self.services.globals.notification_push_path())
			.unwrap_or(dest);
		trace!("Push gateway destination: {dest}");

		let http_request = request
			.try_into_http_request::<BytesMut>(dest, (), ())
			.map_err(|e| {
				err!(BadServerResponse(warn!(
					"Failed to find destination {dest} for push gateway: {e}"
				)))
			})?
			.map(BytesMut::freeze);

		let response = self
			.execute_pusher_request(
				&self.services.client.pusher,
				reqwest::Request::try_from(http_request)?,
			)
			.await?;

		let status = response.status();
		if !status.is_success() {
			debug_warn!("Push gateway response body: {:?}", string_from_bytes(response.body()));
			return Err!(BadServerResponse(warn!(
				"Push gateway {dest} returned unsuccessful HTTP response: {status}"
			)));
		}

		let (parts, body) = response.into_parts();

		T::IncomingResponse::try_from_http_response(http::Response::from_parts(
			parts,
			body.as_ref(),
		))
		.map_err(|e| {
			err!(BadServerResponse(warn!("Push gateway {dest} returned invalid response: {e}")))
		})
	}

	/// Executes a request, checking the destination address on the way out and
	/// on the way in. The status is returned as-is; callers decide which codes
	/// are failures.
	async fn execute_pusher_request(
		&self,
		client: &reqwest::Client,
		request: reqwest::Request,
	) -> Result<http::Response<Vec<u8>>> {
		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) {
					return Err!(BadServerResponse(warn!(
						%url_host,
						"Not allowed to send requests to this IP"
					)));
				}
			}
		}

		let dest = request.url().clone();
		let proxied = self.proxy_scheme(&dest).is_some();

		let mut response = client.execute(request).await.map_err(|e| {
			err!(BadServerResponse(warn!(%dest, "Could not send request to pusher: {e}")))
		})?;

		if !proxied {
			trace!("Checking response destination's IP");
			if let Some(remote_addr) = response.remote_addr() {
				if !self.allowed_ip(remote_addr.ip()) {
					return Err!(BadServerResponse(warn!(
						%dest,
						"Pusher host resolved to a forbidden address after validation"
					)));
				}
			}
		}

		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))]
	pub async fn send_push_notice<E>(
		&self,
		user: &UserId,
		unread: UInt,
		pusher: &Pusher,
		ruleset: Ruleset,
		event: &E,
	) -> Result
	where
		E: Event + Send + Sync,
		for<'a> &'a E: Event + Send,
	{
		let mut notify = None;
		let mut tweaks = Vec::new();
		let Some(room_id) = event.room_id() else {
			// Only v12+ create events have no room ID
			return Ok(());
		};

		let power_levels = self
			.services
			.state_accessor
			.get_room_power_levels(room_id)
			.await;

		let serialized = event.to_format();
		for action in self
			.get_actions(user, &ruleset, power_levels.clone(), &serialized, room_id)
			.await
		{
			let n = match action {
				| Action::Notify => true,
				| Action::SetTweak(tweak) => {
					tweaks.push(tweak.clone());
					continue;
				},
				| _ => false,
			};

			if notify.is_some() {
				return Err!(Database(
					r#"Malformed pushrule contains more than one of these actions: ["dont_notify", "notify", "coalesce"]"#
				));
			}

			notify = Some(n);
		}

		if notify == Some(true) {
			self.send_notice(user, unread, pusher, tweaks, event)
				.await?;
		}
		// Else the event triggered no actions

		Ok(())
	}

	pub async fn push_joined_count(&self, room_id: &RoomId) -> UInt {
		self.services
			.state_cache
			.room_joined_count(room_id)
			.await
			.unwrap_or(1)
			.try_into()
			.unwrap_or_else(|_| uint!(0))
	}

	#[tracing::instrument(skip(self, user), level = "debug")]
	pub async fn push_condition_ctx(
		&self,
		user: &UserId,
		power_levels: RoomPowerLevels,
		room_id: &RoomId,
		room_joined_count: UInt,
	) -> PushConditionRoomCtx {
		let power_levels = PushConditionPowerLevelsCtx::from(power_levels);

		let user_display_name = self
			.services
			.users
			.displayname(user)
			.await
			.unwrap_or_else(|_| user.localpart().to_owned());

		PushConditionRoomCtx::new(
			room_id.to_owned(),
			room_joined_count,
			user.to_owned(),
			user_display_name,
		)
		.with_power_levels(power_levels)
	}

	#[tracing::instrument(skip(self, user, ruleset, pdu), level = "debug")]
	pub async fn get_actions<'a>(
		&self,
		user: &UserId,
		ruleset: &'a Ruleset,
		power_levels: RoomPowerLevels,
		pdu: &Raw<AnySyncTimelineEvent>,
		room_id: &RoomId,
	) -> &'a [Action] {
		let room_joined_count = self.push_joined_count(room_id).await;
		let ctx = self
			.push_condition_ctx(user, power_levels, room_id, room_joined_count)
			.await;

		ruleset.get_actions(pdu, &ctx).await
	}

	#[tracing::instrument(skip(self, unread, pusher, tweaks, event))]
	async fn send_notice<E>(
		&self,
		user: &UserId,
		unread: UInt,
		pusher: &Pusher,
		tweaks: Vec<Tweak>,
		event: &E,
	) -> Result
	where
		E: Event + Send + Sync,
	{
		// TODO: email
		match &pusher.kind {
			| PusherKind::Http(http) => {
				self.validate_push_url(&http.url, PUSH_GATEWAY_SCHEMES)
					.await?;

				let notify = self
					.build_notification(
						user,
						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(())
			},
			| PusherKind::WebPush(data) => {
				let notify = self
					.build_notification(
						user,
						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(()),
		}
	}

	#[allow(clippy::too_many_arguments)]
	async fn build_notification<E>(
		&self,
		user: &UserId,
		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(user.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
	}
}