mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2025-09-11 18:33:01 +02:00
Compare commits
8 commits
676bfffd1d
...
83a97e63d2
Author | SHA1 | Date | |
---|---|---|---|
|
83a97e63d2 | ||
|
70e15c4e1a | ||
|
3230e1acae | ||
|
d0aa9eade4 | ||
|
2b61f91f1f | ||
|
59cd268a3f | ||
|
95a92e5f59 | ||
|
2f643b8fbf |
6 changed files with 75 additions and 3902 deletions
File diff suppressed because it is too large
Load diff
|
@ -275,11 +275,9 @@ where
|
||||||
|
|
||||||
if room_id_server_name != room_create_event.sender().server_name() {
|
if room_id_server_name != room_create_event.sender().server_name() {
|
||||||
warn!(
|
warn!(
|
||||||
"servername of room ID origin ({}) does not match servername of m.room.create \
|
"servername of room ID origin ({}) does not match servername of m.room.create sender ({})",
|
||||||
sender ({})",
|
|
||||||
room_id_server_name,
|
room_id_server_name,
|
||||||
room_create_event.sender().server_name()
|
room_create_event.sender().server_name());
|
||||||
);
|
|
||||||
return Ok(false);
|
return Ok(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -616,7 +616,7 @@ where
|
||||||
.map(ToOwned::to_owned),
|
.map(ToOwned::to_owned),
|
||||||
)
|
)
|
||||||
};
|
};
|
||||||
debug!("running auth check on {:?}", event.event_id());
|
|
||||||
let auth_result =
|
let auth_result =
|
||||||
auth_check(room_version, &event, current_third_party, fetch_state).await;
|
auth_check(room_version, &event, current_third_party, fetch_state).await;
|
||||||
|
|
||||||
|
@ -735,7 +735,7 @@ where
|
||||||
{
|
{
|
||||||
let mut room_id = None;
|
let mut room_id = None;
|
||||||
while let Some(sort_ev) = event {
|
while let Some(sort_ev) = event {
|
||||||
trace!(event_id = sort_ev.event_id().as_str(), "mainline");
|
debug!(event_id = sort_ev.event_id().as_str(), "mainline");
|
||||||
if room_id.is_none() {
|
if room_id.is_none() {
|
||||||
room_id = Some(sort_ev.room_id().to_owned());
|
room_id = Some(sort_ev.room_id().to_owned());
|
||||||
}
|
}
|
||||||
|
|
|
@ -80,7 +80,7 @@ where
|
||||||
// 5. Reject "due to auth events" if can't get all the auth events or some of
|
// 5. Reject "due to auth events" if can't get all the auth events or some of
|
||||||
// the auth events are also rejected "due to auth events"
|
// the auth events are also rejected "due to auth events"
|
||||||
// NOTE: Step 5 is not applied anymore because it failed too often
|
// NOTE: Step 5 is not applied anymore because it failed too often
|
||||||
debug!("Fetching auth events for {}", incoming_pdu.event_id);
|
debug!("Fetching auth events");
|
||||||
Box::pin(self.fetch_and_handle_outliers(
|
Box::pin(self.fetch_and_handle_outliers(
|
||||||
origin,
|
origin,
|
||||||
pdu_event.auth_events(),
|
pdu_event.auth_events(),
|
||||||
|
@ -92,12 +92,12 @@ where
|
||||||
|
|
||||||
// 6. Reject "due to auth events" if the event doesn't pass auth based on the
|
// 6. Reject "due to auth events" if the event doesn't pass auth based on the
|
||||||
// auth events
|
// auth events
|
||||||
debug!("Checking {} based on auth events", incoming_pdu.event_id);
|
debug!("Checking based on auth events");
|
||||||
// Build map of auth events
|
// Build map of auth events
|
||||||
let mut auth_events = HashMap::with_capacity(pdu_event.auth_events().count());
|
let mut auth_events = HashMap::with_capacity(pdu_event.auth_events().count());
|
||||||
for id in pdu_event.auth_events() {
|
for id in pdu_event.auth_events() {
|
||||||
let Ok(auth_event) = self.services.timeline.get_pdu(id).await else {
|
let Ok(auth_event) = self.services.timeline.get_pdu(id).await else {
|
||||||
warn!("Could not find auth event {id} for {}", incoming_pdu.event_id);
|
warn!("Could not find auth event {id}");
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
@ -131,7 +131,6 @@ where
|
||||||
ready(auth_events.get(&key).map(ToOwned::to_owned))
|
ready(auth_events.get(&key).map(ToOwned::to_owned))
|
||||||
};
|
};
|
||||||
|
|
||||||
debug!("running auth check to handle outlier pdu {:?}", incoming_pdu.event_id);
|
|
||||||
let auth_check = state_res::event_auth::auth_check(
|
let auth_check = state_res::event_auth::auth_check(
|
||||||
&to_room_version(&room_version_id),
|
&to_room_version(&room_version_id),
|
||||||
&pdu_event,
|
&pdu_event,
|
||||||
|
|
|
@ -1,6 +1,12 @@
|
||||||
use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant};
|
use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant};
|
||||||
|
|
||||||
use conduwuit::{Err, Result, debug, debug_info, err, implement, matrix::{EventTypeExt, PduEvent, StateKey, state_res}, trace, utils::stream::{BroadbandExt, ReadyExt}, warn, info};
|
use conduwuit::{
|
||||||
|
Err, Result, debug, debug_info, err, implement, is_equal_to,
|
||||||
|
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
|
||||||
|
trace,
|
||||||
|
utils::stream::{BroadbandExt, ReadyExt},
|
||||||
|
warn,
|
||||||
|
};
|
||||||
use futures::{FutureExt, StreamExt, future::ready};
|
use futures::{FutureExt, StreamExt, future::ready};
|
||||||
use ruma::{CanonicalJsonValue, RoomId, ServerName, events::StateEventType};
|
use ruma::{CanonicalJsonValue, RoomId, ServerName, events::StateEventType};
|
||||||
|
|
||||||
|
@ -11,19 +17,22 @@ use crate::rooms::{
|
||||||
};
|
};
|
||||||
|
|
||||||
#[implement(super::Service)]
|
#[implement(super::Service)]
|
||||||
pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
|
||||||
&self,
|
&self,
|
||||||
incoming_pdu: PduEvent,
|
incoming_pdu: PduEvent,
|
||||||
val: BTreeMap<String, CanonicalJsonValue>,
|
val: BTreeMap<String, CanonicalJsonValue>,
|
||||||
create_event: &PduEvent,
|
create_event: &Pdu,
|
||||||
origin: &ServerName,
|
origin: &ServerName,
|
||||||
room_id: &RoomId,
|
room_id: &RoomId,
|
||||||
) -> Result<Option<RawPduId>> {
|
) -> Result<Option<RawPduId>>
|
||||||
|
where
|
||||||
|
Pdu: Event + Send + Sync,
|
||||||
|
{
|
||||||
// Skip the PDU if we already have it as a timeline event
|
// Skip the PDU if we already have it as a timeline event
|
||||||
if let Ok(pduid) = self
|
if let Ok(pduid) = self
|
||||||
.services
|
.services
|
||||||
.timeline
|
.timeline
|
||||||
.get_pdu_id(&incoming_pdu.event_id)
|
.get_pdu_id(incoming_pdu.event_id())
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
return Ok(Some(pduid));
|
return Ok(Some(pduid));
|
||||||
|
@ -32,13 +41,13 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
if self
|
if self
|
||||||
.services
|
.services
|
||||||
.pdu_metadata
|
.pdu_metadata
|
||||||
.is_event_soft_failed(&incoming_pdu.event_id)
|
.is_event_soft_failed(incoming_pdu.event_id())
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
return Err!(Request(InvalidParam("Event has been soft failed")));
|
return Err!(Request(InvalidParam("Event has been soft failed")));
|
||||||
}
|
}
|
||||||
|
|
||||||
debug!("Upgrading pdu {} from outlier to timeline pdu", incoming_pdu.event_id);
|
debug!("Upgrading to timeline pdu");
|
||||||
let timer = Instant::now();
|
let timer = Instant::now();
|
||||||
let room_version_id = get_room_version_id(create_event)?;
|
let room_version_id = get_room_version_id(create_event)?;
|
||||||
|
|
||||||
|
@ -46,8 +55,8 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
// backwards extremities doing all the checks in this list starting at 1.
|
// backwards extremities doing all the checks in this list starting at 1.
|
||||||
// These are not timeline events.
|
// These are not timeline events.
|
||||||
|
|
||||||
debug!("Resolving state at event {}", incoming_pdu.event_id);
|
debug!("Resolving state at event");
|
||||||
let mut state_at_incoming_event = if incoming_pdu.prev_events.len() == 1 {
|
let mut state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
|
||||||
self.state_at_incoming_degree_one(&incoming_pdu).await?
|
self.state_at_incoming_degree_one(&incoming_pdu).await?
|
||||||
} else {
|
} else {
|
||||||
self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_id)
|
self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_id)
|
||||||
|
@ -56,15 +65,16 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
|
|
||||||
if state_at_incoming_event.is_none() {
|
if state_at_incoming_event.is_none() {
|
||||||
state_at_incoming_event = self
|
state_at_incoming_event = self
|
||||||
.fetch_state(origin, create_event, room_id, &incoming_pdu.event_id)
|
.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
|
|
||||||
let state_at_incoming_event =
|
let state_at_incoming_event =
|
||||||
state_at_incoming_event.expect("we always set this to some above");
|
state_at_incoming_event.expect("we always set this to some above");
|
||||||
|
|
||||||
let room_version = to_room_version(&room_version_id);
|
let room_version = to_room_version(&room_version_id);
|
||||||
|
|
||||||
debug!("Performing auth check to upgrade {}", incoming_pdu.event_id);
|
debug!("Performing auth check");
|
||||||
// 11. Check the auth of the event passes based on the state of the event
|
// 11. Check the auth of the event passes based on the state of the event
|
||||||
let state_fetch_state = &state_at_incoming_event;
|
let state_fetch_state = &state_at_incoming_event;
|
||||||
let state_fetch = |k: StateEventType, s: StateKey| async move {
|
let state_fetch = |k: StateEventType, s: StateKey| async move {
|
||||||
|
@ -74,7 +84,6 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
self.services.timeline.get_pdu(event_id).await.ok()
|
self.services.timeline.get_pdu(event_id).await.ok()
|
||||||
};
|
};
|
||||||
|
|
||||||
debug!("running auth check on {}", incoming_pdu.event_id);
|
|
||||||
let auth_check = state_res::event_auth::auth_check(
|
let auth_check = state_res::event_auth::auth_check(
|
||||||
&room_version,
|
&room_version,
|
||||||
&incoming_pdu,
|
&incoming_pdu,
|
||||||
|
@ -88,25 +97,24 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
return Err!(Request(Forbidden("Event has failed auth check with state at the event.")));
|
return Err!(Request(Forbidden("Event has failed auth check with state at the event.")));
|
||||||
}
|
}
|
||||||
|
|
||||||
debug!("Gathering auth events for {}", incoming_pdu.event_id);
|
debug!("Gathering auth events");
|
||||||
let auth_events = self
|
let auth_events = self
|
||||||
.services
|
.services
|
||||||
.state
|
.state
|
||||||
.get_auth_events(
|
.get_auth_events(
|
||||||
room_id,
|
room_id,
|
||||||
&incoming_pdu.kind,
|
incoming_pdu.kind(),
|
||||||
&incoming_pdu.sender,
|
incoming_pdu.sender(),
|
||||||
incoming_pdu.state_key.as_deref(),
|
incoming_pdu.state_key(),
|
||||||
&incoming_pdu.content,
|
incoming_pdu.content(),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
let state_fetch = |k: &StateEventType, s: &str| {
|
let state_fetch = |k: &StateEventType, s: &str| {
|
||||||
let key = k.with_state_key(s);
|
let key = k.with_state_key(s);
|
||||||
ready(auth_events.get(&key).cloned())
|
ready(auth_events.get(&key).map(ToOwned::to_owned))
|
||||||
};
|
};
|
||||||
|
|
||||||
debug!("running auth check on {} with claimed state auth", incoming_pdu.event_id);
|
|
||||||
let auth_check = state_res::event_auth::auth_check(
|
let auth_check = state_res::event_auth::auth_check(
|
||||||
&room_version,
|
&room_version,
|
||||||
&incoming_pdu,
|
&incoming_pdu,
|
||||||
|
@ -117,7 +125,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
|
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
|
||||||
|
|
||||||
// Soft fail check before doing state res
|
// Soft fail check before doing state res
|
||||||
debug!("Performing soft-fail check on {}", incoming_pdu.event_id);
|
debug!("Performing soft-fail check");
|
||||||
let soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_id)) {
|
let soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_id)) {
|
||||||
| (false, _) => true,
|
| (false, _) => true,
|
||||||
| (true, None) => false,
|
| (true, None) => false,
|
||||||
|
@ -125,7 +133,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
!self
|
!self
|
||||||
.services
|
.services
|
||||||
.state_accessor
|
.state_accessor
|
||||||
.user_can_redact(&redact_id, &incoming_pdu.sender, &incoming_pdu.room_id, true)
|
.user_can_redact(&redact_id, incoming_pdu.sender(), incoming_pdu.room_id(), true)
|
||||||
.await?,
|
.await?,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
@ -145,7 +153,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
.map(ToOwned::to_owned)
|
.map(ToOwned::to_owned)
|
||||||
.ready_filter(|event_id| {
|
.ready_filter(|event_id| {
|
||||||
// Remove any that are referenced by this incoming event's prev_events
|
// Remove any that are referenced by this incoming event's prev_events
|
||||||
!incoming_pdu.prev_events.contains(event_id)
|
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
|
||||||
})
|
})
|
||||||
.broad_filter_map(|event_id| async move {
|
.broad_filter_map(|event_id| async move {
|
||||||
// Only keep those extremities were not referenced yet
|
// Only keep those extremities were not referenced yet
|
||||||
|
@ -164,7 +172,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
debug!(
|
debug!(
|
||||||
"Retained {} extremities checked against {} prev_events",
|
"Retained {} extremities checked against {} prev_events",
|
||||||
extremities.len(),
|
extremities.len(),
|
||||||
incoming_pdu.prev_events.len()
|
incoming_pdu.prev_events().count()
|
||||||
);
|
);
|
||||||
|
|
||||||
let state_ids_compressed: Arc<CompressedState> = self
|
let state_ids_compressed: Arc<CompressedState> = self
|
||||||
|
@ -179,20 +187,20 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
.map(Arc::new)
|
.map(Arc::new)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
if incoming_pdu.state_key.is_some() {
|
if incoming_pdu.state_key().is_some() {
|
||||||
debug!("Event is a state-event. Deriving new room state");
|
debug!("Event is a state-event. Deriving new room state");
|
||||||
|
|
||||||
// We also add state after incoming event to the fork states
|
// We also add state after incoming event to the fork states
|
||||||
let mut state_after = state_at_incoming_event.clone();
|
let mut state_after = state_at_incoming_event.clone();
|
||||||
if let Some(state_key) = &incoming_pdu.state_key {
|
if let Some(state_key) = incoming_pdu.state_key() {
|
||||||
let shortstatekey = self
|
let shortstatekey = self
|
||||||
.services
|
.services
|
||||||
.short
|
.short
|
||||||
.get_or_create_shortstatekey(&incoming_pdu.kind.to_string().into(), state_key)
|
.get_or_create_shortstatekey(&incoming_pdu.kind().to_string().into(), state_key)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
let event_id = &incoming_pdu.event_id;
|
let event_id = incoming_pdu.event_id();
|
||||||
state_after.insert(shortstatekey, event_id.clone());
|
state_after.insert(shortstatekey, event_id.to_owned());
|
||||||
}
|
}
|
||||||
|
|
||||||
let new_room_state = self
|
let new_room_state = self
|
||||||
|
@ -216,7 +224,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
// 14. Check if the event passes auth based on the "current state" of the room,
|
// 14. Check if the event passes auth based on the "current state" of the room,
|
||||||
// if not soft fail it
|
// if not soft fail it
|
||||||
if soft_fail {
|
if soft_fail {
|
||||||
info!("Soft failing event {}", incoming_pdu.event_id);
|
debug!("Soft failing event");
|
||||||
let extremities = extremities.iter().map(Borrow::borrow);
|
let extremities = extremities.iter().map(Borrow::borrow);
|
||||||
|
|
||||||
self.services
|
self.services
|
||||||
|
@ -234,9 +242,9 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
// Soft fail, we keep the event as an outlier but don't add it to the timeline
|
// Soft fail, we keep the event as an outlier but don't add it to the timeline
|
||||||
self.services
|
self.services
|
||||||
.pdu_metadata
|
.pdu_metadata
|
||||||
.mark_event_soft_failed(&incoming_pdu.event_id);
|
.mark_event_soft_failed(incoming_pdu.event_id());
|
||||||
|
|
||||||
warn!("Event was soft failed: {incoming_pdu:?}");
|
warn!("Event was soft failed: {:?}", incoming_pdu.event_id());
|
||||||
return Err!(Request(InvalidParam("Event has been soft failed")));
|
return Err!(Request(InvalidParam("Event has been soft failed")));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -247,7 +255,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
|
||||||
let extremities = extremities
|
let extremities = extremities
|
||||||
.iter()
|
.iter()
|
||||||
.map(Borrow::borrow)
|
.map(Borrow::borrow)
|
||||||
.chain(once(incoming_pdu.event_id.borrow()));
|
.chain(once(incoming_pdu.event_id()));
|
||||||
|
|
||||||
let pdu_id = self
|
let pdu_id = self
|
||||||
.services
|
.services
|
||||||
|
|
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue