Merge branch 'fixrestrictedjoin' into 'next'
fix: rejoining restricted rooms over federation See merge request famedly/conduit!431
This commit is contained in:
commit
5894d35eb2
4 changed files with 352 additions and 94 deletions
|
@ -12,12 +12,15 @@ use ruma::{
|
|||
},
|
||||
canonical_json::to_canonical_value,
|
||||
events::{
|
||||
room::member::{MembershipState, RoomMemberEventContent},
|
||||
room::{
|
||||
join_rules::{JoinRule, RoomJoinRulesEventContent},
|
||||
member::{MembershipState, RoomMemberEventContent},
|
||||
},
|
||||
RoomEventType, StateEventType,
|
||||
},
|
||||
serde::Base64,
|
||||
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, OwnedServerName,
|
||||
OwnedUserId, RoomId, RoomVersionId, UserId,
|
||||
state_res, CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
|
||||
OwnedServerName, OwnedUserId, RoomId, RoomVersionId, UserId,
|
||||
};
|
||||
use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
|
||||
use std::{
|
||||
|
@ -480,33 +483,10 @@ async fn join_room_by_id_helper(
|
|||
.state_cache
|
||||
.server_in_room(services().globals.server_name(), room_id)?
|
||||
{
|
||||
let mut make_join_response_and_server = Err(Error::BadServerResponse(
|
||||
"No server available to assist in joining.",
|
||||
));
|
||||
let (make_join_response, remote_server) =
|
||||
make_join_request(sender_user, room_id, servers).await?;
|
||||
|
||||
for remote_server in servers {
|
||||
let make_join_response = services()
|
||||
.sending
|
||||
.send_federation_request(
|
||||
remote_server,
|
||||
federation::membership::prepare_join_event::v1::Request {
|
||||
room_id,
|
||||
user_id: sender_user,
|
||||
ver: &services().globals.supported_room_versions(),
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
make_join_response_and_server = make_join_response.map(|r| (r, remote_server));
|
||||
|
||||
if make_join_response_and_server.is_ok() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
let (make_join_response, remote_server) = make_join_response_and_server?;
|
||||
|
||||
let room_version = match make_join_response.room_version {
|
||||
let room_version_id = match make_join_response.room_version {
|
||||
Some(room_version)
|
||||
if services()
|
||||
.globals
|
||||
|
@ -568,14 +548,14 @@ async fn join_room_by_id_helper(
|
|||
services().globals.server_name().as_str(),
|
||||
services().globals.keypair(),
|
||||
&mut join_event_stub,
|
||||
&room_version,
|
||||
&room_version_id,
|
||||
)
|
||||
.expect("event is valid, we just created it");
|
||||
|
||||
// Generate event id
|
||||
let event_id = format!(
|
||||
"${}",
|
||||
ruma::signatures::reference_hash(&join_event_stub, &room_version)
|
||||
ruma::signatures::reference_hash(&join_event_stub, &room_version_id)
|
||||
.expect("ruma can calculate reference hashes")
|
||||
);
|
||||
let event_id = <&EventId>::try_from(event_id.as_str())
|
||||
|
@ -588,12 +568,12 @@ async fn join_room_by_id_helper(
|
|||
);
|
||||
|
||||
// It has enough fields to be called a proper event now
|
||||
let join_event = join_event_stub;
|
||||
let mut join_event = join_event_stub;
|
||||
|
||||
let send_join_response = services()
|
||||
.sending
|
||||
.send_federation_request(
|
||||
remote_server,
|
||||
&remote_server,
|
||||
federation::membership::create_join_event::v2::Request {
|
||||
room_id,
|
||||
event_id,
|
||||
|
@ -602,9 +582,53 @@ async fn join_room_by_id_helper(
|
|||
)
|
||||
.await?;
|
||||
|
||||
if let Some(signed_raw) = &send_join_response.room_state.event {
|
||||
let (signed_event_id, signed_value) =
|
||||
match gen_event_id_canonical_json(&signed_raw, &room_version_id) {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
// Event could not be converted to canonical json
|
||||
return Err(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Could not convert event to canonical json.",
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
if &signed_event_id != event_id {
|
||||
return Err(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Server sent event with wrong event id",
|
||||
));
|
||||
}
|
||||
|
||||
if let Ok(signature) = signed_value["signatures"]
|
||||
.as_object()
|
||||
.ok_or(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Server sent invalid signatures type",
|
||||
))
|
||||
.and_then(|e| {
|
||||
e.get(remote_server.as_str()).ok_or(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Server did not send its signature",
|
||||
))
|
||||
})
|
||||
{
|
||||
join_event
|
||||
.get_mut("signatures")
|
||||
.expect("we created a valid pdu")
|
||||
.as_object_mut()
|
||||
.expect("we created a valid pdu")
|
||||
.insert(remote_server.to_string(), signature.clone());
|
||||
} else {
|
||||
warn!("Server {} sent invalid sendjoin event", remote_server);
|
||||
}
|
||||
}
|
||||
|
||||
services().rooms.short.get_or_create_shortroomid(room_id)?;
|
||||
|
||||
let parsed_pdu = PduEvent::from_id_val(event_id, join_event.clone())
|
||||
let parsed_join_pdu = PduEvent::from_id_val(event_id, join_event.clone())
|
||||
.map_err(|_| Error::BadServerResponse("Invalid join event PDU."))?;
|
||||
|
||||
let mut state = HashMap::new();
|
||||
|
@ -613,14 +637,14 @@ async fn join_room_by_id_helper(
|
|||
services()
|
||||
.rooms
|
||||
.event_handler
|
||||
.fetch_join_signing_keys(&send_join_response, &room_version, &pub_key_map)
|
||||
.fetch_join_signing_keys(&send_join_response, &room_version_id, &pub_key_map)
|
||||
.await?;
|
||||
|
||||
for result in send_join_response
|
||||
.room_state
|
||||
.state
|
||||
.iter()
|
||||
.map(|pdu| validate_and_add_event_id(pdu, &room_version, &pub_key_map))
|
||||
.map(|pdu| validate_and_add_event_id(pdu, &room_version_id, &pub_key_map))
|
||||
{
|
||||
let (event_id, value) = match result {
|
||||
Ok(t) => t,
|
||||
|
@ -645,31 +669,11 @@ async fn join_room_by_id_helper(
|
|||
}
|
||||
}
|
||||
|
||||
let incoming_shortstatekey = services().rooms.short.get_or_create_shortstatekey(
|
||||
&parsed_pdu.kind.to_string().into(),
|
||||
parsed_pdu
|
||||
.state_key
|
||||
.as_ref()
|
||||
.expect("Pdu is a membership state event"),
|
||||
)?;
|
||||
|
||||
state.insert(incoming_shortstatekey, parsed_pdu.event_id.clone());
|
||||
|
||||
let create_shortstatekey = services()
|
||||
.rooms
|
||||
.short
|
||||
.get_shortstatekey(&StateEventType::RoomCreate, "")?
|
||||
.expect("Room exists");
|
||||
|
||||
if state.get(&create_shortstatekey).is_none() {
|
||||
return Err(Error::BadServerResponse("State contained no create event."));
|
||||
}
|
||||
|
||||
for result in send_join_response
|
||||
.room_state
|
||||
.auth_chain
|
||||
.iter()
|
||||
.map(|pdu| validate_and_add_event_id(pdu, &room_version, &pub_key_map))
|
||||
.map(|pdu| validate_and_add_event_id(pdu, &room_version_id, &pub_key_map))
|
||||
{
|
||||
let (event_id, value) = match result {
|
||||
Ok(t) => t,
|
||||
|
@ -682,6 +686,34 @@ async fn join_room_by_id_helper(
|
|||
.add_pdu_outlier(&event_id, &value)?;
|
||||
}
|
||||
|
||||
if !state_res::event_auth::auth_check(
|
||||
&state_res::RoomVersion::new(&room_version_id).expect("room version is supported"),
|
||||
&parsed_join_pdu,
|
||||
None::<PduEvent>, // TODO: third party invite
|
||||
|k, s| {
|
||||
services()
|
||||
.rooms
|
||||
.timeline
|
||||
.get_pdu(
|
||||
state.get(
|
||||
&services()
|
||||
.rooms
|
||||
.short
|
||||
.get_or_create_shortstatekey(&k.to_string().into(), s)
|
||||
.ok()?,
|
||||
)?,
|
||||
)
|
||||
.ok()?
|
||||
},
|
||||
)
|
||||
.map_err(|_e| Error::BadRequest(ErrorKind::InvalidParam, "Auth check failed"))?
|
||||
{
|
||||
return Err(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Auth check failed",
|
||||
));
|
||||
}
|
||||
|
||||
let (statehash_before_join, new, removed) = services().rooms.state_compressor.save_state(
|
||||
room_id,
|
||||
state
|
||||
|
@ -705,12 +737,12 @@ async fn join_room_by_id_helper(
|
|||
|
||||
// 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 statehash_after_join = services().rooms.state.append_to_state(&parsed_pdu)?;
|
||||
let statehash_after_join = services().rooms.state.append_to_state(&parsed_join_pdu)?;
|
||||
|
||||
services().rooms.timeline.append_pdu(
|
||||
&parsed_pdu,
|
||||
&parsed_join_pdu,
|
||||
join_event,
|
||||
vec![(*parsed_pdu.event_id).to_owned()],
|
||||
vec![(*parsed_join_pdu.event_id).to_owned()],
|
||||
&state_lock,
|
||||
)?;
|
||||
|
||||
|
@ -732,7 +764,8 @@ async fn join_room_by_id_helper(
|
|||
join_authorized_via_users_server: None,
|
||||
};
|
||||
|
||||
services().rooms.timeline.build_and_append_pdu(
|
||||
// Try normal join first
|
||||
let error = match services().rooms.timeline.build_and_append_pdu(
|
||||
PduBuilder {
|
||||
event_type: RoomEventType::RoomMember,
|
||||
content: to_raw_value(&event).expect("event is valid, we just created it"),
|
||||
|
@ -743,14 +776,216 @@ async fn join_room_by_id_helper(
|
|||
sender_user,
|
||||
room_id,
|
||||
&state_lock,
|
||||
) {
|
||||
Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())),
|
||||
Err(e) => e,
|
||||
};
|
||||
|
||||
// TODO: Conduit does not implement restricted join rules yet, we always ask over
|
||||
// federation
|
||||
let join_rules_event = services().rooms.state_accessor.room_state_get(
|
||||
&room_id,
|
||||
&StateEventType::RoomJoinRules,
|
||||
"",
|
||||
)?;
|
||||
|
||||
let join_rules_event_content: Option<RoomJoinRulesEventContent> = join_rules_event
|
||||
.as_ref()
|
||||
.map(|join_rules_event| {
|
||||
serde_json::from_str(join_rules_event.content.get()).map_err(|e| {
|
||||
warn!("Invalid join rules event: {}", e);
|
||||
Error::bad_database("Invalid join rules event in db.")
|
||||
})
|
||||
})
|
||||
.transpose()?;
|
||||
|
||||
if matches!(
|
||||
join_rules_event_content,
|
||||
Some(RoomJoinRulesEventContent {
|
||||
join_rule: JoinRule::Restricted { .. }
|
||||
}) | Some(RoomJoinRulesEventContent {
|
||||
join_rule: JoinRule::KnockRestricted { .. }
|
||||
})
|
||||
) {
|
||||
let (make_join_response, remote_server) =
|
||||
make_join_request(sender_user, room_id, servers).await?;
|
||||
|
||||
let room_version_id = match make_join_response.room_version {
|
||||
Some(room_version_id)
|
||||
if services()
|
||||
.globals
|
||||
.supported_room_versions()
|
||||
.contains(&room_version_id) =>
|
||||
{
|
||||
room_version_id
|
||||
}
|
||||
_ => return Err(Error::BadServerResponse("Room version is not supported")),
|
||||
};
|
||||
let mut join_event_stub: CanonicalJsonObject =
|
||||
serde_json::from_str(make_join_response.event.get()).map_err(|_| {
|
||||
Error::BadServerResponse("Invalid make_join event json received from server.")
|
||||
})?;
|
||||
let join_authorized_via_users_server = join_event_stub
|
||||
.get("content")
|
||||
.map(|s| {
|
||||
s.as_object()?
|
||||
.get("join_authorised_via_users_server")?
|
||||
.as_str()
|
||||
})
|
||||
.and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok());
|
||||
// TODO: Is origin needed?
|
||||
join_event_stub.insert(
|
||||
"origin".to_owned(),
|
||||
CanonicalJsonValue::String(services().globals.server_name().as_str().to_owned()),
|
||||
);
|
||||
join_event_stub.insert(
|
||||
"origin_server_ts".to_owned(),
|
||||
CanonicalJsonValue::Integer(
|
||||
utils::millis_since_unix_epoch()
|
||||
.try_into()
|
||||
.expect("Timestamp is valid js_int value"),
|
||||
),
|
||||
);
|
||||
join_event_stub.insert(
|
||||
"content".to_owned(),
|
||||
to_canonical_value(RoomMemberEventContent {
|
||||
membership: MembershipState::Join,
|
||||
displayname: services().users.displayname(sender_user)?,
|
||||
avatar_url: services().users.avatar_url(sender_user)?,
|
||||
is_direct: None,
|
||||
third_party_invite: None,
|
||||
blurhash: services().users.blurhash(sender_user)?,
|
||||
reason: None,
|
||||
join_authorized_via_users_server,
|
||||
})
|
||||
.expect("event is valid, we just created it"),
|
||||
);
|
||||
|
||||
// We don't leave the event id in the pdu because that's only allowed in v1 or v2 rooms
|
||||
join_event_stub.remove("event_id");
|
||||
|
||||
// In order to create a compatible ref hash (EventID) the `hashes` field needs to be present
|
||||
ruma::signatures::hash_and_sign_event(
|
||||
services().globals.server_name().as_str(),
|
||||
services().globals.keypair(),
|
||||
&mut join_event_stub,
|
||||
&room_version_id,
|
||||
)
|
||||
.expect("event is valid, we just created it");
|
||||
|
||||
// Generate event id
|
||||
let event_id = format!(
|
||||
"${}",
|
||||
ruma::signatures::reference_hash(&join_event_stub, &room_version_id)
|
||||
.expect("ruma can calculate reference hashes")
|
||||
);
|
||||
let event_id = <&EventId>::try_from(event_id.as_str())
|
||||
.expect("ruma's reference hashes are valid event ids");
|
||||
|
||||
// Add event_id back
|
||||
join_event_stub.insert(
|
||||
"event_id".to_owned(),
|
||||
CanonicalJsonValue::String(event_id.as_str().to_owned()),
|
||||
);
|
||||
|
||||
// It has enough fields to be called a proper event now
|
||||
let join_event = join_event_stub;
|
||||
|
||||
let send_join_response = services()
|
||||
.sending
|
||||
.send_federation_request(
|
||||
&remote_server,
|
||||
federation::membership::create_join_event::v2::Request {
|
||||
room_id,
|
||||
event_id,
|
||||
pdu: &PduEvent::convert_to_outgoing_federation_event(join_event.clone()),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
if let Some(signed_raw) = send_join_response.room_state.event {
|
||||
let (signed_event_id, signed_value) =
|
||||
match gen_event_id_canonical_json(&signed_raw, &room_version_id) {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
// Event could not be converted to canonical json
|
||||
return Err(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Could not convert event to canonical json.",
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
if &signed_event_id != event_id {
|
||||
return Err(Error::BadRequest(
|
||||
ErrorKind::InvalidParam,
|
||||
"Server sent event with wrong event id",
|
||||
));
|
||||
}
|
||||
|
||||
drop(state_lock);
|
||||
let pub_key_map = RwLock::new(BTreeMap::new());
|
||||
services()
|
||||
.rooms
|
||||
.event_handler
|
||||
.handle_incoming_pdu(
|
||||
&remote_server,
|
||||
&signed_event_id,
|
||||
room_id,
|
||||
signed_value,
|
||||
true,
|
||||
&pub_key_map,
|
||||
)
|
||||
.await?;
|
||||
} else {
|
||||
return Err(error);
|
||||
}
|
||||
} else {
|
||||
return Err(error);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(join_room_by_id::v3::Response::new(room_id.to_owned()))
|
||||
}
|
||||
|
||||
async fn make_join_request(
|
||||
sender_user: &UserId,
|
||||
room_id: &RoomId,
|
||||
servers: &[OwnedServerName],
|
||||
) -> Result<(
|
||||
federation::membership::prepare_join_event::v1::Response,
|
||||
OwnedServerName,
|
||||
)> {
|
||||
let mut make_join_response_and_server = Err(Error::BadServerResponse(
|
||||
"No server available to assist in joining.",
|
||||
));
|
||||
|
||||
for remote_server in servers {
|
||||
if remote_server == services().globals.server_name() {
|
||||
continue;
|
||||
}
|
||||
let make_join_response = services()
|
||||
.sending
|
||||
.send_federation_request(
|
||||
remote_server,
|
||||
federation::membership::prepare_join_event::v1::Request {
|
||||
room_id,
|
||||
user_id: sender_user,
|
||||
ver: &services().globals.supported_room_versions(),
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
make_join_response_and_server = make_join_response.map(|r| (r, remote_server.clone()));
|
||||
|
||||
if make_join_response_and_server.is_ok() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
make_join_response_and_server
|
||||
}
|
||||
|
||||
fn validate_and_add_event_id(
|
||||
pdu: &RawJsonValue,
|
||||
room_version: &RoomVersionId,
|
||||
|
@ -870,6 +1105,8 @@ pub(crate) async fn invite_helper<'a>(
|
|||
(pdu, pdu_json, invite_room_state)
|
||||
};
|
||||
|
||||
let room_version_id = &services().rooms.state.get_room_version(room_id)?;
|
||||
|
||||
let response = services()
|
||||
.sending
|
||||
.send_federation_request(
|
||||
|
@ -877,7 +1114,7 @@ pub(crate) async fn invite_helper<'a>(
|
|||
create_invite::v2::Request {
|
||||
room_id,
|
||||
event_id: &pdu.event_id,
|
||||
room_version: &services().rooms.state.get_room_version(room_id)?,
|
||||
room_version: &room_version_id,
|
||||
event: &PduEvent::convert_to_outgoing_federation_event(pdu_json.clone()),
|
||||
invite_room_state: &invite_room_state,
|
||||
},
|
||||
|
@ -887,7 +1124,8 @@ pub(crate) async fn invite_helper<'a>(
|
|||
let pub_key_map = RwLock::new(BTreeMap::new());
|
||||
|
||||
// We do not add the event_id field to the pdu here because of signature and hashes checks
|
||||
let (event_id, value) = match gen_event_id_canonical_json(&response.event) {
|
||||
let (event_id, value) = match gen_event_id_canonical_json(&response.event, room_version_id)
|
||||
{
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
// Event could not be converted to canonical json
|
||||
|
|
|
@ -42,8 +42,8 @@ use ruma::{
|
|||
},
|
||||
serde::{Base64, JsonObject, Raw},
|
||||
to_device::DeviceIdOrAllDevices,
|
||||
CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId,
|
||||
OwnedServerName, OwnedServerSigningKeyId, OwnedUserId, RoomId, ServerName,
|
||||
CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId,
|
||||
OwnedRoomId, OwnedServerName, OwnedServerSigningKeyId, OwnedUserId, RoomId, ServerName,
|
||||
};
|
||||
use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
|
||||
use std::{
|
||||
|
@ -664,16 +664,11 @@ pub async fn send_transaction_message_route(
|
|||
// let mut auth_cache = EventMap::new();
|
||||
|
||||
for pdu in &body.pdus {
|
||||
// We do not add the event_id field to the pdu here because of signature and hashes checks
|
||||
let (event_id, value) = match gen_event_id_canonical_json(pdu) {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
// Event could not be converted to canonical json
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
|
||||
warn!("Error parsing incoming event {:?}: {:?}", pdu, e);
|
||||
Error::BadServerResponse("Invalid PDU in server response")
|
||||
})?;
|
||||
|
||||
// 0. Check the server is in the room
|
||||
let room_id: OwnedRoomId = match value
|
||||
.get("room_id")
|
||||
.and_then(|id| RoomId::parse(id.as_str()?).ok())
|
||||
|
@ -681,14 +676,26 @@ pub async fn send_transaction_message_route(
|
|||
Some(id) => id,
|
||||
None => {
|
||||
// Event is invalid
|
||||
resolved_map.insert(
|
||||
event_id,
|
||||
Err(Error::bad_database("Event needs a valid RoomId.")),
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let room_version_id = match services().rooms.state.get_room_version(&room_id) {
|
||||
Ok(v) => v,
|
||||
Err(_) => {
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let (event_id, value) = match gen_event_id_canonical_json(pdu, &room_version_id) {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
// Event could not be converted to canonical json
|
||||
continue;
|
||||
}
|
||||
};
|
||||
// We do not add the event_id field to the pdu here because of signature and hashes checks
|
||||
|
||||
services()
|
||||
.rooms
|
||||
.event_handler
|
||||
|
@ -1407,7 +1414,8 @@ async fn create_join_event(
|
|||
// let mut auth_cache = EventMap::new();
|
||||
|
||||
// We do not add the event_id field to the pdu here because of signature and hashes checks
|
||||
let (event_id, value) = match gen_event_id_canonical_json(pdu) {
|
||||
let room_version_id = services().rooms.state.get_room_version(room_id)?;
|
||||
let (event_id, value) = match gen_event_id_canonical_json(pdu, &room_version_id) {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
// Event could not be converted to canonical json
|
||||
|
@ -1610,8 +1618,12 @@ pub async fn create_invite_route(
|
|||
|
||||
invite_state.push(pdu.to_stripped_state_event());
|
||||
|
||||
// If the room already exists, the remote server will notify us about the join via /send
|
||||
if !services().rooms.metadata.exists(&pdu.room_id)? {
|
||||
// If we are active in the room, the remote server will notify us about the join via /send
|
||||
if !services()
|
||||
.rooms
|
||||
.state_cache
|
||||
.server_in_room(services().globals.server_name(), &body.room_id)?
|
||||
{
|
||||
services().rooms.state_cache.update_membership(
|
||||
&body.room_id,
|
||||
&invited_user,
|
||||
|
|
|
@ -1,4 +1,4 @@
|
|||
use crate::{services, Error};
|
||||
use crate::Error;
|
||||
use ruma::{
|
||||
events::{
|
||||
room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyStateEvent,
|
||||
|
@ -7,7 +7,7 @@ use ruma::{
|
|||
},
|
||||
serde::Raw,
|
||||
state_res, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
|
||||
OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UInt, UserId,
|
||||
OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, RoomVersionId, UInt, UserId,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{
|
||||
|
@ -334,23 +334,17 @@ impl Ord for PduEvent {
|
|||
/// Returns a tuple of the new `EventId` and the PDU as a `BTreeMap<String, CanonicalJsonValue>`.
|
||||
pub(crate) fn gen_event_id_canonical_json(
|
||||
pdu: &RawJsonValue,
|
||||
room_version_id: &RoomVersionId,
|
||||
) -> crate::Result<(OwnedEventId, CanonicalJsonObject)> {
|
||||
let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
|
||||
warn!("Error parsing incoming event {:?}: {:?}", pdu, e);
|
||||
Error::BadServerResponse("Invalid PDU in server response")
|
||||
})?;
|
||||
|
||||
let room_id = value
|
||||
.get("room_id")
|
||||
.and_then(|id| RoomId::parse(id.as_str()?).ok())
|
||||
.ok_or_else(|| Error::bad_database("PDU in db has invalid room_id."))?;
|
||||
|
||||
let room_version_id = services().rooms.state.get_room_version(&room_id);
|
||||
|
||||
let event_id = format!(
|
||||
"${}",
|
||||
// Anything higher than version3 behaves the same
|
||||
ruma::signatures::reference_hash(&value, &room_version_id?)
|
||||
ruma::signatures::reference_hash(&value, room_version_id)
|
||||
.expect("ruma can calculate reference hashes")
|
||||
)
|
||||
.try_into()
|
||||
|
|
|
@ -76,6 +76,7 @@ impl Service {
|
|||
is_timeline_event: bool,
|
||||
pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
|
||||
) -> Result<Option<Vec<u8>>> {
|
||||
// 0. Check the server is in the room
|
||||
if !services().rooms.metadata.exists(room_id)? {
|
||||
return Err(Error::BadRequest(
|
||||
ErrorKind::NotFound,
|
||||
|
@ -101,6 +102,13 @@ impl Service {
|
|||
.room_state_get(room_id, &StateEventType::RoomCreate, "")?
|
||||
.ok_or_else(|| Error::bad_database("Failed to find create event in db."))?;
|
||||
|
||||
let create_event_content: RoomCreateEventContent =
|
||||
serde_json::from_str(create_event.content.get()).map_err(|e| {
|
||||
error!("Invalid create event: {}", e);
|
||||
Error::BadDatabase("Invalid create event in db")
|
||||
})?;
|
||||
let room_version_id = &create_event_content.room_version;
|
||||
|
||||
let first_pdu_in_room = services()
|
||||
.rooms
|
||||
.timeline
|
||||
|
@ -127,6 +135,7 @@ impl Service {
|
|||
origin,
|
||||
&create_event,
|
||||
room_id,
|
||||
room_version_id,
|
||||
pub_key_map,
|
||||
incoming_pdu.prev_events.clone(),
|
||||
)
|
||||
|
@ -341,6 +350,7 @@ impl Service {
|
|||
.collect::<Vec<_>>(),
|
||||
create_event,
|
||||
room_id,
|
||||
room_version_id,
|
||||
pub_key_map,
|
||||
)
|
||||
.await;
|
||||
|
@ -645,6 +655,7 @@ impl Service {
|
|||
.collect::<Vec<_>>(),
|
||||
create_event,
|
||||
room_id,
|
||||
room_version_id,
|
||||
pub_key_map,
|
||||
)
|
||||
.await;
|
||||
|
@ -1025,6 +1036,7 @@ impl Service {
|
|||
events: &'a [Arc<EventId>],
|
||||
create_event: &'a PduEvent,
|
||||
room_id: &'a RoomId,
|
||||
room_version_id: &'a RoomVersionId,
|
||||
pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
|
||||
) -> AsyncRecursiveType<'a, Vec<(Arc<PduEvent>, Option<BTreeMap<String, CanonicalJsonValue>>)>>
|
||||
{
|
||||
|
@ -1107,7 +1119,7 @@ impl Service {
|
|||
Ok(res) => {
|
||||
info!("Got {} over federation", next_id);
|
||||
let (calculated_event_id, value) =
|
||||
match pdu::gen_event_id_canonical_json(&res.pdu) {
|
||||
match pdu::gen_event_id_canonical_json(&res.pdu, room_version_id) {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
back_off((*next_id).to_owned());
|
||||
|
@ -1180,6 +1192,7 @@ impl Service {
|
|||
origin: &ServerName,
|
||||
create_event: &PduEvent,
|
||||
room_id: &RoomId,
|
||||
room_version_id: &RoomVersionId,
|
||||
pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
|
||||
initial_set: Vec<Arc<EventId>>,
|
||||
) -> Result<(
|
||||
|
@ -1205,6 +1218,7 @@ impl Service {
|
|||
&[prev_event_id.clone()],
|
||||
create_event,
|
||||
room_id,
|
||||
room_version_id,
|
||||
pub_key_map,
|
||||
)
|
||||
.await
|
||||
|
|
Loading…
Add table
Reference in a new issue