tuwunel_api_federation/

invite.rs

use std::{collections::BTreeMap, mem::take};

use axum::extract::State;
use base64::{Engine as _, engine::general_purpose};
use futures::StreamExt;
use ruma::{
	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, OwnedUserId,
	RoomId, RoomVersionId, ServerName, UserId,
	api::{
		appservice::event::push_events::{self, v1::DeviceLists},
		federation::membership::{RawStrippedState, create_invite},
	},
	events::{
		AnyStrippedStateEvent, GlobalAccountDataEventType, StateEventType,
		invite_permission_config::InvitePermission,
		push_rules::PushRulesEvent,
		room::member::{MembershipState, RoomMemberEventContent},
	},
	push,
	serde::{JsonObject, Raw},
};
use serde::Deserialize;
use tuwunel_core::{Err, Result, debug_warn, err, utils, utils::hash::sha256};
use tuwunel_matrix::{Event, PduCount, PduEvent, event::gen_event_id};
use tuwunel_service::{
	Services,
	membership::{
		StrippedCreateVerdict, dedup_stripped_state, enforce_stripped_create,
		into_client_stripped, v12_room_ids, without_member,
	},
	rooms::state_cache::MembershipUpdate,
};

use crate::{ClientIp, Ruma};

/// The one membership field the invite acceptance policy reads.
#[derive(Deserialize)]
struct ExtractIsDirect {
	#[serde(default)]
	is_direct: bool,
}

struct LocalInvite<'a> {
	invited_user: &'a UserId,
	sender: &'a UserId,
	invite_state: Vec<Raw<AnyStrippedStateEvent>>,
	pdu: &'a PduEvent,
	invite_event: &'a EventId,
}

/// # `PUT /_matrix/federation/v2/invite/{roomId}/{eventId}`
///
/// Invites a remote user to a room.
#[tracing::instrument(skip_all, fields(%client), name = "invite")]
pub(crate) async fn create_invite_route(
	State(services): State<crate::State>,
	ClientIp(client): ClientIp,
	mut body: Ruma<create_invite::v2::Request>,
) -> Result<create_invite::v2::Response> {
	services
		.sending
		.notify_peer_alive(body.origin())
		.await;

	validate_request(&services, &body).await?;

	// Settled before any reader, so validation and the invitee agree.
	let stripped_state = dedup_stripped_state(take(&mut body.body.invite_room_state));

	enforce_stripped_state(&services, &body, &stripped_state).await?;

	let (mut signed_event, invited_user) = parse_and_validate_event(&services, &body).await?;

	let invite_event = sign_event(&services, &mut signed_event, &body.room_version)?;

	let sender = validate_origins(&signed_event, body.origin())?;

	check_invite_permitted(&services, &body, &invited_user, sender).await?;

	let pdu = build_pdu(&body)?;

	// Built above the room lock, so the per-entry reparse stays outside it.
	let invite_state: Vec<_> = without_member(stripped_state, &invited_user)
		.filter_map(|state| into_client_stripped(&body.room_id, state))
		.chain([pdu.to_format()])
		.collect();

	// Block on the inbound /send applying the departure that removes our last
	// member, so the residency check observes it rather than stale state.
	let _federation_lock = services
		.event_handler
		.mutex_federation
		.lock(&body.room_id)
		.await;

	let invite = LocalInvite {
		invited_user: &invited_user,
		sender,
		invite_state,
		pdu: &pdu,
		invite_event: &invite_event,
	};

	record_local_invite(&services, &body, invite).await?;

	Ok(create_invite::v2::Response {
		event: services
			.federation
			.format_pdu_into(signed_event, Some(&body.room_version))
			.await,
	})
}

async fn validate_request(
	services: &Services,
	body: &Ruma<create_invite::v2::Request>,
) -> Result<()> {
	services
		.event_handler
		.acl_check(body.origin(), &body.room_id)
		.await?;

	services
		.config
		.require_supported_room_version(&body.room_version)?;

	if let Some(server) = body.room_id.server_name()
		&& services
			.config
			.is_forbidden_remote_server_name(server)
	{
		return Err!(Request(Forbidden("Server is banned on this homeserver.")));
	}

	Ok(())
}

