fix: Do not allow fetching cached remote users' profiles over federation

feat: Keep track of avatar urls, displaynames, and blurhashes of remote users for the room directory
2022-12-21 17:46:01 +01:00 · 2022-12-21 17:28:21 +01:00
14 changed files with 915 additions and 680 deletions
--- a/src/api/client_server/membership.rs
+++ b/src/api/client_server/membership.rs
@ -198,7 +198,10 @@ pub async fn kick_user_route(
    );
    let state_lock = mutex_state.lock().await;

-    services().rooms.timeline.build_and_append_pdu(
+    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"),
@ -209,7 +212,8 @@ pub async fn kick_user_route(
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    drop(state_lock);

@ -264,7 +268,10 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
    );
    let state_lock = mutex_state.lock().await;

-    services().rooms.timeline.build_and_append_pdu(
+    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"),
@ -275,7 +282,8 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    drop(state_lock);

@ -321,7 +329,10 @@ pub async fn unban_user_route(
    );
    let state_lock = mutex_state.lock().await;

-    services().rooms.timeline.build_and_append_pdu(
+    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"),
@ -332,7 +343,8 @@ pub async fn unban_user_route(
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    drop(state_lock);

@ -738,12 +750,16 @@ async fn join_room_by_id_helper(
        // 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_join_pdu)?;

-        services().rooms.timeline.append_pdu(
+        services()
+            .rooms
+            .timeline
+            .append_pdu(
                &parsed_join_pdu,
                join_event,
                vec![(*parsed_join_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
@ -853,7 +869,10 @@ async fn join_room_by_id_helper(
        };

        // Try normal join first
-        let error = match services().rooms.timeline.build_and_append_pdu(
+        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"),
@ -864,7 +883,9 @@ async fn join_room_by_id_helper(
                sender_user,
                room_id,
                &state_lock,
-        ) {
+            )
+            .await
+        {
            Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())),
            Err(e) => e,
        };
@ -1259,7 +1280,10 @@ pub(crate) async fn invite_helper<'a>(
    );
    let state_lock = mutex_state.lock().await;

-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomMember,
                content: to_raw_value(&RoomMemberEventContent {
@ -1280,7 +1304,8 @@ pub(crate) async fn invite_helper<'a>(
            sender_user,
            room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    drop(state_lock);

@ -1334,14 +1359,18 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
            )?;

        // We always drop the invite, we can't rely on other servers
-        services().rooms.state_cache.update_membership(
+        services()
+            .rooms
+            .state_cache
+            .update_membership(
                room_id,
                user_id,
-            MembershipState::Leave,
+                RoomMemberEventContent::new(MembershipState::Leave),
                user_id,
                last_state,
                true,
-        )?;
+            )
+            .await?;
    } else {
        let mutex_state = Arc::clone(
            services()
@ -1365,14 +1394,18 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
            None => {
                error!("Trying to leave a room you are not a member of.");

-                services().rooms.state_cache.update_membership(
+                services()
+                    .rooms
+                    .state_cache
+                    .update_membership(
                        room_id,
                        user_id,
-                    MembershipState::Leave,
+                        RoomMemberEventContent::new(MembershipState::Leave),
                        user_id,
                        None,
                        true,
-                )?;
+                    )
+                    .await?;
                return Ok(());
            }
            Some(e) => e,
@ -1383,7 +1416,10 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {

        event.membership = MembershipState::Leave;

-        services().rooms.timeline.build_and_append_pdu(
+        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"),
@ -1394,7 +1430,8 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
                user_id,
                room_id,
                &state_lock,
-        )?;
+            )
+            .await?;
    }

    Ok(())
--- a/src/api/client_server/message.rs
+++ b/src/api/client_server/message.rs
@ -70,7 +70,10 @@ pub async fn send_message_event_route(
    let mut unsigned = BTreeMap::new();
    unsigned.insert("transaction_id".to_owned(), body.txn_id.to_string().into());

-    let event_id = services().rooms.timeline.build_and_append_pdu(
+    let event_id = services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: body.event_type.to_string().into(),
                content: serde_json::from_str(body.body.body.json().get())
@ -82,7 +85,8 @@ pub async fn send_message_event_route(
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    services().transaction_ids.add_txnid(
        sender_user,
--- a/src/api/client_server/profile.rs
+++ b/src/api/client_server/profile.rs
@ -83,12 +83,11 @@ pub async fn set_displayname_route(
        );
        let state_lock = mutex_state.lock().await;

-        let _ = services().rooms.timeline.build_and_append_pdu(
-            pdu_builder,
-            sender_user,
-            &room_id,
-            &state_lock,
-        );
+        let _ = services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
+            .await;

        // Presence update
        services().rooms.edus.presence.update_presence(
@ -115,15 +114,17 @@ pub async fn set_displayname_route(
    Ok(set_display_name::v3::Response {})
 }

-/// # `GET /_matrix/client/r0/profile/{userId}/displayname`
+/// # `GET /_matrix/client/v3/profile/{userId}/displayname`
 ///
 /// Returns the displayname of the user.
 ///
-/// - If user is on another server: Fetches displayname over federation
+/// - If user is on another server and we do not have a copy, fetch over federation
 pub async fn get_displayname_route(
    body: Ruma<get_display_name::v3::Request>,
 ) -> Result<get_display_name::v3::Response> {
-    if body.user_id.server_name() != services().globals.server_name() {
+    if (services().users.exists(&body.user_id)?)
+        && (body.user_id.server_name() != services().globals.server_name())
+    {
        let response = services()
            .sending
            .send_federation_request(
@ -135,6 +136,18 @@ pub async fn get_displayname_route(
            )
            .await?;

+        // Create and update our local copy of the user
+        let _ = services().users.create(&body.user_id, None);
+        let _ = services()
+            .users
+            .set_displayname(&body.user_id, response.displayname.clone());
+        let _ = services()
+            .users
+            .set_avatar_url(&body.user_id, response.avatar_url);
+        let _ = services()
+            .users
+            .set_blurhash(&body.user_id, response.blurhash);
+
        return Ok(get_display_name::v3::Response {
            displayname: response.displayname,
        });
@ -218,12 +231,11 @@ pub async fn set_avatar_url_route(
        );
        let state_lock = mutex_state.lock().await;

-        let _ = services().rooms.timeline.build_and_append_pdu(
-            pdu_builder,
-            sender_user,
-            &room_id,
-            &state_lock,
-        );
+        let _ = services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
+            .await;

        // Presence update
        services().rooms.edus.presence.update_presence(
@ -250,15 +262,17 @@ pub async fn set_avatar_url_route(
    Ok(set_avatar_url::v3::Response {})
 }

-/// # `GET /_matrix/client/r0/profile/{userId}/avatar_url`
+/// # `GET /_matrix/client/v3/profile/{userId}/avatar_url`
 ///
 /// Returns the avatar_url and blurhash of the user.
 ///
-/// - If user is on another server: Fetches avatar_url and blurhash over federation
+/// - If user is on another server and we do not have a copy, fetch over federation
 pub async fn get_avatar_url_route(
    body: Ruma<get_avatar_url::v3::Request>,
 ) -> Result<get_avatar_url::v3::Response> {
-    if body.user_id.server_name() != services().globals.server_name() {
+    if (services().users.exists(&body.user_id)?)
+        && (body.user_id.server_name() != services().globals.server_name())
+    {
        let response = services()
            .sending
            .send_federation_request(
@ -270,6 +284,18 @@ pub async fn get_avatar_url_route(
            )
            .await?;

+        // Create and update our local copy of the user
+        let _ = services().users.create(&body.user_id, None);
+        let _ = services()
+            .users
+            .set_displayname(&body.user_id, response.displayname);
+        let _ = services()
+            .users
+            .set_avatar_url(&body.user_id, response.avatar_url.clone());
+        let _ = services()
+            .users
+            .set_blurhash(&body.user_id, response.blurhash.clone());
+
        return Ok(get_avatar_url::v3::Response {
            avatar_url: response.avatar_url,
            blurhash: response.blurhash,
@ -286,11 +312,13 @@ pub async fn get_avatar_url_route(
 ///
 /// Returns the displayname, avatar_url and blurhash of the user.
 ///
-/// - If user is on another server: Fetches profile over federation
+/// - If user is on another server and we do not have a copy, fetch over federation
 pub async fn get_profile_route(
    body: Ruma<get_profile::v3::Request>,
 ) -> Result<get_profile::v3::Response> {
-    if body.user_id.server_name() != services().globals.server_name() {
+    if (services().users.exists(&body.user_id)?)
+        && (body.user_id.server_name() != services().globals.server_name())
+    {
        let response = services()
            .sending
            .send_federation_request(
@ -302,6 +330,18 @@ pub async fn get_profile_route(
            )
            .await?;

+        // Create and update our local copy of the user
+        let _ = services().users.create(&body.user_id, None);
+        let _ = services()
+            .users
+            .set_displayname(&body.user_id, response.displayname.clone());
+        let _ = services()
+            .users
+            .set_avatar_url(&body.user_id, response.avatar_url.clone());
+        let _ = services()
+            .users
+            .set_blurhash(&body.user_id, response.blurhash.clone());
+
        return Ok(get_profile::v3::Response {
            displayname: response.displayname,
            avatar_url: response.avatar_url,
--- a/src/api/client_server/redact.rs
+++ b/src/api/client_server/redact.rs
@ -30,7 +30,10 @@ pub async fn redact_event_route(
    );
    let state_lock = mutex_state.lock().await;

-    let event_id = services().rooms.timeline.build_and_append_pdu(
+    let event_id = services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomRedaction,
                content: to_raw_value(&RoomRedactionEventContent {
@ -44,7 +47,8 @@ pub async fn redact_event_route(
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    drop(state_lock);

--- a/src/api/client_server/room.rs
+++ b/src/api/client_server/room.rs
@ -173,7 +173,10 @@ pub async fn create_room_route(
    }

    // 1. The room create event
-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomCreate,
                content: to_raw_value(&content).expect("event is valid, we just created it"),
@ -184,10 +187,14 @@ pub async fn create_room_route(
            sender_user,
            &room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // 2. Let the room creator join
-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomMember,
                content: to_raw_value(&RoomMemberEventContent {
@ -208,7 +215,8 @@ pub async fn create_room_route(
            sender_user,
            &room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // 3. Power levels

@ -245,7 +253,10 @@ pub async fn create_room_route(
        }
    }

-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomPowerLevels,
                content: to_raw_value(&power_levels_content)
@ -257,11 +268,15 @@ pub async fn create_room_route(
            sender_user,
            &room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // 4. Canonical room alias
    if let Some(room_alias_id) = &alias {
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomCanonicalAlias,
                    content: to_raw_value(&RoomCanonicalAliasEventContent {
@ -276,13 +291,17 @@ pub async fn create_room_route(
                sender_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;
    }

    // 5. Events set by preset

    // 5.1 Join Rules
-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomJoinRules,
                content: to_raw_value(&RoomJoinRulesEventContent::new(match preset {
@ -298,10 +317,14 @@ pub async fn create_room_route(
            sender_user,
            &room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // 5.2 History Visibility
-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomHistoryVisibility,
                content: to_raw_value(&RoomHistoryVisibilityEventContent::new(
@ -315,10 +338,14 @@ pub async fn create_room_route(
            sender_user,
            &room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // 5.3 Guest Access
-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomGuestAccess,
                content: to_raw_value(&RoomGuestAccessEventContent::new(match preset {
@ -333,7 +360,8 @@ pub async fn create_room_route(
            sender_user,
            &room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // 6. Events listed in initial_state
    for event in &body.initial_state {
@ -352,17 +380,19 @@ pub async fn create_room_route(
            continue;
        }

-        services().rooms.timeline.build_and_append_pdu(
-            pdu_builder,
-            sender_user,
-            &room_id,
-            &state_lock,
-        )?;
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
+            .await?;
    }

    // 7. Events implied by name and topic
    if let Some(name) = &body.name {
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomName,
                    content: to_raw_value(&RoomNameEventContent::new(Some(name.clone())))
@ -374,11 +404,15 @@ pub async fn create_room_route(
                sender_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;
    }

    if let Some(topic) = &body.topic {
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomTopic,
                    content: to_raw_value(&RoomTopicEventContent {
@ -392,7 +426,8 @@ pub async fn create_room_route(
                sender_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;
    }

    // 8. Events implied by invite (and TODO: invite_3pid)
@ -523,7 +558,10 @@ pub async fn upgrade_room_route(

    // 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(
+    let tombstone_event_id = services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomTombstone,
                content: to_raw_value(&RoomTombstoneEventContent {
@ -538,7 +576,8 @@ pub async fn upgrade_room_route(
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    // Change lock to replacement room
    drop(state_lock);
@ -605,7 +644,10 @@ pub async fn upgrade_room_route(
        ));
    }

-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomCreate,
                content: to_raw_value(&create_event_content)
@ -617,10 +659,14 @@ pub async fn upgrade_room_route(
            sender_user,
            &replacement_room,
            &state_lock,
-    )?;
+        )
+        .await?;

    // Join the new room
-    services().rooms.timeline.build_and_append_pdu(
+    services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomMember,
                content: to_raw_value(&RoomMemberEventContent {
@ -641,7 +687,8 @@ pub async fn upgrade_room_route(
            sender_user,
            &replacement_room,
            &state_lock,
-    )?;
+        )
+        .await?;

    // Recommended transferable state events list from the specs
    let transferable_state_events = vec![
@ -668,7 +715,10 @@ pub async fn upgrade_room_route(
                None => continue, // Skipping missing events.
            };

-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: event_type.to_string().into(),
                    content: event_content,
@ -679,7 +729,8 @@ pub async fn upgrade_room_route(
                sender_user,
                &replacement_room,
                &state_lock,
-        )?;
+            )
+            .await?;
    }

    // Moves any local aliases to the new room
@ -713,7 +764,10 @@ pub async fn upgrade_room_route(
    power_levels_event_content.invite = new_level;

    // Modify the power levels in the old room to prevent sending of events and inviting new users
-    let _ = services().rooms.timeline.build_and_append_pdu(
+    let _ = services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: RoomEventType::RoomPowerLevels,
                content: to_raw_value(&power_levels_event_content)
@ -725,7 +779,8 @@ pub async fn upgrade_room_route(
            sender_user,
            &body.room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    drop(state_lock);

--- a/src/api/client_server/state.rs
+++ b/src/api/client_server/state.rs
@ -287,7 +287,10 @@ async fn send_state_event_for_key_helper(
    );
    let state_lock = mutex_state.lock().await;

-    let event_id = services().rooms.timeline.build_and_append_pdu(
+    let event_id = services()
+        .rooms
+        .timeline
+        .build_and_append_pdu(
            PduBuilder {
                event_type: event_type.to_string().into(),
                content: serde_json::from_str(json.json().get()).expect("content is valid json"),
@ -298,7 +301,8 @@ async fn send_state_event_for_key_helper(
            sender_user,
            room_id,
            &state_lock,
-    )?;
+        )
+        .await?;

    Ok(event_id)
 }
--- a/src/api/client_server/sync.rs
+++ b/src/api/client_server/sync.rs
@ -231,7 +231,7 @@ async fn sync_helper(
                    .entry(room_id.clone())
                    .or_default(),
            );
-            let insert_lock = mutex_insert.lock().unwrap();
+            let insert_lock = mutex_insert.lock().await;
            drop(insert_lock);
        }

@ -847,7 +847,7 @@ async fn sync_helper(
                    .entry(room_id.clone())
                    .or_default(),
            );
-            let insert_lock = mutex_insert.lock().unwrap();
+            let insert_lock = mutex_insert.lock().await;
            drop(insert_lock);
        }

@ -979,7 +979,7 @@ async fn sync_helper(
                    .entry(room_id.clone())
                    .or_default(),
            );
-            let insert_lock = mutex_insert.lock().unwrap();
+            let insert_lock = mutex_insert.lock().await;
            drop(insert_lock);
        }

--- a/src/api/server_server.rs
+++ b/src/api/server_server.rs
@ -1615,14 +1615,18 @@ pub async fn create_invite_route(
        .state_cache
        .server_in_room(services().globals.server_name(), &body.room_id)?
    {
-        services().rooms.state_cache.update_membership(
+        services()
+            .rooms
+            .state_cache
+            .update_membership(
                &body.room_id,
                &invited_user,
-            MembershipState::Invite,
+                RoomMemberEventContent::new(MembershipState::Invite),
                &sender,
                Some(invite_state),
                true,
-        )?;
+            )
+            .await?;
    }

    Ok(create_invite::v2::Response {
@ -1712,6 +1716,13 @@ pub async fn get_profile_information_route(
        return Err(Error::bad_config("Federation is disabled."));
    }

+    if body.user_id.server_name() != services().globals.server_name() {
+        return Err(Error::BadRequest(
+            ErrorKind::NotFound,
+            "User does not belong to this server",
+        ));
+    }
+
    let mut displayname = None;
    let mut avatar_url = None;
    let mut blurhash = None;
--- a/src/service/admin/mod.rs
+++ b/src/service/admin/mod.rs
@ -26,7 +26,7 @@ use ruma::{
    EventId, OwnedRoomAliasId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
 };
 use serde_json::value::to_raw_value;
-use tokio::sync::{mpsc, Mutex, MutexGuard};
+use tokio::sync::{mpsc, Mutex};

 use crate::{
    api::client_server::{leave_all_rooms, AUTO_GEN_PASSWORD_LENGTH},
@ -206,26 +206,6 @@ impl Service {
            .expect("Database data for admin room alias must be valid")
            .expect("Admin room must exist");

-        let send_message = |message: RoomMessageEventContent, mutex_lock: &MutexGuard<'_, ()>| {
-            services()
-                .rooms
-                .timeline
-                .build_and_append_pdu(
-                    PduBuilder {
-                        event_type: RoomEventType::RoomMessage,
-                        content: to_raw_value(&message)
-                            .expect("event is valid, we just created it"),
-                        unsigned: None,
-                        state_key: None,
-                        redacts: None,
-                    },
-                    &conduit_user,
-                    &conduit_room,
-                    mutex_lock,
-                )
-                .unwrap();
-        };
-
        loop {
            tokio::select! {
                Some(event) = receiver.recv() => {
@ -245,7 +225,20 @@ impl Service {

                    let state_lock = mutex_state.lock().await;

-                    send_message(message_content, &state_lock);
+                    services().rooms.timeline.build_and_append_pdu(
+                      PduBuilder {
+                        event_type: RoomEventType::RoomMessage,
+                        content: to_raw_value(&message_content)
+                            .expect("event is valid, we just created it"),
+                        unsigned: None,
+                        state_key: None,
+                        redacts: None,
+                      },
+                      &conduit_user,
+                      &conduit_room,
+                      &state_lock)
+                    .await
+                    .unwrap();

                    drop(state_lock);
                }
@ -853,7 +846,10 @@ impl Service {
        content.room_version = services().globals.default_room_version();

        // 1. The room create event
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomCreate,
                    content: to_raw_value(&content).expect("event is valid, we just created it"),
@ -864,10 +860,14 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 2. Make conduit bot join
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomMember,
                    content: to_raw_value(&RoomMemberEventContent {
@ -888,13 +888,17 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 3. Power levels
        let mut users = BTreeMap::new();
        users.insert(conduit_user.clone(), 100.into());

-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomPowerLevels,
                    content: to_raw_value(&RoomPowerLevelsEventContent {
@ -909,10 +913,14 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 4.1 Join Rules
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomJoinRules,
                    content: to_raw_value(&RoomJoinRulesEventContent::new(JoinRule::Invite))
@ -924,10 +932,14 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 4.2 History Visibility
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomHistoryVisibility,
                    content: to_raw_value(&RoomHistoryVisibilityEventContent::new(
@ -941,13 +953,19 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 4.3 Guest Access
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomGuestAccess,
-                content: to_raw_value(&RoomGuestAccessEventContent::new(GuestAccess::Forbidden))
+                    content: to_raw_value(&RoomGuestAccessEventContent::new(
+                        GuestAccess::Forbidden,
+                    ))
                    .expect("event is valid, we just created it"),
                    unsigned: None,
                    state_key: Some("".to_owned()),
@ -956,11 +974,15 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 5. Events implied by name and topic
        let room_name = format!("{} Admin Room", services().globals.server_name());
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomName,
                    content: to_raw_value(&RoomNameEventContent::new(Some(room_name)))
@ -972,9 +994,13 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomTopic,
                    content: to_raw_value(&RoomTopicEventContent {
@ -988,14 +1014,18 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // 6. Room alias
        let alias: OwnedRoomAliasId = format!("#admins:{}", services().globals.server_name())
            .try_into()
            .expect("#admins:server_name is a valid alias name");

-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomCanonicalAlias,
                    content: to_raw_value(&RoomCanonicalAliasEventContent {
@ -1010,7 +1040,8 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        services().rooms.alias.set_alias(&alias, &room_id)?;

@ -1052,7 +1083,10 @@ impl Service {
                .expect("@conduit:server_name is valid");

        // Invite and join the real user
-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomMember,
                    content: to_raw_value(&RoomMemberEventContent {
@ -1073,8 +1107,12 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
-        services().rooms.timeline.build_and_append_pdu(
+            )
+            .await?;
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomMember,
                    content: to_raw_value(&RoomMemberEventContent {
@ -1095,14 +1133,18 @@ impl Service {
                user_id,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // Set power level
        let mut users = BTreeMap::new();
        users.insert(conduit_user.to_owned(), 100.into());
        users.insert(user_id.to_owned(), 100.into());

-        services().rooms.timeline.build_and_append_pdu(
+        services()
+            .rooms
+            .timeline
+            .build_and_append_pdu(
                PduBuilder {
                    event_type: RoomEventType::RoomPowerLevels,
                    content: to_raw_value(&RoomPowerLevelsEventContent {
@ -1117,7 +1159,8 @@ impl Service {
                &conduit_user,
                &room_id,
                &state_lock,
-        )?;
+            )
+            .await?;

        // Send welcome message
        services().rooms.timeline.build_and_append_pdu(
@ -1135,7 +1178,7 @@ impl Service {
            &conduit_user,
            &room_id,
            &state_lock,
-        )?;
+        ).await?;

        Ok(())
    }
--- a/src/service/globals/mod.rs
+++ b/src/service/globals/mod.rs
@ -52,7 +52,7 @@ pub struct Service {
    pub bad_signature_ratelimiter: Arc<RwLock<HashMap<Vec<String>, RateLimitState>>>,
    pub servername_ratelimiter: Arc<RwLock<HashMap<OwnedServerName, Arc<Semaphore>>>>,
    pub sync_receivers: RwLock<HashMap<(OwnedUserId, OwnedDeviceId), SyncHandle>>,
-    pub roomid_mutex_insert: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>,
+    pub roomid_mutex_insert: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>,
    pub roomid_mutex_state: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>,
    pub roomid_mutex_federation: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>, // this lock will be held longer
    pub roomid_federationhandletime: RwLock<HashMap<OwnedRoomId, (OwnedEventId, Instant)>>,
--- a/src/service/rooms/event_handler/mod.rs
+++ b/src/service/rooms/event_handler/mod.rs
@ -801,14 +801,18 @@ impl Service {
        .map_err(|_e| Error::BadRequest(ErrorKind::InvalidParam, "Auth check failed."))?;

        if soft_fail {
-            services().rooms.timeline.append_incoming_pdu(
+            services()
+                .rooms
+                .timeline
+                .append_incoming_pdu(
                    &incoming_pdu,
                    val,
                    extremities.iter().map(|e| (**e).to_owned()).collect(),
                    state_ids_compressed,
                    soft_fail,
                    &state_lock,
-            )?;
+                )
+                .await?;

            // Soft fail, we keep the event as an outlier but don't add it to the timeline
            warn!("Event was soft failed: {:?}", incoming_pdu);
@ -1004,14 +1008,18 @@ impl Service {
        // We use the `state_at_event` instead of `state_after` so we accurately
        // represent the state for this event.

-        let pdu_id = services().rooms.timeline.append_incoming_pdu(
+        let pdu_id = services()
+            .rooms
+            .timeline
+            .append_incoming_pdu(
                &incoming_pdu,
                val,
                extremities.iter().map(|e| (**e).to_owned()).collect(),
                state_ids_compressed,
                soft_fail,
                &state_lock,
-        )?;
+            )
+            .await?;

        info!("Appended incoming pdu");

--- a/src/service/rooms/state/mod.rs
+++ b/src/service/rooms/state/mod.rs
@ -7,14 +7,13 @@ use std::{
 pub use data::Data;
 use ruma::{
    events::{
-        room::{create::RoomCreateEventContent, member::MembershipState},
+        room::{create::RoomCreateEventContent, member::RoomMemberEventContent},
        AnyStrippedStateEvent, RoomEventType, StateEventType,
    },
    serde::Raw,
    state_res::{self, StateMap},
    EventId, OwnedEventId, RoomId, RoomVersionId, UserId,
 };
-use serde::Deserialize;
 use tokio::sync::MutexGuard;
 use tracing::warn;

@ -60,13 +59,9 @@ impl Service {
                Err(_) => continue,
            };

-            #[derive(Deserialize)]
-            struct ExtractMembership {
-                membership: MembershipState,
-            }
-
-            let membership = match serde_json::from_str::<ExtractMembership>(pdu.content.get()) {
-                Ok(e) => e.membership,
+            let membership_event =
+                match serde_json::from_str::<RoomMemberEventContent>(pdu.content.get()) {
+                    Ok(e) => e,
                    Err(_) => continue,
                };

@ -80,14 +75,18 @@ impl Service {
                Err(_) => continue,
            };

-            services().rooms.state_cache.update_membership(
+            services()
+                .rooms
+                .state_cache
+                .update_membership(
                    room_id,
                    &user_id,
-                membership,
+                    membership_event,
                    &pdu.sender,
                    None,
                    false,
-            )?;
+                )
+                .await?;
        }

        services().rooms.state_cache.update_joined_count(room_id)?;
--- a/src/service/rooms/state_cache/mod.rs
+++ b/src/service/rooms/state_cache/mod.rs
@ -4,10 +4,14 @@ use std::{collections::HashSet, sync::Arc};
 pub use data::Data;

 use ruma::{
+    api::federation::{self, query::get_profile_information::v1::ProfileField},
    events::{
        direct::DirectEvent,
        ignored_user_list::IgnoredUserListEvent,
-        room::{create::RoomCreateEventContent, member::MembershipState},
+        room::{
+            create::RoomCreateEventContent,
+            member::{MembershipState, RoomMemberEventContent},
+        },
        AnyStrippedStateEvent, AnySyncStateEvent, GlobalAccountDataEventType,
        RoomAccountDataEventType, StateEventType,
    },
@ -24,19 +28,43 @@ pub struct Service {
 impl Service {
    /// Update current membership data.
    #[tracing::instrument(skip(self, last_state))]
-    pub fn update_membership(
+    pub async fn update_membership(
        &self,
        room_id: &RoomId,
        user_id: &UserId,
-        membership: MembershipState,
+        membership_event: RoomMemberEventContent,
        sender: &UserId,
        last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
        update_joined_count: bool,
    ) -> Result<()> {
+        let membership = membership_event.membership;
        // Keep track what remote users exist by adding them as "deactivated" users
        if user_id.server_name() != services().globals.server_name() {
            services().users.create(user_id, None)?;
-            // TODO: displayname, avatar url
+            // Try to update our local copy of the user if ours does not match
+            if ((services().users.displayname(user_id)? != membership_event.displayname)
+                || (services().users.avatar_url(user_id)? != membership_event.avatar_url)
+                || (services().users.blurhash(user_id)? != membership_event.blurhash))
+                && (membership != MembershipState::Leave)
+            {
+                let response = services()
+                    .sending
+                    .send_federation_request(
+                        user_id.server_name(),
+                        federation::query::get_profile_information::v1::Request {
+                            user_id: user_id.into(),
+                            field: Some(ProfileField::AvatarUrl),
+                        },
+                    )
+                    .await?;
+                let _ = services()
+                    .users
+                    .set_displayname(user_id, response.displayname.clone());
+                let _ = services()
+                    .users
+                    .set_avatar_url(user_id, response.avatar_url);
+                let _ = services().users.set_blurhash(user_id, response.blurhash);
+            };
        }

        match &membership {
--- a/src/service/rooms/timeline/mod.rs
+++ b/src/service/rooms/timeline/mod.rs
@ -15,7 +15,8 @@ use ruma::{
    events::{
        push_rules::PushRulesEvent,
        room::{
-            create::RoomCreateEventContent, member::MembershipState,
+            create::RoomCreateEventContent,
+            member::{MembershipState, RoomMemberEventContent},
            power_levels::RoomPowerLevelsEventContent,
        },
        GlobalAccountDataEventType, RoomEventType, StateEventType,
@ -145,7 +146,7 @@ impl Service {
    ///
    /// Returns pdu id
    #[tracing::instrument(skip(self, pdu, pdu_json, leaves))]
-    pub fn append_pdu<'a>(
+    pub async fn append_pdu<'a>(
        &self,
        pdu: &PduEvent,
        mut pdu_json: CanonicalJsonObject,
@ -211,7 +212,7 @@ impl Service {
                .entry(pdu.room_id.clone())
                .or_default(),
        );
-        let insert_lock = mutex_insert.lock().unwrap();
+        let insert_lock = mutex_insert.lock().await;

        let count1 = services().globals.next_count()?;
        // Mark as read first so the sending client doesn't get a notification even if appending
@ -323,16 +324,11 @@ impl Service {
            }
            RoomEventType::RoomMember => {
                if let Some(state_key) = &pdu.state_key {
-                    #[derive(Deserialize)]
-                    struct ExtractMembership {
-                        membership: MembershipState,
-                    }
-
                    // if the state_key fails
                    let target_user_id = UserId::parse(state_key.clone())
                        .expect("This state_key was previously validated");

-                    let content = serde_json::from_str::<ExtractMembership>(pdu.content.get())
+                    let content = serde_json::from_str::<RoomMemberEventContent>(pdu.content.get())
                        .map_err(|_| Error::bad_database("Invalid content in pdu."))?;

                    let invite_state = match content.membership {
@ -345,14 +341,18 @@ impl Service {

                    // Update our membership info, we do this here incase a user is invited
                    // and immediately leaves we need the DB to record the invite event for auth
-                    services().rooms.state_cache.update_membership(
+                    services()
+                        .rooms
+                        .state_cache
+                        .update_membership(
                            &pdu.room_id,
                            &target_user_id,
-                        content.membership,
+                            content,
                            &pdu.sender,
                            invite_state,
                            true,
-                    )?;
+                        )
+                        .await?;
                }
            }
            RoomEventType::RoomMessage => {
@ -673,7 +673,7 @@ impl Service {
    /// 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))]
-    pub fn build_and_append_pdu(
+    pub async fn build_and_append_pdu(
        &self,
        pdu_builder: PduBuilder,
        sender: &UserId,
@ -687,14 +687,16 @@ impl Service {
        // pdu without it's state. This is okay because append_pdu can't fail.
        let statehashid = services().rooms.state.append_to_state(&pdu)?;

-        let pdu_id = self.append_pdu(
+        let pdu_id = 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
@ -732,7 +734,7 @@ impl Service {
    /// Append the incoming event setting the state snapshot to the state from the
    /// server that sent the event.
    #[tracing::instrument(skip_all)]
-    pub fn append_incoming_pdu<'a>(
+    pub async fn append_incoming_pdu<'a>(
        &self,
        pdu: &PduEvent,
        pdu_json: CanonicalJsonObject,
@ -762,11 +764,11 @@ impl Service {
            return Ok(None);
        }

-        let pdu_id =
-            services()
+        let pdu_id = services()
            .rooms
            .timeline
-                .append_pdu(pdu, pdu_json, new_room_leaves, state_lock)?;
+            .append_pdu(pdu, pdu_json, new_room_leaves, state_lock)
+            .await?;

        Ok(Some(pdu_id))
    }
Author	SHA1	Message	Date
Nyaaori	f458916919	fix: Do not allow fetching cached remote users' profiles over federation	2022-12-21 17:46:01 +01:00
Nyaaori	3ac3bdbac0	feat: Keep track of avatar urls, displaynames, and blurhashes of remote users for the room directory	2022-12-21 17:28:21 +01:00