tuwunel_api_client/

send.rs

use axum::extract::State;
use futures::future::{try_join, try_join3};
use ruma::{
	OwnedEventId, RoomId, UserId,
	api::client::message::{
		send_message_event, send_message_event::v3::Response as SendMessageResponse,
	},
	events::{
		AnyMessageLikeEventContent, MessageLikeEventType,
		reaction::ReactionEventContent,
		room::{encrypted::Relation, redaction::RoomRedactionEventContent},
	},
	serde::Raw,
};
use serde::Deserialize;
use serde_json::from_str;
use tuwunel_core::{Err, Result, debug_warn, err, result::NotFound, warn};
use tuwunel_matrix::{Event, PduEvent, pdu::PduBuilder};
use tuwunel_service::{
	Services,
	transaction_ids::{RoomScope, RoomTxn},
};

use crate::{
	Ruma,
	utils::{is_self_redaction, txnid_event_id, txnid_unsigned},
};

#[derive(Deserialize)]
struct ExtractRelatesTo {
	#[serde(rename = "m.relates_to")]
	relates_to: Relation,
}

/// # `PUT /_matrix/client/v3/rooms/{roomId}/send/{eventType}/{txnId}`
///
/// Send a message event into the room.
///
/// - Is a NOOP if the txn id was already used before and returns the same event
///   id again
/// - The only requirement for the content is that it has to be valid json
/// - Tries to send the event into the room, auth rules will determine if it is
///   allowed
pub(crate) async fn send_message_event_route(
	State(services): State<crate::State>,
	body: Ruma<send_message_event::v3::Request>,
) -> Result<send_message_event::v3::Response> {
	let sender_user = body.sender_user();
	let sender_device = body.sender_device.as_deref();
	let appservice_info = body.appservice_info.as_ref();
	let redacts_id =
		check_send_policy(&services, sender_user, &body.event_type, &body.body.body).await?;

	let state_lock = services.state.mutex.lock(&body.room_id).await;
	let event_type = body.event_type.to_cow_str();
	let txn = RoomTxn {
		user_id: sender_user,
		device_id: sender_device,
		txn_id: &body.txn_id,
		room_id: &body.room_id,
	};

	let content_checks = check_send_content(
		&services,
		sender_user,
		&body.room_id,
		&body.event_type,
		&body.body.body,
	);

	let (existing_txnid, ()) =
		try_join(check_existing_txnid(&services, txn, &event_type), content_checks).await?;

	if let Some(existing_txnid) = existing_txnid {
		return Ok(existing_txnid);
	}

	let content = from_str(body.body.body.json().get())
		.map_err(|e| err!(Request(BadJson("Invalid JSON body: {e}"))))?;

	let pdu = PduBuilder {
		event_type: body.event_type.clone().into(),
		content,
		unsigned: txnid_unsigned(&body.txn_id).into(),
		timestamp: appservice_info.and(body.timestamp),
		redacts: redacts_id,
		..Default::default()
	};

	let event_id = services
		.timeline
		.build_and_append_pdu(pdu, sender_user, &body.room_id, &state_lock)
		.await?;

	services.transaction_ids.add_room_txnid(
		txn,
		RoomScope::Send(&event_type),
		event_id.as_bytes(),
	);

	drop(state_lock);

	Ok(SendMessageResponse { event_id })
}