/// Validate the create event in the invite's stripped state (MSC4311) and
/// reject the invite when the operator's policy requires it.
async fn enforce_stripped_state(
	services: &Services,
	body: &Ruma<create_invite::v2::Request>,
	stripped_state: &[RawStrippedState],
) -> Result {
	// query-depth firewall
	let verdict = Box::pin(services.membership.validate_stripped_create(
		stripped_state,
		&body.room_id,
		&body.room_version,
	))
	.await?;

	if verdict != StrippedCreateVerdict::Valid {
		debug_warn!(
			?verdict,
			room_id = %body.room_id,
			"MSC4311 invite create-event validation failed",
		);
	}

	let enforce = services
		.config
		.federation
		.enforce_stripped_state_pdu_validation;

	if enforce_stripped_create(verdict, v12_room_ids(&body.room_version), enforce) {
		return Err!(Request(MissingParam(
			"The invite's m.room.create event is missing or does not validate for this room."
		)));
	}

	Ok(())
}

async fn parse_and_validate_event(
	services: &Services,
	body: &Ruma<create_invite::v2::Request>,
) -> Result<(CanonicalJsonObject, OwnedUserId)> {
	let signed_event = utils::to_canonical_object(&body.event)
		.map_err(|_| err!(Request(InvalidParam("Invite event is invalid."))))?;

	let room_id: OwnedRoomId = signed_event
		.get("room_id")
		.try_into()
		.map(RoomId::to_owned)
		.map_err(|e| err!(Request(InvalidParam("Invalid room_id property: {e}"))))?;

	if body.room_id != room_id {
		return Err!(Request(InvalidParam("Event room_id does not match the request path.")));
	}

	let kind: StateEventType = signed_event
		.get("type")
		.and_then(CanonicalJsonValue::as_str)
		.ok_or_else(|| err!(Request(BadJson("Missing type in event."))))?
		.into();

	if kind != StateEventType::RoomMember {
		return Err!(Request(InvalidParam("Event must be m.room.member type.")));
	}

	let invited_user: OwnedUserId = signed_event
		.get("state_key")
		.try_into()
		.map(UserId::to_owned)
		.map_err(|e| err!(Request(InvalidParam("Invalid state_key property: {e}"))))?;

	if !services.globals.user_is_local(&invited_user) {
		return Err!(Request(InvalidParam("User does not belong to this homeserver.")));
	}

	let content: RoomMemberEventContent = signed_event
		.get("content")
		.cloned()
		.map(Into::into)
		.map(serde_json::from_value)
		.transpose()
		.map_err(|e| err!(Request(InvalidParam("Invalid content object in event: {e}"))))?
		.ok_or_else(|| err!(Request(BadJson("Missing content in event."))))?;

	if content.membership != MembershipState::Invite {
		return Err!(Request(InvalidParam("Event membership must be invite.")));
	}

	services
		.event_handler
		.acl_check(invited_user.server_name(), &body.room_id)
		.await?;

	Ok((signed_event, invited_user))
}

fn sign_event(
	services: &Services,
	signed_event: &mut CanonicalJsonObject,
	room_version: &RoomVersionId,
) -> Result<OwnedEventId> {
	services
		.server_keys
		.hash_and_sign_event(signed_event, room_version)
		.map_err(|e| err!(Request(InvalidParam("Failed to sign event: {e}"))))?;

	let event_id = gen_event_id(signed_event, room_version)?;
	signed_event.insert("event_id".into(), CanonicalJsonValue::String(event_id.to_string()));

	Ok(event_id)
}

fn validate_origins<'a>(
	signed_event: &'a CanonicalJsonObject,
	body_origin: &ServerName,
) -> Result<&'a UserId> {
	let origin: Option<&str> = signed_event
		.get("origin")
		.and_then(CanonicalJsonValue::as_str);

	let sender: &UserId = signed_event
		.get("sender")
		.try_into()
		.map_err(|e| err!(Request(InvalidParam("Invalid sender property: {e}"))))?;

	if sender.server_name() != body_origin {
		return Err!(Request(Forbidden("Can only send invites on behalf of your users.")));
	}

	if origin.is_some_and(|origin| origin != body_origin) {
		return Err!(Request(Forbidden("Can only send events from your origin.")));
	}

	Ok(sender)
}

async fn check_invite_permitted(
	services: &Services,
	body: &Ruma<create_invite::v2::Request>,
	invited_user: &UserId,
	sender: &UserId,
) -> Result<()> {
	if services.metadata.is_banned(&body.room_id).await
		&& !services.admin.user_is_admin(invited_user).await
	{
		return Err!(Request(Forbidden("This room is banned on this homeserver.")));
	}

	if services.config.rooms.invites.block_non_admin
		&& !services.admin.user_is_admin(invited_user).await
	{
		return Err!(Request(Forbidden("This server does not allow room invites.")));
	}

	// The recipient's own policy is the only per-sender answer this route
	// gives, so it waits for the origin check to bind the sender. Answering
	// earlier would let any server read a local user's rules for user ids it
	// cannot speak for.
	if services
		.users
		.invite_permission(sender, invited_user)
		.await
		.eq(&InvitePermission::Block)
	{
		return Err!(Request(InviteBlocked("{invited_user} has blocked this invite.")));
	}

	Ok(())
}

