oss/conduit
oss/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-08-11 17:50:59 +00:00
Code Wiki Activity

feat: MSC4291, Room IDs as hashes of the create event (2/2)

Browse source
This commit is contained in:
Matthias Ahouansou 2025-08-01 16:56:26 +02:00
parent 4b833037ea
commit bd8686ec20
No known key found for this signature in database
20 changed files with 432 additions and 251 deletions
Download patch file Download diff file Expand all files Collapse all files

22
Cargo.lock generated
View file

@ -2521,7 +2521,7 @@ dependencies = [
[[package]]
name = "ruma"
version = "0.12.6"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"assign",
"js_int",
@ -2540,7 +2540,7 @@ dependencies = [
[[package]]
name = "ruma-appservice-api"
version = "0.12.2"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"js_int",
"ruma-common",
@ -2552,7 +2552,7 @@ dependencies = [
[[package]]
name = "ruma-client-api"
version = "0.20.4"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"as_variant",
"assign",
@ -2575,7 +2575,7 @@ dependencies = [
[[package]]
name = "ruma-common"
version = "0.15.4"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"as_variant",
"base64 0.22.1",
@ -2607,7 +2607,7 @@ dependencies = [
[[package]]
name = "ruma-events"
version = "0.30.5"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"as_variant",
"indexmap 2.9.0",
@ -2631,7 +2631,7 @@ dependencies = [
[[package]]
name = "ruma-federation-api"
version = "0.11.2"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"bytes",
"headers",
@ -2653,7 +2653,7 @@ dependencies = [
[[package]]
name = "ruma-identifiers-validation"
version = "0.10.1"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"js_int",
"thiserror 2.0.12",
@ -2662,7 +2662,7 @@ dependencies = [
[[package]]
name = "ruma-macros"
version = "0.15.2"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"cfg-if",
"proc-macro-crate",
@ -2677,7 +2677,7 @@ dependencies = [
[[package]]
name = "ruma-push-gateway-api"
version = "0.11.0"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"js_int",
"ruma-common",
@ -2689,7 +2689,7 @@ dependencies = [
[[package]]
name = "ruma-signatures"
version = "0.17.1"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"base64 0.22.1",
"ed25519-dalek",
@ -2705,7 +2705,7 @@ dependencies = [
[[package]]
name = "ruma-state-res"
version = "0.13.0"
source = "git+https://github.com/ruma/ruma.git#69914f4a34da2eb6bedc6481db1674abee74be36"
source = "git+https://github.com/ruma/ruma.git#796cc9f3aacb7e69f6ec3a0d5c2ba900c3e65910"
dependencies = [
"js_int",
"ruma-common",

2
src/api/client_server/context.rs
View file

@ -46,7 +46,7 @@ pub async fn get_context_route(
"Base event not found.",
))?;
let room_id = base_event.room_id.clone();
let room_id = base_event.room_id();
if !services()
.rooms

7
src/api/client_server/membership.rs
View file

@ -705,16 +705,15 @@ pub(crate) async fn invite_helper(
timestamp: None,
},
sender_user,
room_id,
&state_lock,
Some((room_id, &state_lock)),
)?;
let mut invite_room_state = services()
.rooms
.state
.stripped_state_federation(&pdu.room_id)?;
.stripped_state_federation(&pdu.room_id())?;
if let Some(sender) = services().rooms.state_accessor.room_state_get(
&pdu.room_id,
&pdu.room_id(),
&StateEventType::RoomMember,
sender_user.as_str(),
)? {

139
src/api/client_server/room.rs
View file

@ -23,7 +23,7 @@ use ruma::{
},
int,
serde::JsonObject,
CanonicalJsonObject, CanonicalJsonValue, OwnedRoomAliasId, OwnedUserId, RoomAliasId, RoomId,
CanonicalJsonObject, CanonicalJsonValue, OwnedRoomAliasId, OwnedUserId, RoomAliasId,
};
use serde::Deserialize;
use serde_json::{json, value::to_raw_value};
@ -57,21 +57,6 @@ pub async fn create_room_route(
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let room_id = RoomId::new_v1(services().globals.server_name());
services().rooms.short.get_or_create_shortroomid(&room_id)?;
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(room_id.clone())
.or_default(),
);
let state_lock = mutex_state.lock().await;
if !services().globals.allow_room_creation()
&& body.appservice_info.is_none()
&& !services().users.is_admin(sender_user)?
@ -250,23 +235,16 @@ pub async fn create_room_route(
}
// 1. The room create event
services()
let (room_id, mutex_state) = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: TimelineEventType::RoomCreate,
content: to_raw_value(&content).expect("event is valid, we just created it"),
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
timestamp: None,
},
.send_create_room(
to_raw_value(&content).expect("event is valid, we just created it"),
sender_user,
&room_id,
&state_lock,
&rules,
)
.await?;
let state_lock = mutex_state.lock().await;
// 2. Let the room creator join
services()
@ -541,7 +519,7 @@ pub async fn get_room_event_route(
if !services().rooms.state_accessor.user_can_see_event(
sender_user,
&event.room_id,
&event.room_id(),
&body.event_id,
)? {
return Err(Error::BadRequest(
@ -621,61 +599,6 @@ pub async fn upgrade_room_route(
.expect("Supported room version must have rules.")
.authorization;
// Create a replacement room
let replacement_room = RoomId::new_v1(services().globals.server_name());
services()
.rooms
.short
.get_or_create_shortroomid(&replacement_room)?;
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(body.room_id.clone())
.or_default(),
);
let state_lock = mutex_state.lock().await;
// Send a m.room.tombstone event to the old room to indicate that it is not intended to be used any further
// Fail if the sender does not have the required permissions
let tombstone_event_id = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: TimelineEventType::RoomTombstone,
content: to_raw_value(&RoomTombstoneEventContent {
body: "This room has been replaced".to_owned(),
replacement_room: replacement_room.clone(),
})
.expect("event is valid, we just created it"),
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
timestamp: None,
},
sender_user,
&body.room_id,
&state_lock,
)
.await?;
// Change lock to replacement room
drop(state_lock);
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(replacement_room.clone())
.or_default(),
);
let state_lock = mutex_state.lock().await;
// Get the old room creation event
let mut create_event_content = serde_json::from_str::<CanonicalJsonObject>(
services()
@ -689,11 +612,9 @@ pub async fn upgrade_room_route(
.map_err(|_| Error::bad_database("Invalid room event in database."))?;
// Use the m.room.tombstone event as the predecessor
#[allow(deprecated)]
let predecessor = Some(ruma::events::room::create::PreviousRoom {
room_id: body.room_id.clone(),
event_id: Some((*tombstone_event_id).to_owned()),
});
let predecessor = Some(ruma::events::room::create::PreviousRoom::new(
body.room_id.clone(),
));
// Send a m.room.create event containing a predecessor field and the applicable room_version
if rules.use_room_create_sender {
@ -746,25 +667,57 @@ pub async fn upgrade_room_route(
));
}
// Lock the room being replaced
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(body.room_id.clone())
.or_default(),
);
let state_lock = mutex_state.lock().await;
// Create a replacement room
let (replacement_room, mutex_state) = services()
.rooms
.timeline
.send_create_room(
to_raw_value(&create_event_content).expect("event is valid, we just created it"),
sender_user,
&rules,
)
.await?;
// Send a m.room.tombstone event to the old room to indicate that it is not intended to be used any further
// Fail if the sender does not have the required permissions
services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: TimelineEventType::RoomCreate,
content: to_raw_value(&create_event_content)
.expect("event is valid, we just created it"),
event_type: TimelineEventType::RoomTombstone,
content: to_raw_value(&RoomTombstoneEventContent {
body: "This room has been replaced".to_owned(),
replacement_room: replacement_room.clone(),
})
.expect("event is valid, we just created it"),
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
timestamp: None,
},
sender_user,
&replacement_room,
&body.room_id,
&state_lock,
)
.await?;
// Change lock to replacement room
drop(state_lock);
let state_lock = mutex_state.lock().await;
// Join the new room
services()
.rooms

2
src/api/client_server/search.rs
View file

@ -93,7 +93,7 @@ pub async fn search_events_route(
&& services()
.rooms
.state_accessor
.user_can_see_event(sender_user, &pdu.room_id, &pdu.event_id)
.user_can_see_event(sender_user, &pdu.room_id(), &pdu.event_id)
.unwrap_or(false)
})
.map(|pdu| pdu.to_room_event())

83
src/api/client_server/state.rs
View file

@ -204,6 +204,89 @@ pub async fn get_state_event_for_empty_key_route(
Ok(response.into())
}
/// # `GET /_matrix/client/r0/rooms/{roomid}/state/{eventType}/{stateKey}`
///
/// Get single state event of a room.
///
/// - If not joined: Only works if current room history visibility is world readable
pub async fn get_state_events_for_key_route(
body: Ruma<get_state_event_for_key::v3::Request>,
) -> Result<get_state_event_for_key::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
if !services()
.rooms
.state_accessor
.user_can_see_state_events(sender_user, &body.room_id)?
{
return Err(Error::BadRequest(
ErrorKind::forbidden(),
"You don't have permission to view the room state.",
));
}
let event = services()
.rooms
.state_accessor
.room_state_get(&body.room_id, &body.event_type, &body.state_key)?
.ok_or_else(|| {
warn!(
"State event {:?} not found in room {:?}",
&body.event_type, &body.room_id
);
Error::BadRequest(ErrorKind::NotFound, "State event not found.")
})?;
Ok(get_state_event_for_key::v3::Response {
// NOTE: In the sprit of implementing each change atomically, this will be properly
// handled in the next commit.
event_or_content: serde_json::from_str(event.content.get())
.map_err(|_| Error::bad_database("Invalid event content in database"))?,
})
}
/// # `GET /_matrix/client/r0/rooms/{roomid}/state/{eventType}`
///
/// Get single state event of a room.
///
/// - If not joined: Only works if current room history visibility is world readable
pub async fn get_state_events_for_empty_key_route(
body: Ruma<get_state_event_for_key::v3::Request>,
) -> Result<RumaResponse<get_state_event_for_key::v3::Response>> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
if !services()
.rooms
.state_accessor
.user_can_see_state_events(sender_user, &body.room_id)?
{
return Err(Error::BadRequest(
ErrorKind::forbidden(),
"You don't have permission to view the room state.",
));
}
let event = services()
.rooms
.state_accessor
.room_state_get(&body.room_id, &body.event_type, "")?
.ok_or_else(|| {
warn!(
"State event {:?} not found in room {:?}",
&body.event_type, &body.room_id
);
Error::BadRequest(ErrorKind::NotFound, "State event not found.")
})?;
Ok(get_state_event_for_key::v3::Response {
// NOTE: In the sprit of implementing each change atomically, this will be properly
// handled in the next commit.
event_or_content: serde_json::from_str(event.content.get())
.map_err(|_| Error::bad_database("Invalid event content in database"))?,
}
.into())
}
async fn send_state_event_for_key_helper(
sender: &UserId,
room_id: &RoomId,

2
src/api/client_server/sync.rs
View file

@ -346,7 +346,7 @@ async fn sync_helper(
state_key: Some(sender_user.to_string()),
unsigned: None,
// The following keys are dropped on conversion
room_id: room_id.clone(),
room_id: Some(room_id.clone()),
prev_events: vec![],
depth: uint!(1),
auth_events: vec![],

5
src/api/server_server.rs
View file

@ -1207,7 +1207,7 @@ pub async fn get_backfill_route(
matches!(
services().rooms.state_accessor.server_can_see_event(
sender_servername,
&e.room_id,
&e.room_id(),
&e.event_id,
),
Ok(true),
@ -1647,8 +1647,7 @@ fn create_membership_template(
timestamp: None,
},
user_id,
room_id,
&state_lock,
Some((room_id, &state_lock)),
)?;
drop(state_lock);

2
src/database/key_value/rooms/timeline.rs
View file

@ -171,7 +171,7 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
self.lasttimelinecount_cache
.lock()
.unwrap()
.insert(pdu.room_id.clone(), PduCount::Normal(count));
.insert(pdu.room_id().into_owned(), PduCount::Normal(count));
self.eventid_pduid.insert(pdu.event_id.as_bytes(), pdu_id)?;
self.eventid_outlierpdu.remove(pdu.event_id.as_bytes())?;

4
src/database/mod.rs
View file

@ -690,8 +690,8 @@ impl KeyValueDatabase {
.unwrap()
.unwrap();
if Some(&pdu.room_id) != current_room.as_ref() {
current_room = Some(pdu.room_id.clone());
if Some(pdu.room_id().as_ref()) != current_room.as_deref() {
current_room = Some(pdu.room_id().into_owned());
}
}

32
src/service/admin/mod.rs
View file

@ -1672,21 +1672,6 @@ impl Service {
/// Users in this room are considered admins by conduit, and the room can be
/// used to issue admin commands by talking to the server user inside it.
pub(crate) async fn create_admin_room(&self) -> Result<()> {
let room_id = RoomId::new_v1(services().globals.server_name());
services().rooms.short.get_or_create_shortroomid(&room_id)?;
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(room_id.clone())
.or_default(),
);
let state_lock = mutex_state.lock().await;
// Create a user for the server
let conduit_user = services().globals.server_user();
@ -1707,23 +1692,16 @@ impl Service {
content.room_version = room_version;
// 1. The room create event
services()
let (room_id, mutex_state) = services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: TimelineEventType::RoomCreate,
content: to_raw_value(&content).expect("event is valid, we just created it"),
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
timestamp: None,
},
.send_create_room(
to_raw_value(&content).expect("event is valid, we just created it"),
conduit_user,
&room_id,
&state_lock,
&rules,
)
.await?;
let state_lock = mutex_state.lock().await;
// 2. Make conduit bot join
services()

33
src/service/pdu.rs
View file

@ -19,7 +19,7 @@ use serde_json::{
json,
value::{to_raw_value, RawValue as RawJsonValue},
};
use std::{cmp::Ordering, collections::BTreeMap, sync::Arc};
use std::{borrow::Cow, cmp::Ordering, collections::BTreeMap, sync::Arc};
use tracing::warn;
/// Content hashes of a PDU.
@ -32,7 +32,8 @@ pub struct EventHash {
#[derive(Clone, Deserialize, Debug, Serialize)]
pub struct PduEvent {
pub event_id: Arc<EventId>,
pub room_id: OwnedRoomId,
#[serde(skip_serializing_if = "Option::is_none")]
pub room_id: Option<OwnedRoomId>,
pub sender: OwnedUserId,
pub origin_server_ts: UInt,
#[serde(rename = "type")]
@ -149,6 +150,22 @@ impl PduEvent {
(self.redacts.clone(), self.content.clone())
}
/// Gets the room id of the PDU.
///
/// From `org.matrix.hydra.11`, this constructs the room ID from the event ID if it is an
/// `m.room.create` event.
pub fn room_id(&self) -> Cow<RoomId> {
if let Some(room_id) = &self.room_id {
Cow::Borrowed(room_id)
} else {
Cow::Owned(RoomId::new_v2(self.event_id.localpart()).expect(
"Only create events from `org.matrix.hydra.11` onwards\
don't have a room id, and those events use the\
event id as reference hash format",
))
}
}
#[tracing::instrument(skip(self))]
pub fn to_sync_room_event(&self) -> Raw<AnySyncTimelineEvent> {
let (redacts, content) = self.copy_redacts();
@ -206,7 +223,7 @@ impl PduEvent {
"event_id": self.event_id,
"sender": self.sender,
"origin_server_ts": self.origin_server_ts,
"room_id": self.room_id,
"room_id": self.room_id(),
});
if let Some(unsigned) = &self.unsigned {
@ -231,7 +248,7 @@ impl PduEvent {
"event_id": self.event_id,
"sender": self.sender,
"origin_server_ts": self.origin_server_ts,
"room_id": self.room_id,
"room_id": self.room_id(),
});
if let Some(unsigned) = &self.unsigned {
@ -255,7 +272,7 @@ impl PduEvent {
"event_id": self.event_id,
"sender": self.sender,
"origin_server_ts": self.origin_server_ts,
"room_id": self.room_id,
"room_id": self.room_id(),
"state_key": self.state_key,
});
@ -318,7 +335,7 @@ impl PduEvent {
"sender": self.sender,
"origin_server_ts": self.origin_server_ts,
"redacts": self.redacts,
"room_id": self.room_id,
"room_id": self.room_id(),
"state_key": self.state_key,
});
@ -373,8 +390,8 @@ impl state_res::Event for PduEvent {
&self.event_id
}
fn room_id(&self) -> &RoomId {
&self.room_id
fn room_id(&self) -> Option<&RoomId> {
self.room_id.as_deref()
}
fn sender(&self) -> &UserId {

12
src/service/pusher/mod.rs
View file

@ -139,7 +139,10 @@ impl Service {
let mut notify = None;
let mut tweaks = Vec::new();
let power_levels = services().rooms.state_accessor.power_levels(&pdu.room_id)?;
let power_levels = services()
.rooms
.state_accessor
.power_levels(&pdu.room_id())?;
for action in self
.get_actions(
@ -147,7 +150,7 @@ impl Service {
&ruleset,
power_levels.into(),
&pdu.to_sync_room_event(),
&pdu.room_id,
&pdu.room_id(),
)
.await?
{
@ -231,7 +234,7 @@ impl Service {
notifi.prio = NotificationPriority::Low;
notifi.event_id = Some((*event.event_id).to_owned());
notifi.room_id = Some((*event.room_id).to_owned());
notifi.room_id = Some((*event.room_id()).to_owned());
// TODO: missed calls
notifi.counts = NotificationCounts::new(unread, uint!(0));
@ -258,7 +261,8 @@ impl Service {
notifi.sender_display_name = services().users.displayname(&event.sender)?;
notifi.room_name = services().rooms.state_accessor.get_name(&event.room_id)?;
notifi.room_name =
services().rooms.state_accessor.get_name(&event.room_id())?;
self.send_request(&http.url, send_event_notification::v1::Request::new(notifi))
.await?;

2
src/service/rooms/auth_chain/mod.rs
View file

@ -132,7 +132,7 @@ impl Service {
while let Some(event_id) = todo.pop() {
match services().rooms.timeline.get_pdu(&event_id) {
Ok(Some(pdu)) => {
if pdu.room_id != room_id {
if pdu.room_id().as_ref() != room_id {
return Err(Error::BadRequest(
ErrorKind::forbidden(),
"Evil event in db",

39
src/service/rooms/event_handler/mod.rs
View file

@ -460,12 +460,16 @@ impl Service {
// Build map of auth events
let mut auth_events = HashMap::new();
let mut auth_events_by_event_id = HashMap::new();
for id in &incoming_pdu.auth_events {
let insert_auth_event = |auth_events: &mut HashMap<_, _>,
auth_events_by_event_id: &mut HashMap<_, _>,
id|
-> Result<()> {
let auth_event = match services().rooms.timeline.get_pdu(id)? {
Some(e) => e,
None => {
warn!("Could not find auth event {}", id);
continue;
return Ok(());
}
};
@ -480,6 +484,25 @@ impl Service {
),
auth_event,
);
Ok(())
};
for id in &incoming_pdu.auth_events {
insert_auth_event(&mut auth_events, &mut auth_events_by_event_id, id)?;
}
// Create event is always needed to authorize any event (besides the create events itself)
if room_version_rules
.authorization
.room_create_event_id_as_room_id
{
if let Some(room_id) = &incoming_pdu.room_id {
let event_id = EventId::parse(format!("${}",room_id.strip_sigil())).map_err(|_| {
Error::BadRequest(ErrorKind::InvalidParam, "Room ID cannot be converted to event ID, despite room version rules requiring so.")
})?;
insert_auth_event(&mut auth_events, &mut auth_events_by_event_id, &event_id)?;
}
}
// first time we are doing any sort of auth check, so we check state-independent
@ -856,7 +879,7 @@ impl Service {
!services().rooms.state_accessor.user_can_redact(
redact_id,
&incoming_pdu.sender,
&incoming_pdu.room_id,
&incoming_pdu.room_id(),
true,
)?
} else {
@ -866,7 +889,7 @@ impl Service {
!services().rooms.state_accessor.user_can_redact(
redact_id,
&incoming_pdu.sender,
&incoming_pdu.room_id,
&incoming_pdu.room_id(),
true,
)?
} else {
@ -1979,8 +2002,12 @@ impl Service {
}
fn check_room_id(&self, room_id: &RoomId, pdu: &PduEvent) -> Result<()> {
if pdu.room_id != room_id {
warn!("Found event from room {} in room {}", pdu.room_id, room_id);
if pdu.room_id().as_ref() != room_id {
warn!(
"Found event from room {} in room {}",
pdu.room_id(),
room_id
);
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
"Event has wrong room id",

12
src/service/rooms/state/mod.rs
View file

@ -100,7 +100,7 @@ impl Service {
.roomid_spacehierarchy_cache
.lock()
.await
.remove(&pdu.room_id);
.remove(pdu.room_id().as_ref());
}
_ => continue,
}
@ -197,7 +197,7 @@ impl Service {
.short
.get_or_create_shorteventid(&new_pdu.event_id)?;
let previous_shortstatehash = self.get_room_shortstatehash(&new_pdu.room_id)?;
let previous_shortstatehash = self.get_room_shortstatehash(&new_pdu.room_id())?;
if let Some(p) = previous_shortstatehash {
self.db.set_event_state(shorteventid, p)?;
@ -365,10 +365,16 @@ impl Service {
return Ok(HashMap::new());
};
let auth_events =
let mut auth_events =
state_res::auth_types_for_event(kind, sender, state_key, content, auth_rules)
.expect("content is a valid JSON object");
// We always need the room create to check the state anyways, we just need to make sure
// to remove it when creating events if required to do so by the auth rules.
if auth_rules.room_create_event_id_as_room_id {
auth_events.push((StateEventType::RoomCreate, "".into()));
}
let mut sauthevents = auth_events
.into_iter()
.filter_map(|(event_type, state_key)| {

2
src/service/rooms/state_accessor/mod.rs
View file

@ -330,7 +330,7 @@ impl Service {
Ok(services()
.rooms
.timeline
.create_hash_and_sign_event(new_event, sender, room_id, state_lock)
.create_hash_and_sign_event(new_event, sender, Some((room_id, state_lock)))
.is_ok())
}

9
src/service/rooms/state_cache/mod.rs
View file

@ -420,12 +420,9 @@ impl Service {
.map(|event| event.sender().server_name().to_owned()),
);
servers.push(
room_id
.server_name()
.expect("Room IDs should always have a server name")
.into(),
);
if let Some(server_name) = room_id.server_name() {
servers.push(server_name.to_owned())
};
(servers, room_id)
}

272
src/service/rooms/timeline/mod.rs
View file

@ -3,6 +3,7 @@ mod data;
use std::{
cmp::Ordering,
collections::{BTreeMap, HashMap, HashSet},
ops::Deref as _,
sync::Arc,
};
@ -20,6 +21,7 @@ use ruma::{
GlobalAccountDataEventType, StateEventType, TimelineEventType,
},
push::{Action, PushConditionPowerLevelsCtx, Ruleset, Tweak},
room_version_rules::AuthorizationRules,
state_res::{self, Event},
uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
OwnedEventId, OwnedRoomId, OwnedServerName, RoomId, ServerName, UserId,
@ -208,7 +210,7 @@ impl Service {
let shortroomid = services()
.rooms
.short
.get_shortroomid(&pdu.room_id)?
.get_shortroomid(&pdu.room_id())?
.expect("room exists");
// Make unsigned fields correct. This is not properly documented in the spec, but state
@ -249,11 +251,11 @@ impl Service {
services()
.rooms
.pdu_metadata
.mark_as_referenced(&pdu.room_id, &pdu.prev_events)?;
.mark_as_referenced(&pdu.room_id(), &pdu.prev_events)?;
services()
.rooms
.state
.set_forward_extremities(&pdu.room_id, leaves, state_lock)?;
.set_forward_extremities(&pdu.room_id(), leaves, state_lock)?;
let mutex_insert = Arc::clone(
services()
@ -261,7 +263,7 @@ impl Service {
.roomid_mutex_insert
.write()
.await
.entry(pdu.room_id.clone())
.entry(pdu.room_id().into_owned())
.or_default(),
);
let insert_lock = mutex_insert.lock().await;
@ -273,11 +275,11 @@ impl Service {
.rooms
.edus
.read_receipt
.private_read_set(&pdu.room_id, &pdu.sender, count1)?;
.private_read_set(&pdu.room_id(), &pdu.sender, count1)?;
services()
.rooms
.user
.reset_notification_counts(&pdu.sender, &pdu.room_id)?;
.reset_notification_counts(&pdu.sender, &pdu.room_id())?;
let count2 = services().globals.next_count()?;
let mut pdu_id = shortroomid.to_be_bytes().to_vec();
@ -298,7 +300,7 @@ impl Service {
if let Ok(power_levels) = services()
.rooms
.state_accessor
.power_levels(&pdu.room_id)
.power_levels(&pdu.room_id())
.map(PushConditionPowerLevelsCtx::from)
{
let sync_pdu = pdu.to_sync_room_event();
@ -309,7 +311,7 @@ impl Service {
let mut push_target = services()
.rooms
.state_cache
.get_our_real_users(&pdu.room_id)?;
.get_our_real_users(&pdu.room_id())?;
if pdu.kind == TimelineEventType::RoomMember {
if let Some(state_key) = &pdu.state_key {
@ -355,7 +357,7 @@ impl Service {
&rules_for_user,
power_levels.clone(),
&sync_pdu,
&pdu.room_id,
&pdu.room_id(),
)
.await?
{
@ -382,13 +384,13 @@ impl Service {
}
self.db
.increment_notification_counts(&pdu.room_id, notifies, highlights)?;
.increment_notification_counts(&pdu.room_id(), notifies, highlights)?;
}
}
match pdu.kind {
TimelineEventType::RoomRedaction => {
let room_version_id = services().rooms.state.get_room_version(&pdu.room_id)?;
let room_version_id = services().rooms.state.get_room_version(&pdu.room_id())?;
let rules = room_version_id
.rules()
.expect("Supported room version must have rules.")
@ -404,7 +406,7 @@ impl Service {
if services().rooms.state_accessor.user_can_redact(
redact_id,
&pdu.sender,
&pdu.room_id,
&pdu.room_id(),
false,
)? {
self.redact_pdu(redact_id, pdu, shortroomid)?;
@ -414,7 +416,7 @@ impl Service {
if services().rooms.state_accessor.user_can_redact(
redact_id,
&pdu.sender,
&pdu.room_id,
&pdu.room_id(),
false,
)? {
self.redact_pdu(redact_id, pdu, shortroomid)?;
@ -429,7 +431,7 @@ impl Service {
.roomid_spacehierarchy_cache
.lock()
.await
.remove(&pdu.room_id);
.remove(pdu.room_id().deref());
}
}
TimelineEventType::RoomMember => {
@ -448,8 +450,10 @@ impl Service {
let stripped_state = match content.membership {
MembershipState::Invite | MembershipState::Knock => {
let mut state =
services().rooms.state.stripped_state_client(&pdu.room_id)?;
let mut state = services()
.rooms
.state
.stripped_state_client(&pdu.room_id())?;
// So that clients can get info about who invitied them (not relevant for knocking), the reason, when, etc.
state.push(pdu.to_stripped_state_event().cast());
Some(state)
@ -465,7 +469,7 @@ impl Service {
// shouldn't be joining automatically anyways, since it may surprise users to
// suddenly join rooms which clients didn't even show as being knocked on before.
services().rooms.state_cache.update_membership(
&pdu.room_id,
&pdu.room_id(),
&target_user_id,
content.membership,
&pdu.sender,
@ -504,7 +508,7 @@ impl Service {
if let Some(admin_room) = services().admin.get_admin_room()? {
if to_conduit
&& !from_conduit
&& admin_room == pdu.room_id
&& admin_room == *pdu.room_id()
&& services()
.rooms
.state_cache
@ -578,7 +582,7 @@ impl Service {
if services()
.rooms
.state_cache
.appservice_in_room(&pdu.room_id, appservice)?
.appservice_in_room(&pdu.room_id(), appservice)?
{
services()
.sending
@ -621,11 +625,11 @@ impl Service {
services()
.rooms
.alias
.local_aliases_for_room(&pdu.room_id)
.local_aliases_for_room(&pdu.room_id())
.filter_map(Result::ok)
.any(|room_alias| appservice.aliases.is_match(room_alias.as_str()))
|| if let Ok(Some(pdu)) = services().rooms.state_accessor.room_state_get(
&pdu.room_id,
&pdu.room_id(),
&StateEventType::RoomCanonicalAlias,
"",
) {
@ -644,7 +648,7 @@ impl Service {
};
if matching_aliases()
|| appservice.rooms.is_match(pdu.room_id.as_str())
|| appservice.rooms.is_match(pdu.room_id().as_str())
|| matching_users()
{
services()
@ -660,8 +664,9 @@ impl Service {
&self,
pdu_builder: PduBuilder,
sender: &UserId,
room_id: &RoomId,
_mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
// Take mutex guard to make sure users get the room state mutex
// If is `None`, we're creating a new room
room_id_and_state_lock: Option<(&RoomId, &MutexGuard<'_, ()>)>,
) -> Result<(PduEvent, CanonicalJsonObject)> {
let PduBuilder {
event_type,
@ -672,44 +677,69 @@ impl Service {
timestamp,
} = pdu_builder;
let prev_events: Vec<_> = services()
.rooms
.state
.get_forward_extremities(room_id)?
.into_iter()
.take(20)
.collect();
let (prev_events, room_version_rules, auth_events, room_id) =
if let Some((room_id, _)) = room_id_and_state_lock {
let prev_events: Vec<_> = services()
.rooms
.state
.get_forward_extremities(room_id)?
.into_iter()
.take(20)
.collect();
// If there was no create event yet, assume we are creating a room
let room_version_id = services()
.rooms
.state
.get_room_version(room_id)
.or_else(|_| {
if event_type == TimelineEventType::RoomCreate {
let content = serde_json::from_str::<RoomCreateEventContent>(content.get())
.expect("Invalid content in RoomCreate pdu.");
Ok(content.room_version)
} else {
Err(Error::InconsistentRoomState(
"non-create event for room of unknown version",
room_id.to_owned(),
))
}
})?;
// If there was no create event yet, assume we are creating a room
let room_version_id =
services()
.rooms
.state
.get_room_version(room_id)
.or_else(|_| {
if event_type == TimelineEventType::RoomCreate {
let content =
serde_json::from_str::<RoomCreateEventContent>(content.get())
.expect("Invalid content in RoomCreate pdu.");
Ok(content.room_version)
} else {
Err(Error::InconsistentRoomState(
"non-create event for room of unknown version",
room_id.to_owned(),
))
}
})?;
let room_version_rules = room_version_id
.rules()
.expect("Supported room version has rules");
let room_version_rules = room_version_id
.rules()
.expect("Supported room version has rules");
let auth_events = services().rooms.state.get_auth_events(
room_id,
&event_type,
sender,
state_key.as_deref(),
&content,
&room_version_rules.authorization,
)?;
(
prev_events,
room_version_rules,
auth_events,
Some(room_id.to_owned()),
)
} else {
let content = serde_json::from_str::<RoomCreateEventContent>(content.get())
.map_err(|_| {
Error::BadRequest(ErrorKind::InvalidParam, "Invalid room create content")
})?;
let room_version_rules = content
.room_version
.rules()
.expect("Supported room version has rules");
(Vec::new(), room_version_rules, HashMap::new(), None)
};
let auth_events = services().rooms.state.get_auth_events(
room_id,
&event_type,
sender,
state_key.as_deref(),
&content,
&room_version_rules.authorization,
)?;
let mut auth_events_by_event_id = HashMap::new();
for event in auth_events.values() {
auth_events_by_event_id.insert(event.event_id.clone(), event.clone());
@ -725,7 +755,7 @@ impl Service {
let mut unsigned = unsigned.unwrap_or_default();
if let Some(state_key) = &state_key {
if let Some((state_key, room_id)) = state_key.as_deref().zip(room_id.as_deref()) {
if let Some(prev_pdu) = services().rooms.state_accessor.room_state_get(
room_id,
&event_type.to_string().into(),
@ -744,7 +774,7 @@ impl Service {
let mut pdu = PduEvent {
event_id: ruma::event_id!("$thiswillbefilledinlater").into(),
room_id: room_id.to_owned(),
room_id,
sender: sender.to_owned(),
origin_server_ts: timestamp
.map(|ts| ts.get())
@ -756,6 +786,14 @@ impl Service {
depth,
auth_events: auth_events
.values()
.filter(|event| {
// See `services().rooms.state.get_auth_events()`, we always add room create,
// since it's needed to authorize events in `state_res::check_*_auth_rules`.
event.kind != TimelineEventType::RoomCreate
|| !room_version_rules
.authorization
.room_create_event_id_as_room_id
})
.map(|pdu| pdu.event_id.clone())
.collect(),
redacts,
@ -835,13 +873,8 @@ impl Service {
// Generate event id
pdu.event_id = EventId::parse_arc(format!(
"${}",
ruma::signatures::reference_hash(
&pdu_json,
&room_version_id
.rules()
.expect("Supported room version has rules")
)
.expect("Event format validated when event was hashed")
ruma::signatures::reference_hash(&pdu_json, &room_version_rules)
.expect("Event format validated when event was hashed")
))
.expect("ruma's reference hashes are valid event ids");
@ -859,6 +892,91 @@ impl Service {
Ok((pdu, pdu_json))
}
/// Sends a room create event, returning the room ID and state mutex
#[tracing::instrument(skip(self, content, sender, rules))]
pub async fn send_create_room(
&self,
content: Box<RawJsonValue>,
sender: &UserId,
rules: &AuthorizationRules,
) -> Result<(OwnedRoomId, Arc<Mutex<()>>)> {
let builder = PduBuilder {
event_type: TimelineEventType::RoomCreate,
content,
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
timestamp: None,
};
let (pdu, pdu_json, room_id, mutex_state) = if rules.room_create_event_id_as_room_id {
let (pdu, pdu_json) = self.create_hash_and_sign_event(builder, sender, None)?;
let room_id = pdu.room_id().into_owned();
services().rooms.short.get_or_create_shortroomid(&room_id)?;
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(room_id.clone())
.or_default(),
);
(pdu, pdu_json, room_id, mutex_state)
} else {
let room_id = RoomId::new_v1(services().globals.server_name());
services().rooms.short.get_or_create_shortroomid(&room_id)?;
let mutex_state = Arc::clone(
services()
.globals
.roomid_mutex_state
.write()
.await
.entry(room_id.clone())
.or_default(),
);
let state_lock = mutex_state.lock().await;
let (pdu, pdu_json) =
self.create_hash_and_sign_event(builder, sender, Some((&room_id, &state_lock)))?;
drop(state_lock);
(pdu, pdu_json, room_id, mutex_state)
};
let state_lock = mutex_state.lock().await;
// We append to state before appending the pdu, so we don't have a moment in time with the
// pdu without it's state. This is okay because append_pdu can't fail.
let statehashid = services().rooms.state.append_to_state(&pdu)?;
self.append_pdu(
&pdu,
pdu_json,
// Since this PDU references all pdu_leaves we can update the leaves
// of the room
vec![(*pdu.event_id).to_owned()],
&state_lock,
)
.await?;
// 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
services()
.rooms
.state
.set_room_state(&room_id, statehashid, &state_lock)?;
drop(state_lock);
Ok((pdu.room_id().into_owned(), mutex_state))
}
/// 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 mutate the room state.
#[tracing::instrument(skip(self, state_lock))]
@ -870,10 +988,10 @@ impl Service {
state_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
) -> Result<Arc<EventId>> {
let (pdu, pdu_json) =
self.create_hash_and_sign_event(pdu_builder, sender, room_id, state_lock)?;
self.create_hash_and_sign_event(pdu_builder, sender, Some((room_id, state_lock)))?;
if let Some(admin_room) = services().admin.get_admin_room()? {
if admin_room == room_id {
if admin_room == *room_id {
match pdu.event_type() {
TimelineEventType::RoomEncryption => {
warn!("Encryption is not allowed in the admins room");
@ -956,7 +1074,7 @@ impl Service {
// If redaction event is not authorized, do not append it to the timeline
if pdu.kind == TimelineEventType::RoomRedaction {
let room_version_id = services().rooms.state.get_room_version(&pdu.room_id)?;
let room_version_id = services().rooms.state.get_room_version(&pdu.room_id())?;
let rules = room_version_id
.rules()
.expect("Supported room version must have rules.")
@ -970,7 +1088,7 @@ impl Service {
if !services().rooms.state_accessor.user_can_redact(
redact_id,
&pdu.sender,
&pdu.room_id,
&pdu.room_id(),
false,
)? {
return Err(Error::BadRequest(
@ -983,7 +1101,7 @@ impl Service {
if !services().rooms.state_accessor.user_can_redact(
redact_id,
&pdu.sender,
&pdu.room_id,
&pdu.room_id(),
false,
)? {
return Err(Error::BadRequest(
@ -1058,7 +1176,7 @@ impl Service {
// pdu without it's state. This is okay because append_pdu can't fail.
services().rooms.state.set_event_state(
&pdu.event_id,
&pdu.room_id,
&pdu.room_id(),
state_ids_compressed,
)?;
@ -1066,9 +1184,9 @@ impl Service {
services()
.rooms
.pdu_metadata
.mark_as_referenced(&pdu.room_id, &pdu.prev_events)?;
.mark_as_referenced(&pdu.room_id(), &pdu.prev_events)?;
services().rooms.state.set_forward_extremities(
&pdu.room_id,
&pdu.room_id(),
new_room_leaves,
state_lock,
)?;
@ -1143,7 +1261,7 @@ impl Service {
.deindex_pdu(shortroomid, &pdu_id, &content.body)?;
}
let room_version_id = services().rooms.state.get_room_version(&pdu.room_id)?;
let room_version_id = services().rooms.state.get_room_version(&pdu.room_id())?;
pdu.redact(
room_version_id
.rules()

2
src/service/sending/mod.rs
View file

@ -585,7 +585,7 @@ impl Service {
let unread: UInt = services()
.rooms
.user
.notification_count(userid, &pdu.room_id)
.notification_count(userid, &pdu.room_id())
.map_err(|e| (kind.clone(), e))?
.try_into()
.expect("notification count can't go that high");