/// Applies the sender's policy for sending a message event, before the room
/// lock.
///
/// Refuses encrypted events when encryption is disabled, local redactions
/// when the server disables them, and a suspended sender's events other than
/// self-redactions. Returns the target of a redaction for the top-level
/// `redacts` field.
#[tracing::instrument(level = "debug", skip_all)]
pub(crate) async fn check_send_policy(
	services: &Services,
	sender_user: &UserId,
	event_type: &MessageLikeEventType,
	content: &Raw<AnyMessageLikeEventContent>,
) -> Result<Option<OwnedEventId>> {
	// Forbid m.room.encrypted if encryption is disabled
	if *event_type == MessageLikeEventType::RoomEncrypted && !services.config.encryption.allow {
		return Err!(Request(Forbidden("Encryption has been disabled")));
	}

	// MSC4169: clients sending m.room.redaction via /send put `redacts` in
	// `content`. Pre-v11 auth rules read it from the top level; lift it so
	// `redacts_id(...)` resolves regardless of room version. Mirrors the
	// /redact handler.
	let redaction_content = || {
		content
			.deserialize_as_unchecked::<RoomRedactionEventContent>()
			.inspect_err(|_| {
				debug_warn!(
					%sender_user,
					event = %content.json(),
					"Client sent invalid redaction event"
				);
			})
			.ok()
	};

	let redacts_id = event_type
		.eq(&MessageLikeEventType::RoomRedaction)
		.then(redaction_content)
		.flatten()
		.and_then(|content| content.redacts);

	if *event_type == MessageLikeEventType::RoomRedaction
		&& services.config.rooms.redaction.disable_local
		&& !services.admin.user_is_admin(sender_user).await
	{
		warn!(
			%sender_user,
			?redacts_id,
			"Local redactions are disabled, non-admin user attempted to redact an event"
		);

		return Err!(Request(Forbidden("Redactions are disabled on this server.")));
	}

	if services.users.is_suspended(sender_user).await {
		if *event_type != MessageLikeEventType::RoomRedaction {
			return Err!(Request(UserSuspended(
				"Cannot send non-redaction events while suspended."
			)));
		}

		let is_self = match &redacts_id {
			| None => false,
			| Some(redacts_id) => is_self_redaction(services, sender_user, redacts_id).await,
		};

		if !is_self {
			return Err!(Request(UserSuspended("Can only redact own events while suspended.")));
		}
	}

	Ok(redacts_id)
}

/// Applies the room's content rules for sending a message event, under the
/// room lock.
///
/// Refuses a duplicate reaction, a call invite in a public room, and a thread
/// rooted at an event that itself carries a relation.
#[tracing::instrument(level = "debug", skip_all)]
pub(crate) async fn check_send_content(
	services: &Services,
	sender_user: &UserId,
	room_id: &RoomId,
	event_type: &MessageLikeEventType,
	content: &Raw<AnyMessageLikeEventContent>,
) -> Result {
	try_join3(
		check_duplicate_reaction(services, event_type, sender_user, content),
		check_public_call_invite(services, event_type, room_id),
		check_nested_thread(services, content),
	)
	.await
	.map(|((), (), ())| ())
}

async fn check_public_call_invite(
	services: &Services,
	event_type: &MessageLikeEventType,
	room_id: &RoomId,
) -> Result {
	if *event_type != MessageLikeEventType::CallInvite {
		return Ok(());
	}

	if !services.directory.is_public_room(room_id).await {
		return Ok(());
	}

	Err!(Request(Forbidden("Room call invites are not allowed in public rooms")))
}

// Forbid duplicate reactions
async fn check_duplicate_reaction(
	services: &Services,
	event_type: &MessageLikeEventType,
	sender_user: &UserId,
	body: &Raw<AnyMessageLikeEventContent>,
) -> Result {
	if *event_type != MessageLikeEventType::Reaction {
		return Ok(());
	}

	let Ok(content) = body.deserialize_as_unchecked::<ReactionEventContent>() else {
		return Ok(());
	};

	if !services
		.pdu_metadata
		.event_has_relation(
			&content.relates_to.event_id,
			Some(sender_user),
			None,
			Some(&content.relates_to.key),
		)
		.await
	{
		return Ok(());
	}

	Err!(Request(DuplicateAnnotation("Duplicate reactions are not allowed.")))
}

// MSC3440/Matrix 1.4: a thread may only target an event which itself carries
// no rel_type; the spec assigns this rejection 400 M_UNKNOWN.
async fn check_nested_thread(
	services: &Services,
	body: &Raw<AnyMessageLikeEventContent>,
) -> Result {
	let Ok(ExtractRelatesTo { relates_to: Relation::Thread(thread) }) =
		body.deserialize_as_unchecked()
	else {
		return Ok(());
	};

	let Ok(root) = services.timeline.get_pdu(&thread.event_id).await else {
		return Ok(());
	};

	let nested = root
		.get_content()
		.is_ok_and(|content: ExtractRelatesTo| content.relates_to.rel_type().is_some());

	if !nested {
		return Ok(());
	}

	Err!(Request(Unknown("Cannot start threads from an event with a relation.")))
}

