Merge branch 'v9' into 'next'

Support all room versions from V3 to V9

Closes #161

See merge request famedly/conduit!257

src/client_server/capabilities.rs

@@ -1,9 +1,6 @@
-use crate::{Result, Ruma};
+use crate::{database::DatabaseGuard, Result, Ruma};
-use ruma::{
+use ruma::api::client::discovery::get_capabilities::{
-    api::client::discovery::get_capabilities::{
+    self, Capabilities, RoomVersionStability, RoomVersionsCapability,
-        self, Capabilities, RoomVersionStability, RoomVersionsCapability,
-    },
-    RoomVersionId,
 };
 use std::collections::BTreeMap;
 
@@ -11,15 +8,26 @@ use std::collections::BTreeMap;
 ///
 /// Get information on the supported feature set and other relevent capabilities of this server.
 pub async fn get_capabilities_route(
+    db: DatabaseGuard,
     _body: Ruma<get_capabilities::v3::IncomingRequest>,
 ) -> Result<get_capabilities::v3::Response> {
     let mut available = BTreeMap::new();
-    available.insert(RoomVersionId::V5, RoomVersionStability::Stable);
+    if db.globals.allow_unstable_room_versions() {
-    available.insert(RoomVersionId::V6, RoomVersionStability::Stable);
+        for room_version in &db.globals.unstable_room_versions {
+            available.insert(room_version.clone(), RoomVersionStability::Stable);
+        }
+    } else {
+        for room_version in &db.globals.unstable_room_versions {
+            available.insert(room_version.clone(), RoomVersionStability::Unstable);
+        }
+    }
+    for room_version in &db.globals.stable_room_versions {
+        available.insert(room_version.clone(), RoomVersionStability::Stable);
+    }
 
     let mut capabilities = Capabilities::new();
     capabilities.room_versions = RoomVersionsCapability {
-        default: RoomVersionId::V6,
+        default: db.globals.default_room_version(),
         available,
     };
 

src/client_server/membership.rs

@@ -492,7 +492,7 @@ async fn join_room_by_id_helper(
                     federation::membership::prepare_join_event::v1::Request {
                         room_id,
                         user_id: sender_user,
-                        ver: &[RoomVersionId::V5, RoomVersionId::V6],
+                        ver: &db.globals.supported_room_versions(),
                     },
                 )
                 .await;
@@ -507,11 +507,7 @@ async fn join_room_by_id_helper(
         let (make_join_response, remote_server) = make_join_response_and_server?;
 
         let room_version = match make_join_response.room_version {
-            Some(room_version)
+            Some(room_version) if db.rooms.is_supported_version(&db, &room_version) => room_version,
-                if room_version == RoomVersionId::V5 || room_version == RoomVersionId::V6 =>
-            {
-                room_version
-            }
             _ => return Err(Error::BadServerResponse("Room version is not supported")),
         };
 
@@ -828,9 +824,12 @@ pub(crate) async fn invite_helper<'a>(
                 })
                 .transpose()?;
 
-            // If there was no create event yet, assume we are creating a version 6 room right now
+            // If there was no create event yet, assume we are creating a room with the default
+            // version right now
             let room_version_id = create_event_content
-                .map_or(RoomVersionId::V6, |create_event| create_event.room_version);
+                .map_or(db.globals.default_room_version(), |create_event| {
+                    create_event.room_version
+                });
             let room_version =
                 RoomVersion::new(&room_version_id).expect("room version is supported");
 
@@ -976,7 +975,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 crate::pdu::gen_event_id_canonical_json(&response.event) {
+        let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&response.event, &db)
+        {
             Ok(t) => t,
             Err(_) => {
                 // Event could not be converted to canonical json

src/client_server/room.rs

@@ -23,7 +23,7 @@ use ruma::{
     },
     int,
     serde::{CanonicalJsonObject, JsonObject},
-    RoomAliasId, RoomId, RoomVersionId,
+    RoomAliasId, RoomId,
 };
 use serde_json::{json, value::to_raw_value};
 use std::{cmp::max, collections::BTreeMap, sync::Arc};
@@ -100,7 +100,7 @@ pub async fn create_room_route(
 
     let room_version = match body.room_version.clone() {
         Some(room_version) => {
-            if room_version == RoomVersionId::V5 || room_version == RoomVersionId::V6 {
+            if db.rooms.is_supported_version(&db, &room_version) {
                 room_version
             } else {
                 return Err(Error::BadRequest(
@@ -109,7 +109,7 @@ pub async fn create_room_route(
                 ));
             }
         }
-        None => RoomVersionId::V6,
+        None => db.globals.default_room_version(),
     };
 
     let content = match &body.creation_content {
@@ -484,7 +484,7 @@ pub async fn upgrade_room_route(
 ) -> Result<upgrade_room::v3::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    if !matches!(body.new_version, RoomVersionId::V5 | RoomVersionId::V6) {
+    if !db.rooms.is_supported_version(&db, &body.new_version) {
         return Err(Error::BadRequest(
             ErrorKind::UnsupportedRoomVersion,
             "This server does not support that room version.",

src/client_server/session.rs

@@ -14,7 +14,7 @@ use tracing::info;
 #[derive(Debug, Deserialize)]
 struct Claims {
     sub: String,
-    exp: usize,
+    //exp: usize,
 }
 
 /// # `GET /_matrix/client/r0/login`

src/client_server/typing.rs

@@ -1,5 +1,5 @@
-use crate::{database::DatabaseGuard, utils, Result, Ruma};
+use crate::{database::DatabaseGuard, utils, Error, Result, Ruma};
-use ruma::api::client::typing::create_typing_event;
+use ruma::api::client::{error::ErrorKind, typing::create_typing_event};
 
 /// # `PUT /_matrix/client/r0/rooms/{roomId}/typing/{userId}`
 ///
@@ -12,6 +12,13 @@ pub async fn create_typing_event_route(
 
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
+    if !db.rooms.is_joined(sender_user, &body.room_id)? {
+        return Err(Error::BadRequest(
+            ErrorKind::Forbidden,
+            "You are not in this room.",
+        ));
+    }
+
     if let Typing::Yes(duration) = body.state {
         db.rooms.edus.typing_add(
             sender_user,

src/config.rs

@@ -4,7 +4,7 @@ use std::{
     net::{IpAddr, Ipv4Addr},
 };
 
-use ruma::ServerName;
+use ruma::{RoomVersionId, ServerName};
 use serde::{de::IgnoredAny, Deserialize};
 use tracing::warn;
 
@@ -46,6 +46,10 @@ pub struct Config {
     pub allow_federation: bool,
     #[serde(default = "true_fn")]
     pub allow_room_creation: bool,
+    #[serde(default = "true_fn")]
+    pub allow_unstable_room_versions: bool,
+    #[serde(default = "default_default_room_version")]
+    pub default_room_version: RoomVersionId,
     #[serde(default = "false_fn")]
     pub allow_jaeger: bool,
     #[serde(default = "false_fn")]
@@ -246,3 +250,8 @@ fn default_log() -> String {
 fn default_turn_ttl() -> u64 {
     60 * 60 * 24
 }
+
+// I know, it's a great name
+fn default_default_room_version() -> RoomVersionId {
+    RoomVersionId::V6
+}

src/database.rs

@@ -21,8 +21,8 @@ use futures_util::{stream::FuturesUnordered, StreamExt};
 use lru_cache::LruCache;
 use ruma::{
     events::{
-        push_rules::PushRulesEventContent, room::message::RoomMessageEventContent, EventType,
+        push_rules::PushRulesEventContent, room::message::RoomMessageEventContent,
-        GlobalAccountDataEvent,
+        GlobalAccountDataEvent, GlobalAccountDataEventType,
     },
     push::Ruleset,
     DeviceId, EventId, RoomId, UserId,
@@ -968,7 +968,7 @@ fn set_emergency_access(db: &Database) -> Result<bool> {
     db.account_data.update(
         None,
         &conduit_user,
-        EventType::PushRules,
+        GlobalAccountDataEventType::PushRules.to_string().into(),
         &GlobalAccountDataEvent {
             content: PushRulesEventContent { global: ruleset },
         },

src/database/globals.rs

@@ -4,7 +4,8 @@ use ruma::{
         client::sync::sync_events,
         federation::discovery::{ServerSigningKeys, VerifyKey},
     },
-    DeviceId, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName, ServerSigningKeyId, UserId,
+    DeviceId, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, ServerName,
+    ServerSigningKeyId, UserId,
 };
 use std::{
     collections::{BTreeMap, HashMap},
@@ -41,6 +42,8 @@ pub struct Globals {
     jwt_decoding_key: Option<jsonwebtoken::DecodingKey<'static>>,
     federation_client: reqwest::Client,
     default_client: reqwest::Client,
+    pub stable_room_versions: Vec<RoomVersionId>,
+    pub unstable_room_versions: Vec<RoomVersionId>,
     pub(super) server_signingkeys: Arc<dyn Tree>,
     pub bad_event_ratelimiter: Arc<RwLock<HashMap<Box<EventId>, RateLimitState>>>,
     pub bad_signature_ratelimiter: Arc<RwLock<HashMap<Vec<String>, RateLimitState>>>,
@@ -145,7 +148,17 @@ impl Globals {
             })
             .build()?;
 
-        let s = Self {
+        // Supported and stable room versions
+        let stable_room_versions = vec![
+            RoomVersionId::V6,
+            RoomVersionId::V7,
+            RoomVersionId::V8,
+            RoomVersionId::V9,
+        ];
+        // Experimental, partially supported room versions
+        let unstable_room_versions = vec![RoomVersionId::V3, RoomVersionId::V4, RoomVersionId::V5];
+
+        let mut s = Self {
             globals,
             config,
             keypair: Arc::new(keypair),
@@ -162,6 +175,8 @@ impl Globals {
             default_client,
             server_signingkeys,
             jwt_decoding_key,
+            stable_room_versions,
+            unstable_room_versions,
             bad_event_ratelimiter: Arc::new(RwLock::new(HashMap::new())),
             bad_signature_ratelimiter: Arc::new(RwLock::new(HashMap::new())),
             servername_ratelimiter: Arc::new(RwLock::new(HashMap::new())),
@@ -174,6 +189,14 @@ impl Globals {
 
         fs::create_dir_all(s.get_media_folder())?;
 
+        if !s
+            .supported_room_versions()
+            .contains(&s.config.default_room_version)
+        {
+            error!("Room version in config isn't supported, falling back to Version 6");
+            s.config.default_room_version = RoomVersionId::V6;
+        };
+
         Ok(s)
     }
 
@@ -232,6 +255,14 @@ impl Globals {
         self.config.allow_room_creation
     }
 
+    pub fn allow_unstable_room_versions(&self) -> bool {
+        self.config.allow_unstable_room_versions
+    }
+
+    pub fn default_room_version(&self) -> RoomVersionId {
+        self.config.default_room_version.clone()
+    }
+
     pub fn trusted_servers(&self) -> &[Box<ServerName>] {
         &self.config.trusted_servers
     }
@@ -268,6 +299,15 @@ impl Globals {
         &self.config.emergency_password
     }
 
+    pub fn supported_room_versions(&self) -> Vec<RoomVersionId> {
+        let mut room_versions: Vec<RoomVersionId> = vec![];
+        room_versions.extend(self.stable_room_versions.clone());
+        if self.allow_unstable_room_versions() {
+            room_versions.extend(self.unstable_room_versions.clone());
+        };
+        room_versions
+    }
+
     /// TODO: the key valid until timestamp is only honored in room version > 4
     /// Remove the outdated keys and insert the new ones.
     ///

src/database/rooms.rs

@@ -133,6 +133,12 @@ pub struct Rooms {
 }
 
 impl Rooms {
+    /// Returns true if a given room version is supported
+    #[tracing::instrument(skip(self, db))]
+    pub fn is_supported_version(&self, db: &Database, room_version: &RoomVersionId) -> bool {
+        db.globals.supported_room_versions().contains(room_version)
+    }
+
     /// Builds a StateMap by iterating over all keys that start
     /// with state_hash, this gives the full state for the given state_hash.
     #[tracing::instrument(skip(self))]
@@ -1839,9 +1845,12 @@ impl Rooms {
             })
             .transpose()?;
 
-        // If there was no create event yet, assume we are creating a version 6 room right now
+        // If there was no create event yet, assume we are creating a room with the default
+        // version right now
         let room_version_id = create_event_content
-            .map_or(RoomVersionId::V6, |create_event| create_event.room_version);
+            .map_or(db.globals.default_room_version(), |create_event| {
+                create_event.room_version
+            });
         let room_version = RoomVersion::new(&room_version_id).expect("room version is supported");
 
         let auth_events =
@@ -2672,9 +2681,7 @@ impl Rooms {
         let (make_leave_response, remote_server) = make_leave_response_and_server?;
 
         let room_version_id = match make_leave_response.room_version {
-            Some(version) if version == RoomVersionId::V5 || version == RoomVersionId::V6 => {
+            Some(version) if self.is_supported_version(&db, &version) => version,
-                version
-            }
             _ => return Err(Error::BadServerResponse("Room version is not supported")),
         };
 
@@ -3430,4 +3437,24 @@ impl Rooms {
 
         Ok(())
     }
+
+    /// Returns the room's version.
+    #[tracing::instrument(skip(self))]
+    pub fn get_room_version(&self, room_id: &RoomId) -> Result<RoomVersionId> {
+        let create_event = self.room_state_get(room_id, &StateEventType::RoomCreate, "")?;
+
+        let create_event_content: Option<RoomCreateEventContent> = create_event
+            .as_ref()
+            .map(|create_event| {
+                serde_json::from_str(create_event.content.get()).map_err(|e| {
+                    warn!("Invalid create event: {}", e);
+                    Error::bad_database("Invalid create event in db.")
+                })
+            })
+            .transpose()?;
+        let room_version = create_event_content
+            .map(|create_event| create_event.room_version)
+            .ok_or_else(|| Error::BadDatabase("Invalid room version"))?;
+        Ok(room_version)
+    }
 }

src/main.rs

@@ -141,7 +141,7 @@ async fn run_server(config: &Config, db: Arc<RwLock<Database>>) -> io::Result<()
         .compression()
         .layer(
             CorsLayer::new()
-                .allow_origin(cors::any())
+                .allow_origin(cors::Any)
                 .allow_methods([
                     Method::GET,
                     Method::POST,

src/pdu.rs

@@ -1,11 +1,11 @@
-use crate::Error;
+use crate::{Database, Error};
 use ruma::{
     events::{
         room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyRoomEvent, AnyStateEvent,
         AnyStrippedStateEvent, AnySyncRoomEvent, AnySyncStateEvent, RoomEventType, StateEvent,
     },
     serde::{CanonicalJsonObject, CanonicalJsonValue, Raw},
-    state_res, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, UInt, UserId,
+    state_res, EventId, MilliSecondsSinceUnixEpoch, RoomId, UInt, UserId,
 };
 use serde::{Deserialize, Serialize};
 use serde_json::{
@@ -51,7 +51,7 @@ impl PduEvent {
         self.unsigned = None;
 
         let allowed: &[&str] = match self.kind {
-            RoomEventType::RoomMember => &["membership"],
+            RoomEventType::RoomMember => &["join_authorised_via_users_server", "membership"],
             RoomEventType::RoomCreate => &["creator"],
             RoomEventType::RoomJoinRules => &["join_rule"],
             RoomEventType::RoomPowerLevels => &[
@@ -332,16 +332,24 @@ 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,
+    db: &Database,
 ) -> crate::Result<(Box<EventId>, CanonicalJsonObject)> {
-    let value = serde_json::from_str(pdu.get()).map_err(|e| {
+    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 = db.rooms.get_room_version(&room_id);
+
     let event_id = format!(
         "${}",
         // Anything higher than version3 behaves the same
-        ruma::signatures::reference_hash(&value, &RoomVersionId::V6)
+        ruma::signatures::reference_hash(&value, &room_version_id?)
             .expect("ruma can calculate reference hashes")
     )
     .try_into()

src/server_server.rs

@@ -42,6 +42,7 @@ use ruma::{
         receipt::{ReceiptEvent, ReceiptEventContent},
         room::{
             create::RoomCreateEventContent,
+            join_rules::{JoinRule, RoomJoinRulesEventContent},
             member::{MembershipState, RoomMemberEventContent},
             server_acl::RoomServerAclEventContent,
         },
@@ -658,7 +659,7 @@ pub async fn send_transaction_message_route(
 
     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 crate::pdu::gen_event_id_canonical_json(pdu) {
+        let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(pdu, &db) {
             Ok(t) => t,
             Err(_) => {
                 // Event could not be converted to canonical json
@@ -769,17 +770,21 @@ pub async fn send_transaction_message_route(
                 }
             }
             Edu::Typing(typing) => {
-                if typing.typing {
+                if db.rooms.is_joined(&typing.user_id, &typing.room_id)? {
-                    db.rooms.edus.typing_add(
+                    if typing.typing {
-                        &typing.user_id,
+                        db.rooms.edus.typing_add(
-                        &typing.room_id,
+                            &typing.user_id,
-                        3000 + utils::millis_since_unix_epoch(),
+                            &typing.room_id,
-                        &db.globals,
+                            3000 + utils::millis_since_unix_epoch(),
-                    )?;
+                            &db.globals,
-                } else {
+                        )?;
-                    db.rooms
+                    } else {
-                        .edus
+                        db.rooms.edus.typing_remove(
-                        .typing_remove(&typing.user_id, &typing.room_id, &db.globals)?;
+                            &typing.user_id,
+                            &typing.room_id,
+                            &db.globals,
+                        )?;
+                    }
                 }
             }
             Edu::DeviceListUpdate(DeviceListUpdateContent { user_id, .. }) => {
@@ -1858,7 +1863,7 @@ pub(crate) fn fetch_and_handle_outliers<'a>(
                     Ok(res) => {
                         warn!("Got {} over federation", next_id);
                         let (calculated_event_id, value) =
-                            match crate::pdu::gen_event_id_canonical_json(&res.pdu) {
+                            match crate::pdu::gen_event_id_canonical_json(&res.pdu, &db) {
                                 Ok(t) => t,
                                 Err(_) => {
                                     back_off((*next_id).to_owned());
@@ -2590,6 +2595,33 @@ pub async fn create_join_event_template_route(
 
     acl_check(sender_servername, &body.room_id, &db)?;
 
+    // TODO: Conduit does not implement restricted join rules yet, we always reject
+    let join_rules_event =
+        db.rooms
+            .room_state_get(&body.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 let Some(join_rules_event_content) = join_rules_event_content {
+        if matches!(
+            join_rules_event_content.join_rule,
+            JoinRule::Restricted { .. }
+        ) {
+            return Err(Error::BadRequest(
+                ErrorKind::Unknown,
+                "Conduit does not support restricted rooms yet.",
+            ));
+        }
+    }
+
     let prev_events: Vec<_> = db
         .rooms
         .get_pdu_leaves(&body.room_id)?
@@ -2611,9 +2643,12 @@ pub async fn create_join_event_template_route(
         })
         .transpose()?;
 
-    // If there was no create event yet, assume we are creating a version 6 room right now
+    // If there was no create event yet, assume we are creating a room with the default version
-    let room_version_id =
+    // right now
-        create_event_content.map_or(RoomVersionId::V6, |create_event| create_event.room_version);
+    let room_version_id = create_event_content
+        .map_or(db.globals.default_room_version(), |create_event| {
+            create_event.room_version
+        });
     let room_version = RoomVersion::new(&room_version_id).expect("room version is supported");
 
     if !body.ver.contains(&room_version_id) {
@@ -2749,6 +2784,33 @@ async fn create_join_event(
 
     acl_check(sender_servername, room_id, db)?;
 
+    // TODO: Conduit does not implement restricted join rules yet, we always reject
+    let join_rules_event = db
+        .rooms
+        .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 let Some(join_rules_event_content) = join_rules_event_content {
+        if matches!(
+            join_rules_event_content.join_rule,
+            JoinRule::Restricted { .. }
+        ) {
+            return Err(Error::BadRequest(
+                ErrorKind::Unknown,
+                "Conduit does not support restricted rooms yet.",
+            ));
+        }
+    }
+
     // We need to return the state prior to joining, let's keep a reference to that here
     let shortstatehash = db
         .rooms
@@ -2762,7 +2824,7 @@ 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 crate::pdu::gen_event_id_canonical_json(pdu) {
+    let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(pdu, &db) {
         Ok(t) => t,
         Err(_) => {
             // Event could not be converted to canonical json
@@ -2888,7 +2950,7 @@ pub async fn create_invite_route(
 
     acl_check(sender_servername, &body.room_id, &db)?;
 
-    if body.room_version != RoomVersionId::V5 && body.room_version != RoomVersionId::V6 {
+    if !db.rooms.is_supported_version(&db, &body.room_version) {
         return Err(Error::BadRequest(
             ErrorKind::IncompatibleRoomVersion {
                 room_version: body.room_version.clone(),