fn build_pdu(body: &Ruma<create_invite::v2::Request>) -> Result<PduEvent> {
	let mut event: JsonObject = serde_json::from_str(body.event.get())
		.map_err(|e| err!(Request(BadJson("Invalid invite event PDU: {e}"))))?;

	event.insert("event_id".into(), "$placeholder".into());

	serde_json::from_value(event.into())
		.map_err(|e| err!(Request(BadJson("Invalid invite event PDU: {e}"))))
}

/// Record an invite for a room we are not currently in.
///
/// When we are active in the room, the remote server will notify us about the
/// join/invite through `/send`. When we are not in the room, the invited state
/// must be recorded manually for client `/sync` through `update_membership()`,
/// and the invite PDU pushed to the relevant appservices.
async fn record_local_invite(
	services: &Services,
	body: &Ruma<create_invite::v2::Request>,
	LocalInvite {
		invited_user,
		sender,
		invite_state,
		pdu,
		invite_event,
	}: LocalInvite<'_>,
) -> Result {
	let state_lock = services.state.mutex.lock(&body.room_id).await;

	if services
		.state_cache
		.server_in_room(services.globals.server_name(), &body.room_id)
		.await
	{
		return Ok(());
	}

	if services
		.state_accessor
		.room_state_get_content::<RoomMemberEventContent>(
			&body.room_id,
			&StateEventType::RoomMember,
			invited_user.as_str(),
		)
		.await
		.is_ok_and(|content| content.membership == MembershipState::Ban)
	{
		debug_warn!(
			room_id = %body.room_id,
			user_id = %invited_user,
			"Recording invite while local room state shows banned membership.",
		);
	}

	let count = services.globals.next_count();
	services
		.state_cache
		.update_membership(MembershipUpdate {
			room_id: &body.room_id,
			user_id: invited_user,
			membership_event: RoomMemberEventContent::new(MembershipState::Invite),
			sender,
			last_state: Some(invite_state),
			invite_via: body.via.clone(),
			invite_event: Some(invite_event),
			update_joined_count: true,
			replaced_join: None,
			count: PduCount::Normal(*count),
		})
		.await?;

	drop(count);
	drop(state_lock);

	let is_direct = pdu
		.get_content()
		.is_ok_and(|content: ExtractIsDirect| content.is_direct);

	services
		.membership
		.auto_accept(&body.room_id, invited_user, sender, is_direct);

	notify_pushers(services, invited_user, pdu).await;

	for appservice in services.appservice.read().await.values() {
		if appservice.is_user_match(invited_user) {
			services
				.appservice
				.send_request(appservice.registration.clone(), push_events::v1::Request {
					events: vec![pdu.to_format()],
					txn_id: general_purpose::URL_SAFE_NO_PAD
						.encode(sha256::hash(pdu.event_id.as_bytes()))
						.into(),
					ephemeral: Vec::new(),
					to_device: Vec::new(),
					device_lists: DeviceLists::new(),
					device_one_time_keys_count: BTreeMap::new(),
					device_unused_fallback_key_types: BTreeMap::new(),
				})
				.await
				.map_err(|_| {
					err!(BadServerResponse("Failed to notify appservice about incoming invite."))
				})?;
		}
	}

	Ok(())
}

async fn notify_pushers(services: &Services, invited_user: &UserId, pdu: &PduEvent) {
	services
		.pusher
		.get_pushkeys(invited_user)
		.map(ToOwned::to_owned)
		.for_each(async |pushkey| {
			let Ok(pusher) = services
				.pusher
				.get_pusher(invited_user, &pushkey)
				.await
			else {
				return;
			};

			let ruleset = services
				.account_data
				.get_global(invited_user, GlobalAccountDataEventType::PushRules)
				.await
				.map_or_else(
					|_| push::Ruleset::server_default(invited_user),
					|ev: PushRulesEvent| ev.content.global,
				);

			services
				.pusher
				.send_push_notice(invited_user, &pusher, &ruleset, pdu)
				.await
				.ok();
		})
		.await;
}