mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2026-11-13 04:06:54 +00:00
fix: Opportunistically re-use room format rules when parsing incoming PDUs
This commit is contained in:
1 parent
87030a3a22
commit
fbf81fcdeb
9 files changed
+48
-47
No files matched your search
@@ -486,7 +486,7 @@ impl crate::Context<'_> {
|
||||
.services
|
||||
.rooms
|
||||
.event_handler
|
||||
.parse_incoming_pdu(&response.pdu)
|
||||
.parse_incoming_pdu(&response.pdu, None)
|
||||
.boxed()
|
||||
.await;
|
||||
|
||||
@@ -832,7 +832,7 @@ impl crate::Context<'_> {
|
||||
.services
|
||||
.rooms
|
||||
.event_handler
|
||||
.parse_incoming_pdu(&pdu)
|
||||
.parse_incoming_pdu(&pdu, Some(&room_version_rules))
|
||||
.await
|
||||
{
|
||||
| Ok(t) => t,
|
||||
|
||||
@@ -250,6 +250,7 @@ async fn validate_invite_membership_event(
|
||||
room_id: OwnedRoomId,
|
||||
event_id: OwnedEventId,
|
||||
) -> Result<(CanonicalJsonObject, OwnedUserId, OwnedUserId)> {
|
||||
trace!(?body, "Invite membership event");
|
||||
let (pdu, target_membership, sender_user, recipient_user) = validate_any_membership_event(
|
||||
services,
|
||||
body,
|
||||
@@ -309,7 +310,7 @@ async fn validate_invite_state(
|
||||
let (state_event_room_id, state_event_id, state_event_json) = services
|
||||
.rooms
|
||||
.event_handler
|
||||
.parse_incoming_pdu(&raw_pdu)
|
||||
.parse_incoming_pdu(&raw_pdu, Some(room_version_rules))
|
||||
.await
|
||||
.map_err(|e| {
|
||||
err!(Request(InvalidParam(debug_warn!("Invalid PDU in invite state: {e}"))))
|
||||
|
||||
@@ -150,7 +150,7 @@ async fn process_inbound_transaction(
|
||||
.pdus
|
||||
.iter()
|
||||
.stream()
|
||||
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu))
|
||||
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu, None))
|
||||
.inspect_err(|e| warn!("Could not parse incoming PDU: {e}"))
|
||||
.ready_filter_map(Result::ok);
|
||||
|
||||
|
||||
@@ -94,7 +94,7 @@ pub(crate) async fn validate_any_membership_event(
|
||||
let (template_room_id, template_event_id, pdu) = services
|
||||
.rooms
|
||||
.event_handler
|
||||
.parse_incoming_pdu(body)
|
||||
.parse_incoming_pdu(body, Some(room_version_rules))
|
||||
.await
|
||||
.map_err(|e| err!(Request(BadJson("Invalid membership PDU: {e}"))))?;
|
||||
|
||||
@@ -102,9 +102,11 @@ pub(crate) async fn validate_any_membership_event(
|
||||
return Err!(Request(InvalidParam("Membership event does not belong to requested room")));
|
||||
}
|
||||
if template_event_id != expected_event_id {
|
||||
return Err!(Request(InvalidParam(
|
||||
return Err!(Request(InvalidParam(debug_warn!(
|
||||
%template_event_id,
|
||||
%expected_event_id,
|
||||
"Membership event ID does not match provided event ID"
|
||||
)));
|
||||
))));
|
||||
}
|
||||
|
||||
services
|
||||
|
||||
@@ -241,7 +241,7 @@ impl super::Service {
|
||||
|
||||
latest_events.clear();
|
||||
for raw_event in response.events {
|
||||
let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event).await?;
|
||||
let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event, None).await?;
|
||||
let pdu = PduEvent::from_id_val(&event_id, pdu_json).map_err(|e| {
|
||||
err!(Request(BadJson("Failed to parse gapfilled event {event_id}: {e}")))
|
||||
})?;
|
||||
@@ -384,8 +384,9 @@ impl super::Service {
|
||||
.send_federation_request(&remote, get_event::v1::Request::new(event_id.clone()))
|
||||
.await?;
|
||||
|
||||
let (calculated_event_id, value) =
|
||||
Self::parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)?;
|
||||
let (_, calculated_event_id, value) = self
|
||||
.parse_incoming_pdu(&res.pdu, Some(room_version_rules))
|
||||
.await?;
|
||||
|
||||
if calculated_event_id != event_id {
|
||||
Err!(Request(BadJson(warn!(
|
||||
|
||||
@@ -48,14 +48,16 @@ impl super::Service {
|
||||
let mut auth_chain_map = HashMap::with_capacity(event_auth.auth_chain.len());
|
||||
|
||||
for auth_pdu_json in event_auth.auth_chain {
|
||||
let (auth_event_room_id, auth_event_id, auth_pdu_json) =
|
||||
match self.parse_incoming_pdu(&auth_pdu_json).await {
|
||||
| Ok(parsed) => parsed,
|
||||
| Err(e) => {
|
||||
warn!(error=?e, "Dropping auth chain event as it could not be parsed");
|
||||
continue;
|
||||
},
|
||||
};
|
||||
let (auth_event_room_id, auth_event_id, auth_pdu_json) = match self
|
||||
.parse_incoming_pdu(&auth_pdu_json, Some(room_version_rules))
|
||||
.await
|
||||
{
|
||||
| Ok(parsed) => parsed,
|
||||
| Err(e) => {
|
||||
warn!(error=?e, "Dropping auth chain event as it could not be parsed");
|
||||
continue;
|
||||
},
|
||||
};
|
||||
if let Err(e) = Self::pdu_format_check_1(
|
||||
&auth_pdu_json,
|
||||
room_version_rules,
|
||||
|
||||
@@ -314,7 +314,7 @@ impl super::Service {
|
||||
.iter()
|
||||
.stream()
|
||||
.broad_filter_map(|raw_event_json| async {
|
||||
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok()
|
||||
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json, None).await.ok()
|
||||
&& parsed.0 == room_id
|
||||
{
|
||||
Some(parsed)
|
||||
@@ -351,7 +351,7 @@ impl super::Service {
|
||||
.iter()
|
||||
.stream()
|
||||
.broad_filter_map(|raw_event_json| async {
|
||||
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok()
|
||||
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json, None).await.ok()
|
||||
&& parsed.0 == room_id
|
||||
{
|
||||
Some(parsed)
|
||||
|
||||
@@ -82,25 +82,14 @@ pub(super) fn expect_event_id_array(
|
||||
}
|
||||
|
||||
impl super::Service {
|
||||
/// Parses an incoming PDU JSON object, generating an event ID for it. Does
|
||||
/// not insert the event ID into the returned object. Does not discover the
|
||||
/// room ID.
|
||||
pub(super) fn parse_incoming_pdu_with_known_room(
|
||||
pdu: &RawJsonValue,
|
||||
room_version_rules: &RoomVersionRules,
|
||||
) -> Result<(OwnedEventId, CanonicalJsonObject)> {
|
||||
let (event_id, value) =
|
||||
gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| {
|
||||
err!(Request(InvalidParam("Could not convert event to canonical json: {e}")))
|
||||
})?;
|
||||
// NOTE: validation checks are now performed by `pdu_format_check_1`.
|
||||
Ok((event_id, value))
|
||||
}
|
||||
|
||||
/// Parses an incoming PDU JSON object, generating an event ID for it and
|
||||
/// attempts to discover the associated room ID. Does not insert the event
|
||||
/// ID into the returned object.
|
||||
pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
|
||||
pub async fn parse_incoming_pdu(
|
||||
&self,
|
||||
pdu: &RawJsonValue,
|
||||
room_version_rules: Option<&RoomVersionRules>,
|
||||
) -> Result<Parsed> {
|
||||
let value = serde_json::from_str::<CanonicalJsonObject>(pdu.get()).map_err(|e| {
|
||||
err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}")))
|
||||
})?;
|
||||
@@ -111,17 +100,20 @@ impl super::Service {
|
||||
|
||||
let room_id = extract_room_id(event_type, &value)?;
|
||||
|
||||
let room_version_rules = self
|
||||
.services
|
||||
.state
|
||||
.get_room_version(&room_id)
|
||||
.await
|
||||
.unwrap_or(RoomVersionId::V1)
|
||||
.rules()
|
||||
.unwrap();
|
||||
let room_version_rules = match room_version_rules {
|
||||
| Some(r) => r,
|
||||
| None => &self
|
||||
.services
|
||||
.state
|
||||
.get_room_version(&room_id)
|
||||
.await
|
||||
.unwrap_or(RoomVersionId::V1)
|
||||
.rules()
|
||||
.unwrap(),
|
||||
};
|
||||
|
||||
let (event_id, value) =
|
||||
gen_event_id_canonical_json(pdu, &room_version_rules).map_err(|e| {
|
||||
gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| {
|
||||
err!(Request(InvalidParam("Could not convert event to canonical json: {e}")))
|
||||
})?;
|
||||
// NOTE: validation checks are now performed by `pdu_format_check_1`.
|
||||
|
||||
@@ -166,8 +166,11 @@ impl super::Service {
|
||||
/// Backfills a single PDU.
|
||||
#[tracing::instrument(skip(self, pdu), level = "debug")]
|
||||
pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) -> Result<()> {
|
||||
let (room_id, event_id, value) =
|
||||
self.services.event_handler.parse_incoming_pdu(&pdu).await?;
|
||||
let (room_id, event_id, value) = self
|
||||
.services
|
||||
.event_handler
|
||||
.parse_incoming_pdu(&pdu, None)
|
||||
.await?;
|
||||
|
||||
// Lock so we cannot backfill the same pdu twice at the same time
|
||||
let mutex_lock = self
|
||||
|
||||
Reference in new issue
Block a user