/// Check if this is a new transaction id.
///
/// Returns Some when the transaction id exists and the send must then be
/// terminated by returning the contained result.
async fn check_existing_txnid(
	services: &Services,
	txn: RoomTxn<'_>,
	event_type: &str,
) -> Result<Option<SendMessageResponse>> {
	let scope = RoomScope::Send(event_type);
	let response = services
		.transaction_ids
		.existing_room_txnid(txn, scope)
		.await;

	if let Some(response) = response.optional()? {
		return txnid_response(&response).map(Some);
	}

	let response = services
		.transaction_ids
		.existing_txnid(txn.user_id, txn.device_id, txn.txn_id)
		.await;

	let Some(response) = response.optional()? else {
		return Ok(None);
	};

	let Some(response) = legacy_txnid_response(&response)? else {
		return Ok(None);
	};

	let event_id = &response.event_id;
	let Some(pdu) = services
		.timeline
		.get_non_outlier_pdu(event_id)
		.await
		.optional()?
	else {
		return Ok(None);
	};

	if !legacy_txnid_matches(&pdu, txn.room_id, event_type, txn.user_id) {
		return Ok(None);
	}

	services
		.transaction_ids
		.add_room_txnid(txn, scope, event_id.as_bytes());

	Ok(Some(response))
}

fn txnid_response(response: &[u8]) -> Result<SendMessageResponse> {
	txnid_event_id(response).map(SendMessageResponse::new)
}

fn legacy_txnid_response(response: &[u8]) -> Result<Option<SendMessageResponse>> {
	if response.is_empty() {
		return Ok(None);
	}

	txnid_response(response).map(Some)
}

fn legacy_txnid_matches(
	pdu: &PduEvent,
	room_id: &RoomId,
	event_type: &str,
	sender_user: &UserId,
) -> bool {
	pdu.room_id == room_id && pdu.kind.to_cow_str() == event_type && pdu.sender == sender_user
}

#[cfg(test)]
mod tests {
	use ruma::{event_id, room_id, user_id};
	use serde_json::json;

	use super::{PduEvent, legacy_txnid_matches, legacy_txnid_response};

	#[test]
	fn legacy_response_requires_a_valid_nonempty_event_id() {
		assert!(
			legacy_txnid_response(b"")
				.expect("empty marker is valid")
				.is_none()
		);

		legacy_txnid_response(b"not an event ID").expect_err("invalid event ID");
		legacy_txnid_response(&[0xFF]).expect_err("invalid UTF-8");

		let response = legacy_txnid_response(b"$event:example.com")
			.expect("valid event ID")
			.expect("nonempty response");

		assert_eq!(response.event_id, event_id!("$event:example.com"));
	}

	#[test]
	fn legacy_response_requires_matching_provenance() {
		let pdu = pdu();
		assert!(legacy_txnid_matches(
			&pdu,
			room_id!("!room:example.com"),
			"m.room.message",
			user_id!("@alice:example.com"),
		));

		assert!(!legacy_txnid_matches(
			&pdu,
			room_id!("!other:example.com"),
			"m.room.message",
			user_id!("@alice:example.com"),
		));

		assert!(!legacy_txnid_matches(
			&pdu,
			room_id!("!room:example.com"),
			"m.room.encrypted",
			user_id!("@alice:example.com"),
		));

		assert!(!legacy_txnid_matches(
			&pdu,
			room_id!("!room:example.com"),
			"m.room.message",
			user_id!("@bob:example.com"),
		));
	}

	fn pdu() -> PduEvent {
		serde_json::from_value(json!({
			"type": "m.room.message",
			"content": {},
			"event_id": "$event:example.com",
			"room_id": "!room:example.com",
			"sender": "@alice:example.com",
			"prev_events": ["$prev:example.com"],
			"auth_events": ["$auth:example.com"],
			"origin_server_ts": 1,
			"depth": 1,
			"hashes": { "sha256": "thishashcoversallfieldsincasethisisredacted" },
		}))
		.expect("valid PDU")
	}
}