fix: Don't process admin escape commands for local users from federation

Reviewed-By: timedout <git@nexy7574.co.uk>

src/service/admin/mod.rs

@@ -530,7 +530,12 @@ impl Service {
 		Ok(())
 	}
 
-	pub async fn is_admin_command<E>(&self, event: &E, body: &str) -> Option<InvocationSource>
+	pub async fn is_admin_command<E>(
+		&self,
+		event: &E,
+		body: &str,
+		sent_locally: bool,
+	) -> Option<InvocationSource>
 	where
 		E: Event + Send + Sync,
 	{
@@ -580,6 +585,15 @@ impl Service {
 				return None;
 			}
 
+			// Escaped commands must be sent locally (via client API), not via federation
+			if !sent_locally {
+				conduwuit::warn!(
+					"Ignoring escaped admin command from {} that arrived via federation",
+					event.sender()
+				);
+				return None;
+			}
+
 			// Looks good
 			Some(InvocationSource::EscapedCommand)
 		}

src/service/rooms/timeline/append.rs

@@ -72,6 +72,26 @@ where
 		.append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id)
 		.await?;
 
+	// Process admin commands for federation events
+	if *pdu.kind() == TimelineEventType::RoomMessage {
+		let content: ExtractBody = pdu.get_content()?;
+		if let Some(body) = content.body {
+			if let Some(source) = self
+				.services
+				.admin
+				.is_admin_command(pdu, &body, false)
+				.await
+			{
+				self.services.admin.command_with_sender(
+					body,
+					Some(pdu.event_id().into()),
+					source,
+					pdu.sender.clone().into(),
+				)?;
+			}
+		}
+	}
+
 	Ok(Some(pdu_id))
 }
 
@@ -334,15 +354,6 @@ where
 			let content: ExtractBody = pdu.get_content()?;
 			if let Some(body) = content.body {
 				self.services.search.index_pdu(shortroomid, &pdu_id, &body);
-
-				if let Some(source) = self.services.admin.is_admin_command(pdu, &body).await {
-					self.services.admin.command_with_sender(
-						body,
-						Some((pdu.event_id()).into()),
-						source,
-						pdu.sender.clone().into(),
-					)?;
-				}
 			}
 		},
 		| _ => {},

src/service/rooms/timeline/build.rs

@@ -18,7 +18,7 @@ use ruma::{
 	},
 };
 
-use super::RoomMutexGuard;
+use super::{ExtractBody, RoomMutexGuard};
 
 /// Creates a new persisted data unit and adds it to a room. This function
 /// takes a roomid_mutex_state, meaning that only this function is able to
@@ -126,6 +126,26 @@ pub async fn build_and_append_pdu(
 		.boxed()
 		.await?;
 
+	// Process admin commands for locally sent events
+	if *pdu.kind() == TimelineEventType::RoomMessage {
+		let content: ExtractBody = pdu.get_content()?;
+		if let Some(body) = content.body {
+			if let Some(source) = self
+				.services
+				.admin
+				.is_admin_command(&pdu, &body, true)
+				.await
+			{
+				self.services.admin.command_with_sender(
+					body,
+					Some(pdu.event_id().into()),
+					source,
+					pdu.sender.clone().into(),
+				)?;
+			}
+		}
+	}
+
 	// We set the room state after inserting the pdu, so that we never have a moment
 	// in time where events in the current room state do not exist
 	trace!("Setting room state for room {room_id}");
@@ -167,6 +187,8 @@ pub async fn build_and_append_pdu(
 	Ok(pdu.event_id().to_owned())
 }
 
+/// Assert invariants about the admin room, to prevent (for example) all admins
+/// from leaving or being banned from the room
 #[implement(super::Service)]
 #[tracing::instrument(skip_all, level = "debug")]
 async fn check_pdu_for_admin_room<Pdu>(&self, pdu: &Pdu, sender: &UserId) -> Result