Merge branch 'docs' into 'next'

fix: config options for well_known have changed

Closes #474

See merge request famedly/conduit!723

.gitlab-ci.yml

@@ -103,6 +103,11 @@ artifacts:
     - ./bin/nix-build-and-cache .#static-aarch64-unknown-linux-musl
     - cp result/bin/conduit aarch64-unknown-linux-musl
 
+    - mkdir -p target/aarch64-unknown-linux-musl/release
+    - cp result/bin/conduit target/aarch64-unknown-linux-musl/release
+    - direnv exec . cargo deb --no-strip --no-build --target aarch64-unknown-linux-musl
+    - mv target/aarch64-unknown-linux-musl/debian/*.deb aarch64-unknown-linux-musl.deb
+
     - ./bin/nix-build-and-cache .#oci-image-aarch64-unknown-linux-musl
     - cp result oci-image-arm64v8.tar.gz
 
@@ -114,6 +119,7 @@ artifacts:
       - x86_64-unknown-linux-musl
       - aarch64-unknown-linux-musl
       - x86_64-unknown-linux-musl.deb
+      - aarch64-unknown-linux-musl.deb
       - oci-image-amd64.tar.gz
       - oci-image-arm64v8.tar.gz
       - public

Cargo.toml

@@ -16,10 +16,10 @@ license = "Apache-2.0"
 name = "conduit"
 readme = "README.md"
 repository = "https://gitlab.com/famedly/conduit"
-version = "0.8.0-alpha"
+version = "0.10.0-alpha"
 
 # See also `rust-toolchain.toml`
-rust-version = "1.78.0"
+rust-version = "1.79.0"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -28,42 +28,24 @@ workspace = true
 
 [dependencies]
 # Web framework
-axum = { version = "0.6.18", default-features = false, features = [
+axum = { version = "0.7", default-features = false, features = [
   "form",
-  "headers",
   "http1",
   "http2",
   "json",
   "matched-path",
 ], optional = true }
-axum-server = { version = "0.5.1", features = ["tls-rustls"] }
+axum-extra = { version = "0.9", features = ["typed-header"] }
+axum-server = { version = "0.6", features = ["tls-rustls"] }
 tower = { version = "0.4.13", features = ["util"] }
-tower-http = { version = "0.4.1", features = [
+tower-http = { version = "0.5", features = [
   "add-extension",
   "cors",
   "sensitive-headers",
   "trace",
   "util",
 ] }
-
-# Used for matrix spec type definitions and helpers
-#ruma = { version = "0.4.0", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
-ruma = { git = "https://github.com/ruma/ruma", rev = "5495b85aa311c2805302edb0a7de40399e22b397", features = [
-  "appservice-api-c",
-  "client-api",
-  "compat",
-  "federation-api",
-  "push-gateway-api-c",
-  "rand",
-  "ring-compat",
-  "state-res",
-  "unstable-exhaustive-types",
-  "unstable-msc2448",
-  "unstable-msc3575",
-  "unstable-unspecified",
-] }
-#ruma = { git = "https://github.com/timokoesters/ruma", rev = "4ec9c69bb7e09391add2382b3ebac97b6e8f4c64", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-msc2448", "unstable-msc3575", "unstable-exhaustive-types", "ring-compat", "unstable-unspecified" ] }
-#ruma = { path = "../ruma/crates/ruma", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-msc2448", "unstable-msc3575", "unstable-exhaustive-types", "ring-compat", "unstable-unspecified" ] }
+tower-service = "0.3"
 
 # Async runtime and utilities
 tokio = { version = "1.28.1", features = ["fs", "macros", "signal", "sync"] }
@@ -74,7 +56,7 @@ persy = { version = "1.4.4", optional = true, features = ["background_ops"] }
 
 # Used for the http request / response body type for Ruma endpoints used with reqwest
 bytes = "1.4.0"
-http = "0.2.9"
+http = "1"
 # Used to find data directory for default db path
 directories = "5"
 # Used for ruma wrapper
@@ -88,8 +70,14 @@ rand = "0.8.5"
 # Used to hash passwords
 rust-argon2 = "2"
 # Used to send requests
-hyper = "0.14.26"
-reqwest = { version = "0.11.18", default-features = false, features = [
+hyper = "1.1"
+hyper-util = { version = "0.1", features = [
+  "client",
+  "client-legacy",
+  "http1",
+  "http2",
+] }
+reqwest = { version = "0.12", default-features = false, features = [
   "rustls-tls-native-roots",
   "socks",
 ] }
@@ -112,11 +100,13 @@ regex = "1.8.1"
 # jwt jsonwebtokens
 jsonwebtoken = "9.2.0"
 # Performance measurements
-opentelemetry = { version = "0.18.0", features = ["rt-tokio"] }
-opentelemetry-jaeger = { version = "0.17.0", features = ["rt-tokio"] }
-tracing = { version = "0.1.37", features = [] }
+opentelemetry = "0.22"
+opentelemetry-jaeger-propagator = "0.1"
+opentelemetry-otlp = "0.15"
+opentelemetry_sdk = { version = "0.22", features = ["rt-tokio"] }
+tracing = "0.1.37"
 tracing-flame = "0.2.0"
-tracing-opentelemetry = "0.18.0"
+tracing-opentelemetry = "0.23"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 
 lru-cache = "0.1.2"
@@ -157,6 +147,25 @@ tikv-jemallocator = { version = "0.5.0", features = [
 
 sd-notify = { version = "0.4.1", optional = true }
 
+# Used for matrix spec type definitions and helpers
+[dependencies.ruma]
+features = [
+  "appservice-api-c",
+  "client-api",
+  "compat",
+  "federation-api",
+  "push-gateway-api-c",
+  "rand",
+  "ring-compat",
+  "server-util",
+  "state-res",
+  "unstable-exhaustive-types",
+  "unstable-msc2448",
+  "unstable-msc3575",
+  "unstable-unspecified",
+]
+git = "https://github.com/ruma/ruma"
+
 [dependencies.rocksdb]
 features = ["lz4", "multi-threaded-cf", "zstd"]
 optional = true

README.md

@@ -56,6 +56,13 @@ If you have any questions, feel free to
 - Send an direct message to `@timokoesters:fachschaften.org` on Matrix
 - [Open an issue on GitLab](https://gitlab.com/famedly/conduit/-/issues/new)
 
+#### Security
+
+If you believe you have found a security issue, please send a message to [Timo](https://matrix.to/#/@timo:conduit.rs)
+and/or [Matthias](https://matrix.to/#/@matthias:ahouansou.cz) on Matrix, or send an email to
+[conduit@koesters.xyz](mailto:conduit@koesters.xyz). Please do not disclose details about the issue to anyone else before
+a fix is released publically.
+
 #### Thanks to
 
 Thanks to FUTO, Famedly, Prototype Fund (DLR and German BMBF) and all individuals for financially supporting this project.

complement/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.78.0
+FROM rust:1.79.0
 
 WORKDIR /workdir
 

docs/configuration.md

@@ -6,6 +6,8 @@
 
 > **Note:** If you update the configuration file, you must restart Conduit for the changes to take effect
 
+> **Note:** You can also configure Conduit by using `CONDUIT_{field_name}` environment variables. To set values inside a table, use `CONDUIT_{table_name}__{field_name}`. Example: `CONDUIT_SERVER_NAME="example.org"`
+
 Conduit's configuration file is divided into the following sections:
 
 - [Global](#global)
@@ -56,7 +58,8 @@ The `global` section contains the following fields:
 | `turn_secret` | `string` | The TURN secret | `""` |
 | `turn_ttl` | `integer` | The TURN TTL in seconds | `86400` |
 | `emergency_password` | `string` | Set a password to login as the `conduit` user in case of emergency | N/A |
-| `well_known` | `table` | Used for [delegation](delegation.md) | See [delegation](delegation.md) |
+| `well_known_client` | `string` | Used for [delegation](delegation.md) | See [delegation](delegation.md) |
+| `well_known_server` | `string` | Used for [delegation](delegation.md) | See [delegation](delegation.md) |
 
 
 ### TLS

docs/delegation.md

@@ -16,18 +16,18 @@ are connected to the server running Conduit using something like a VPN.
 
 > **Note**: this will automatically allow you to use [sliding sync][0] without any extra configuration
 
-To configure it, use the following options in the `global.well_known` table:
+To configure it, use the following options:
 | Field | Type | Description | Default |
 | --- | --- | --- | --- |
-| `client` | `String` | The URL that clients should use to connect to Conduit | `https://<server_name>` |
-| `server` | `String` | The hostname and port servers should use to connect to Conduit | `<server_name>:443` |
+| `well_known_client` | `String` | The URL that clients should use to connect to Conduit | `https://<server_name>` |
+| `well_known_server` | `String` | The hostname and port servers should use to connect to Conduit | `<server_name>:443` |
 
 ### Example
 
 ```toml
-[global.well_known]
-client = "https://matrix.example.org"
-server = "matrix.example.org:443"
+[global]
+well_known_client = "https://matrix.example.org"
+well_known_server = "matrix.example.org:443"
 ```
 
 ## Manual

docs/deploying/docker.md

@@ -64,6 +64,7 @@ docker run -d -p 8448:6167 \
   -e CONDUIT_MAX_REQUEST_SIZE="20000000" \
   -e CONDUIT_TRUSTED_SERVERS="[\"matrix.org\"]" \
   -e CONDUIT_MAX_CONCURRENT_REQUESTS="100" \
+  -e CONDUIT_PORT="6167" \
   --name conduit <link>
 ```
 

docs/deploying/generic.md

@@ -17,6 +17,7 @@ You may simply download the binary that fits your machine. Run `uname -m` to see
 | Target | Type | Download |
 |-|-|-|
 | `x86_64-unknown-linux-musl` | Statically linked Debian package | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/master/raw/x86_64-unknown-linux-musl.deb?job=artifacts) |
+| `aarch64-unknown-linux-musl` | Statically linked Debian package | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/master/raw/aarch64-unknown-linux-musl.deb?job=artifacts) |
 | `x86_64-unknown-linux-musl` | Statically linked binary | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/master/raw/x86_64-unknown-linux-musl?job=artifacts) |
 | `aarch64-unknown-linux-musl` | Statically linked binary | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/master/raw/aarch64-unknown-linux-musl?job=artifacts) |
 | `x86_64-unknown-linux-gnu` | OCI image | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/master/raw/oci-image-amd64.tar.gz?job=artifacts) |
@@ -30,6 +31,7 @@ If you use a system with an older glibc version (e.g. RHEL8), you might need to
 | Target | Type | Download |
 |-|-|-|
 | `x86_64-unknown-linux-musl` | Statically linked Debian package | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/next/raw/x86_64-unknown-linux-musl.deb?job=artifacts) |
+| `aarch64-unknown-linux-musl` | Statically linked Debian package | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/next/raw/aarch64-unknown-linux-musl.deb?job=artifacts) |
 | `x86_64-unknown-linux-musl` | Statically linked binary | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/next/raw/x86_64-unknown-linux-musl?job=artifacts) |
 | `aarch64-unknown-linux-musl` | Statically linked binary | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/next/raw/aarch64-unknown-linux-musl?job=artifacts) |
 | `x86_64-unknown-linux-gnu` | OCI image | [link](https://gitlab.com/api/v4/projects/famedly%2Fconduit/jobs/artifacts/next/raw/oci-image-amd64.tar.gz?job=artifacts) |

docs/faq.md

@@ -35,3 +35,7 @@ Here is an example:
 
 Not really. You can reuse the domain of your current server with Conduit, but you will not be able to migrate accounts automatically.
 Rooms that were federated can be re-joined via the other participating servers, however media and the like may be deleted from remote servers after some time, and hence might not be recoverable.
+
+## How do I make someone an admin?
+
+Simply invite them to the admin room. Once joined, they can administer the server by interacting with the `@conduit:<server_name>` user.

docs/turn.md

@@ -2,7 +2,7 @@
 
 ## General instructions
 
-* It is assumed you have a [Coturn server](https://github.com/coturn/coturn) up and running. See [Synapse reference implementation](https://github.com/matrix-org/synapse/blob/develop/docs/turn-howto.md).
+* It is assumed you have a [Coturn server](https://github.com/coturn/coturn) up and running. See [Synapse reference implementation](https://github.com/element-hq/synapse/blob/develop/docs/turn-howto.md).
 
 ## Edit/Add a few settings to your existing conduit.toml
 

flake.nix

@@ -59,7 +59,7 @@
             file = ./rust-toolchain.toml;
 
             # See also `rust-toolchain.toml`
-            sha256 = "sha256-opUgs6ckUQCyDxcB9Wy51pqhd0MPGHUVbwRKKPGiwZU=";
+            sha256 = "sha256-Ngiz76YP4HTY75GGdH2P+APE/DEIx2R/Dn+BwwOyzZU=";
           };
       });
     in

nix/shell.nix

@@ -23,7 +23,7 @@ mkShell {
   };
 
   # Development tools
-  nativeBuildInputs = default.nativeBuildInputs ++ [
+  nativeBuildInputs = [
     # Always use nightly rustfmt because most of its options are unstable
     #
     # This needs to come before `toolchain` in this list, otherwise
@@ -57,5 +57,5 @@ mkShell {
 
     # Useful for editing the book locally
     mdbook
-  ];
+  ] ++ default.nativeBuildInputs ;
 }

rust-toolchain.toml

@@ -2,7 +2,6 @@
 #
 # Other files that need upkeep when this changes:
 #
-# * `.gitlab-ci.yml`
 # * `Cargo.toml`
 # * `flake.nix`
 #
@@ -10,7 +9,7 @@
 # If you're having trouble making the relevant changes, bug a maintainer.
 
 [toolchain]
-channel = "1.78.0"
+channel = "1.79.0"
 components = [
   # For rust-analyzer
   "rust-src",

src/api/client_server/account.rs

@@ -75,9 +75,9 @@ pub async fn get_register_available_route(
 /// - Creates a new account and populates it with default account data
 /// - If `inhibit_login` is false: Creates a device and returns device id and access_token
 pub async fn register_route(body: Ruma<register::v3::Request>) -> Result<register::v3::Response> {
-    if !services().globals.allow_registration() && body.appservice_info.is_none() {
+    if !services().globals.allow_registration().await && body.appservice_info.is_none() {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Registration has been disabled.",
         ));
     }
@@ -315,7 +315,11 @@ pub async fn register_route(body: Ruma<register::v3::Request>) -> Result<registe
 pub async fn change_password_route(
     body: Ruma<change_password::v3::Request>,
 ) -> Result<change_password::v3::Response> {
-    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let sender_user = body
+        .sender_user
+        .as_ref()
+        // In the future password changes could be performed with UIA with 3PIDs, but we don't support that currently
+        .ok_or_else(|| Error::BadRequest(ErrorKind::MissingToken, "Missing access token."))?;
     let sender_device = body.sender_device.as_ref().expect("user is authenticated");
 
     let mut uiaainfo = UiaaInfo {
@@ -402,7 +406,11 @@ pub async fn whoami_route(body: Ruma<whoami::v3::Request>) -> Result<whoami::v3:
 pub async fn deactivate_route(
     body: Ruma<deactivate::v3::Request>,
 ) -> Result<deactivate::v3::Response> {
-    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let sender_user = body
+        .sender_user
+        .as_ref()
+        // In the future password changes could be performed with UIA with SSO, but we don't support that currently
+        .ok_or_else(|| Error::BadRequest(ErrorKind::MissingToken, "Missing access token."))?;
     let sender_device = body.sender_device.as_ref().expect("user is authenticated");
 
     let mut uiaainfo = UiaaInfo {
@@ -475,7 +483,7 @@ pub async fn request_3pid_management_token_via_email_route(
 ) -> Result<request_3pid_management_token_via_email::v3::Response> {
     Err(Error::BadRequest(
         ErrorKind::ThreepidDenied,
-        "Third party identifier is not allowed",
+        "Third party identifiers are currently unsupported by this server implementation",
     ))
 }
 
@@ -489,6 +497,6 @@ pub async fn request_3pid_management_token_via_msisdn_route(
 ) -> Result<request_3pid_management_token_via_msisdn::v3::Response> {
     Err(Error::BadRequest(
         ErrorKind::ThreepidDenied,
-        "Third party identifier is not allowed",
+        "Third party identifiers are currently unsupported by this server implementation",
     ))
 }

src/api/client_server/alias.rs

@@ -18,6 +18,8 @@ use ruma::{
 pub async fn create_alias_route(
     body: Ruma<create_alias::v3::Request>,
 ) -> Result<create_alias::v3::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
     if body.room_alias.server_name() != services().globals.server_name() {
         return Err(Error::BadRequest(
             ErrorKind::InvalidParam,
@@ -55,7 +57,7 @@ pub async fn create_alias_route(
     services()
         .rooms
         .alias
-        .set_alias(&body.room_alias, &body.room_id)?;
+        .set_alias(&body.room_alias, &body.room_id, sender_user)?;
 
     Ok(create_alias::v3::Response::new())
 }
@@ -64,11 +66,12 @@ pub async fn create_alias_route(
 ///
 /// Deletes a room alias from this server.
 ///
-/// - TODO: additional access control checks
 /// - TODO: Update canonical alias event
 pub async fn delete_alias_route(
     body: Ruma<delete_alias::v3::Request>,
 ) -> Result<delete_alias::v3::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
     if body.room_alias.server_name() != services().globals.server_name() {
         return Err(Error::BadRequest(
             ErrorKind::InvalidParam,
@@ -94,7 +97,10 @@ pub async fn delete_alias_route(
         ));
     }
 
-    services().rooms.alias.remove_alias(&body.room_alias)?;
+    services()
+        .rooms
+        .alias
+        .remove_alias(&body.room_alias, sender_user)?;
 
     // TODO: update alt_aliases?
 

src/api/client_server/context.rs

@@ -54,7 +54,7 @@ pub async fn get_context_route(
         .user_can_see_event(sender_user, &room_id, &body.event_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view this event.",
         ));
     }

src/api/client_server/media.rs

@@ -1,12 +1,24 @@
+// Unauthenticated media is deprecated
+#![allow(deprecated)]
+
 use std::time::Duration;
 
 use crate::{service::media::FileMeta, services, utils, Error, Result, Ruma};
-use ruma::api::client::{
-    error::ErrorKind,
-    media::{
-        create_content, get_content, get_content_as_filename, get_content_thumbnail,
-        get_media_config,
+use http::header::{CONTENT_DISPOSITION, CONTENT_TYPE};
+use ruma::{
+    api::{
+        client::{
+            authenticated_media::{
+                get_content, get_content_as_filename, get_content_thumbnail, get_media_config,
             },
+            error::ErrorKind,
+            media::{self, create_content},
+        },
+        federation::authenticated_media::{self as federation_media, FileOrLocation},
+    },
+    http_headers::{ContentDisposition, ContentDispositionType},
+    media::Method,
+    ServerName, UInt,
 };
 
 const MXC_LENGTH: usize = 32;
@@ -15,19 +27,22 @@ const MXC_LENGTH: usize = 32;
 ///
 /// Returns max upload size.
 pub async fn get_media_config_route(
-    _body: Ruma<get_media_config::v3::Request>,
-) -> Result<get_media_config::v3::Response> {
-    Ok(get_media_config::v3::Response {
+    _body: Ruma<media::get_media_config::v3::Request>,
+) -> Result<media::get_media_config::v3::Response> {
+    Ok(media::get_media_config::v3::Response {
         upload_size: services().globals.max_request_size().into(),
     })
 }
 
-fn sanitize_content_type(content_type: String) -> String {
-    if content_type == "image/jpeg" || content_type == "image/png" {
-        content_type
-    } else {
-        "application/octet-stream".to_owned()
-    }
+/// # `GET /_matrix/client/v1/media/config`
+///
+/// Returns max upload size.
+pub async fn get_media_config_auth_route(
+    _body: Ruma<get_media_config::v1::Request>,
+) -> Result<get_media_config::v1::Response> {
+    Ok(get_media_config::v1::Response {
+        upload_size: services().globals.max_request_size().into(),
+    })
 }
 
 /// # `POST /_matrix/media/r0/upload`
@@ -49,10 +64,10 @@ pub async fn create_content_route(
         .media
         .create(
             mxc.clone(),
-            body.filename
-                .as_ref()
-                .map(|filename| "inline; filename=".to_owned() + filename)
-                .as_deref(),
+            Some(
+                ContentDisposition::new(ContentDispositionType::Inline)
+                    .with_filename(body.filename.clone()),
+            ),
             body.content_type.as_deref(),
             &body.file,
         )
@@ -66,28 +81,67 @@ pub async fn create_content_route(
 
 pub async fn get_remote_content(
     mxc: &str,
-    server_name: &ruma::ServerName,
+    server_name: &ServerName,
     media_id: String,
-) -> Result<get_content::v3::Response, Error> {
-    let content_response = services()
+) -> Result<get_content::v1::Response, Error> {
+    let content_response = match services()
         .sending
         .send_federation_request(
             server_name,
-            get_content::v3::Request {
-                allow_remote: false,
+            federation_media::get_content::v1::Request {
+                media_id: media_id.clone(),
+                timeout_ms: Duration::from_secs(20),
+            },
+        )
+        .await
+    {
+        Ok(federation_media::get_content::v1::Response {
+            metadata: _,
+            content: FileOrLocation::File(content),
+        }) => get_content::v1::Response {
+            file: content.file,
+            content_type: content.content_type,
+            content_disposition: content.content_disposition,
+        },
+
+        Ok(federation_media::get_content::v1::Response {
+            metadata: _,
+            content: FileOrLocation::Location(url),
+        }) => get_location_content(url).await?,
+        Err(Error::BadRequest(ErrorKind::Unrecognized, _)) => {
+            let media::get_content::v3::Response {
+                file,
+                content_type,
+                content_disposition,
+                ..
+            } = services()
+                .sending
+                .send_federation_request(
+                    server_name,
+                    media::get_content::v3::Request {
                         server_name: server_name.to_owned(),
                         media_id,
                         timeout_ms: Duration::from_secs(20),
-                allow_redirect: false,
+                        allow_remote: false,
+                        allow_redirect: true,
                     },
                 )
                 .await?;
 
+            get_content::v1::Response {
+                file,
+                content_type,
+                content_disposition,
+            }
+        }
+        Err(e) => return Err(e),
+    };
+
     services()
         .media
         .create(
             mxc.to_owned(),
-            content_response.content_disposition.as_deref(),
+            content_response.content_disposition.clone(),
             content_response.content_type.as_deref(),
             &content_response.file,
         )
@@ -102,31 +156,57 @@ pub async fn get_remote_content(
 ///
 /// - Only allows federation if `allow_remote` is true
 pub async fn get_content_route(
-    body: Ruma<get_content::v3::Request>,
-) -> Result<get_content::v3::Response> {
-    let mxc = format!("mxc://{}/{}", body.server_name, body.media_id);
-
-    if let Some(FileMeta {
+    body: Ruma<media::get_content::v3::Request>,
+) -> Result<media::get_content::v3::Response> {
+    let get_content::v1::Response {
+        file,
         content_disposition,
+        content_type,
+    } = get_content(&body.server_name, body.media_id.clone(), body.allow_remote).await?;
+
+    Ok(media::get_content::v3::Response {
         file,
-        ..
-    }) = services().media.get(mxc.clone()).await?
-    {
-        Ok(get_content::v3::Response {
-            file,
-            content_type: Some("application/octet-stream".to_owned()),
+        content_type,
         content_disposition,
         cross_origin_resource_policy: Some("cross-origin".to_owned()),
     })
-    } else if &*body.server_name != services().globals.server_name() && body.allow_remote {
-        let remote_content_response =
-            get_remote_content(&mxc, &body.server_name, body.media_id.clone()).await?;
+}
 
-        Ok(get_content::v3::Response {
+/// # `GET /_matrix/client/v1/media/download/{serverName}/{mediaId}`
+///
+/// Load media from our server or over federation.
+pub async fn get_content_auth_route(
+    body: Ruma<get_content::v1::Request>,
+) -> Result<get_content::v1::Response> {
+    get_content(&body.server_name, body.media_id.clone(), true).await
+}
+
+async fn get_content(
+    server_name: &ServerName,
+    media_id: String,
+    allow_remote: bool,
+) -> Result<get_content::v1::Response, Error> {
+    let mxc = format!("mxc://{}/{}", server_name, media_id);
+
+    if let Ok(Some(FileMeta {
+        content_disposition,
+        content_type,
+        file,
+    })) = services().media.get(mxc.clone()).await
+    {
+        Ok(get_content::v1::Response {
+            file,
+            content_type,
+            content_disposition: Some(content_disposition),
+        })
+    } else if server_name != services().globals.server_name() && allow_remote {
+        let remote_content_response =
+            get_remote_content(&mxc, server_name, media_id.clone()).await?;
+
+        Ok(get_content::v1::Response {
             content_disposition: remote_content_response.content_disposition,
-            content_type: Some("application/octet-stream".to_owned()),
+            content_type: remote_content_response.content_type,
             file: remote_content_response.file,
-            cross_origin_resource_policy: Some("cross-origin".to_owned()),
         })
     } else {
         Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
@@ -139,26 +219,74 @@ pub async fn get_content_route(
 ///
 /// - Only allows federation if `allow_remote` is true
 pub async fn get_content_as_filename_route(
-    body: Ruma<get_content_as_filename::v3::Request>,
-) -> Result<get_content_as_filename::v3::Response> {
-    let mxc = format!("mxc://{}/{}", body.server_name, body.media_id);
-
-    if let Some(FileMeta { file, .. }) = services().media.get(mxc.clone()).await? {
-        Ok(get_content_as_filename::v3::Response {
+    body: Ruma<media::get_content_as_filename::v3::Request>,
+) -> Result<media::get_content_as_filename::v3::Response> {
+    let get_content_as_filename::v1::Response {
         file,
-            content_type: Some("application/octet-stream".to_owned()),
-            content_disposition: Some(format!("inline; filename={}", body.filename)),
+        content_type,
+        content_disposition,
+    } = get_content_as_filename(
+        &body.server_name,
+        body.media_id.clone(),
+        body.filename.clone(),
+        body.allow_remote,
+    )
+    .await?;
+
+    Ok(media::get_content_as_filename::v3::Response {
+        file,
+        content_type,
+        content_disposition,
         cross_origin_resource_policy: Some("cross-origin".to_owned()),
     })
-    } else if &*body.server_name != services().globals.server_name() && body.allow_remote {
-        let remote_content_response =
-            get_remote_content(&mxc, &body.server_name, body.media_id.clone()).await?;
+}
 
-        Ok(get_content_as_filename::v3::Response {
-            content_disposition: Some(format!("inline: filename={}", body.filename)),
-            content_type: Some("application/octet-stream".to_owned()),
+/// # `GET /_matrix/client/v1/media/download/{serverName}/{mediaId}/{fileName}`
+///
+/// Load media from our server or over federation, permitting desired filename.
+pub async fn get_content_as_filename_auth_route(
+    body: Ruma<get_content_as_filename::v1::Request>,
+) -> Result<get_content_as_filename::v1::Response, Error> {
+    get_content_as_filename(
+        &body.server_name,
+        body.media_id.clone(),
+        body.filename.clone(),
+        true,
+    )
+    .await
+}
+
+async fn get_content_as_filename(
+    server_name: &ServerName,
+    media_id: String,
+    filename: String,
+    allow_remote: bool,
+) -> Result<get_content_as_filename::v1::Response, Error> {
+    let mxc = format!("mxc://{}/{}", server_name, media_id);
+
+    if let Ok(Some(FileMeta {
+        file, content_type, ..
+    })) = services().media.get(mxc.clone()).await
+    {
+        Ok(get_content_as_filename::v1::Response {
+            file,
+            content_type,
+            content_disposition: Some(
+                ContentDisposition::new(ContentDispositionType::Inline)
+                    .with_filename(Some(filename.clone())),
+            ),
+        })
+    } else if server_name != services().globals.server_name() && allow_remote {
+        let remote_content_response =
+            get_remote_content(&mxc, server_name, media_id.clone()).await?;
+
+        Ok(get_content_as_filename::v1::Response {
+            content_disposition: Some(
+                ContentDisposition::new(ContentDispositionType::Inline)
+                    .with_filename(Some(filename.clone())),
+            ),
+            content_type: remote_content_response.content_type,
             file: remote_content_response.file,
-            cross_origin_resource_policy: Some("cross-origin".to_owned()),
         })
     } else {
         Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
@@ -171,66 +299,169 @@ pub async fn get_content_as_filename_route(
 ///
 /// - Only allows federation if `allow_remote` is true
 pub async fn get_content_thumbnail_route(
-    body: Ruma<get_content_thumbnail::v3::Request>,
-) -> Result<get_content_thumbnail::v3::Response> {
-    let mxc = format!("mxc://{}/{}", body.server_name, body.media_id);
+    body: Ruma<media::get_content_thumbnail::v3::Request>,
+) -> Result<media::get_content_thumbnail::v3::Response> {
+    let get_content_thumbnail::v1::Response { file, content_type } = get_content_thumbnail(
+        &body.server_name,
+        body.media_id.clone(),
+        body.height,
+        body.width,
+        body.method.clone(),
+        body.animated,
+        body.allow_remote,
+    )
+    .await?;
 
-    if let Some(FileMeta {
+    Ok(media::get_content_thumbnail::v3::Response {
+        file,
+        content_type,
+        cross_origin_resource_policy: Some("cross-origin".to_owned()),
+    })
+}
+
+/// # `GET /_matrix/client/v1/media/thumbnail/{serverName}/{mediaId}`
+///
+/// Load media thumbnail from our server or over federation.
+pub async fn get_content_thumbnail_auth_route(
+    body: Ruma<get_content_thumbnail::v1::Request>,
+) -> Result<get_content_thumbnail::v1::Response> {
+    get_content_thumbnail(
+        &body.server_name,
+        body.media_id.clone(),
+        body.height,
+        body.width,
+        body.method.clone(),
+        body.animated,
+        true,
+    )
+    .await
+}
+
+async fn get_content_thumbnail(
+    server_name: &ServerName,
+    media_id: String,
+    height: UInt,
+    width: UInt,
+    method: Option<Method>,
+    animated: Option<bool>,
+    allow_remote: bool,
+) -> Result<get_content_thumbnail::v1::Response, Error> {
+    let mxc = format!("mxc://{}/{}", server_name, media_id);
+
+    if let Ok(Some(FileMeta {
         file, content_type, ..
-    }) = services()
+    })) = services()
         .media
         .get_thumbnail(
             mxc.clone(),
-            body.width
+            width
                 .try_into()
                 .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Width is invalid."))?,
-            body.height
+            height
                 .try_into()
-                .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Width is invalid."))?,
+                .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Height is invalid."))?,
         )
-        .await?
+        .await
     {
-        Ok(get_content_thumbnail::v3::Response {
-            file,
-            content_type: content_type.map(sanitize_content_type),
-            cross_origin_resource_policy: Some("cross-origin".to_owned()),
-        })
-    } else if &*body.server_name != services().globals.server_name() && body.allow_remote {
-        let mut get_thumbnail_response = services()
+        Ok(get_content_thumbnail::v1::Response { file, content_type })
+    } else if server_name != services().globals.server_name() && allow_remote {
+        let thumbnail_response = match services()
             .sending
             .send_federation_request(
-                &body.server_name,
-                get_content_thumbnail::v3::Request {
-                    allow_remote: false,
-                    height: body.height,
-                    width: body.width,
-                    method: body.method.clone(),
-                    server_name: body.server_name.clone(),
-                    media_id: body.media_id.clone(),
+                server_name,
+                federation_media::get_content_thumbnail::v1::Request {
+                    height,
+                    width,
+                    method: method.clone(),
+                    media_id: media_id.clone(),
+                    timeout_ms: Duration::from_secs(20),
+                    animated,
+                },
+            )
+            .await
+        {
+            Ok(federation_media::get_content_thumbnail::v1::Response {
+                metadata: _,
+                content: FileOrLocation::File(content),
+            }) => get_content_thumbnail::v1::Response {
+                file: content.file,
+                content_type: content.content_type,
+            },
+
+            Ok(federation_media::get_content_thumbnail::v1::Response {
+                metadata: _,
+                content: FileOrLocation::Location(url),
+            }) => {
+                let get_content::v1::Response {
+                    file, content_type, ..
+                } = get_location_content(url).await?;
+
+                get_content_thumbnail::v1::Response { file, content_type }
+            }
+            Err(Error::BadRequest(ErrorKind::Unrecognized, _)) => {
+                let media::get_content_thumbnail::v3::Response {
+                    file, content_type, ..
+                } = services()
+                    .sending
+                    .send_federation_request(
+                        server_name,
+                        media::get_content_thumbnail::v3::Request {
+                            height,
+                            width,
+                            method: method.clone(),
+                            server_name: server_name.to_owned(),
+                            media_id: media_id.clone(),
                             timeout_ms: Duration::from_secs(20),
                             allow_redirect: false,
+                            animated,
+                            allow_remote: false,
                         },
                     )
                     .await?;
 
+                get_content_thumbnail::v1::Response { file, content_type }
+            }
+            Err(e) => return Err(e),
+        };
+
         services()
             .media
             .upload_thumbnail(
                 mxc,
-                None,
-                get_thumbnail_response.content_type.as_deref(),
-                body.width.try_into().expect("all UInts are valid u32s"),
-                body.height.try_into().expect("all UInts are valid u32s"),
-                &get_thumbnail_response.file,
+                thumbnail_response.content_type.as_deref(),
+                width.try_into().expect("all UInts are valid u32s"),
+                height.try_into().expect("all UInts are valid u32s"),
+                &thumbnail_response.file,
             )
             .await?;
 
-        get_thumbnail_response.content_type = get_thumbnail_response
-            .content_type
-            .map(sanitize_content_type);
-
-        Ok(get_thumbnail_response)
+        Ok(thumbnail_response)
     } else {
         Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
     }
 }
+
+async fn get_location_content(url: String) -> Result<get_content::v1::Response, Error> {
+    let client = services().globals.default_client();
+    let response = client.get(url).send().await?;
+    let headers = response.headers();
+
+    let content_type = headers
+        .get(CONTENT_TYPE)
+        .and_then(|header| header.to_str().ok())
+        .map(ToOwned::to_owned);
+
+    let content_disposition = headers
+        .get(CONTENT_DISPOSITION)
+        .map(|header| header.as_bytes())
+        .map(TryFrom::try_from)
+        .and_then(Result::ok);
+
+    let file = response.bytes().await?.to_vec();
+
+    Ok(get_content::v1::Response {
+        file,
+        content_type,
+        content_disposition,
+    })
+}

src/api/client_server/membership.rs

@@ -18,9 +18,8 @@ use ruma::{
         },
         StateEventType, TimelineEventType,
     },
-    serde::Base64,
-    state_res, CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
-    OwnedServerName, OwnedUserId, RoomId, RoomVersionId, UserId,
+    state_res, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
+    OwnedEventId, OwnedRoomId, OwnedServerName, OwnedUserId, RoomId, RoomVersionId, UserId,
 };
 use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
 use std::{
@@ -32,7 +31,10 @@ use tokio::sync::RwLock;
 use tracing::{debug, error, info, warn};
 
 use crate::{
-    service::pdu::{gen_event_id_canonical_json, PduBuilder},
+    service::{
+        globals::SigningKeys,
+        pdu::{gen_event_id_canonical_json, PduBuilder},
+    },
     services, utils, Error, PduEvent, Result, Ruma,
 };
 
@@ -95,7 +97,7 @@ pub async fn join_room_by_id_or_alias_route(
 
     let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias) {
         Ok(room_id) => {
-            let mut servers = body.server_name.clone();
+            let mut servers = body.via.clone();
             servers.extend(
                 services()
                     .rooms
@@ -186,15 +188,7 @@ pub async fn kick_user_route(
 ) -> Result<kick_user::v3::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    if let Ok(true) = services()
-        .rooms
-        .state_cache
-        .is_left(sender_user, &body.room_id)
-    {
-        return Ok(kick_user::v3::Response {});
-    }
-
-    let mut event: RoomMemberEventContent = serde_json::from_str(
+    let event: RoomMemberEventContent = serde_json::from_str(
         services()
             .rooms
             .state_accessor
@@ -205,15 +199,26 @@ pub async fn kick_user_route(
             )?
             .ok_or(Error::BadRequest(
                 ErrorKind::BadState,
-                "Cannot kick member that's not in the room.",
+                "Cannot kick a user who is not in the room.",
             ))?
             .content
             .get(),
     )
     .map_err(|_| Error::bad_database("Invalid member event in database."))?;
 
-    event.membership = MembershipState::Leave;
-    event.reason.clone_from(&body.reason);
+    // If they are already kicked and the reason is unchanged, there isn't any point in sending a new event.
+    if event.membership == MembershipState::Leave && event.reason == body.reason {
+        return Ok(kick_user::v3::Response {});
+    }
+
+    let event = RoomMemberEventContent {
+        is_direct: None,
+        membership: MembershipState::Leave,
+        third_party_invite: None,
+        reason: body.reason.clone(),
+        join_authorized_via_users_server: None,
+        ..event
+    };
 
     let mutex_state = Arc::clone(
         services()
@@ -236,6 +241,7 @@ pub async fn kick_user_route(
                 unsigned: None,
                 state_key: Some(body.user_id.to_string()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &body.room_id,
@@ -254,17 +260,7 @@ pub async fn kick_user_route(
 pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_user::v3::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    if let Ok(Some(membership_event)) = services()
-        .rooms
-        .state_accessor
-        .get_member(&body.room_id, sender_user)
-    {
-        if membership_event.membership == MembershipState::Ban {
-            return Ok(ban_user::v3::Response {});
-        }
-    }
-
-    let event = services()
+    let event = if let Some(event) = services()
         .rooms
         .state_accessor
         .room_state_get(
@@ -272,27 +268,30 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
             &StateEventType::RoomMember,
             body.user_id.as_ref(),
         )?
-        .map_or(
-            Ok(RoomMemberEventContent {
+        // Even when the previous member content is invalid, we should let the ban go through anyways.
+        .and_then(|event| serde_json::from_str::<RoomMemberEventContent>(event.content.get()).ok())
+    {
+        // If they are already banned and the reason is unchanged, there isn't any point in sending a new event.
+        if event.membership == MembershipState::Ban && event.reason == body.reason {
+            return Ok(ban_user::v3::Response {});
+        }
+
+        RoomMemberEventContent {
             membership: MembershipState::Ban,
-                displayname: services().users.displayname(&body.user_id)?,
-                avatar_url: services().users.avatar_url(&body.user_id)?,
-                is_direct: None,
-                third_party_invite: None,
-                blurhash: services().users.blurhash(&body.user_id)?,
+            join_authorized_via_users_server: None,
             reason: body.reason.clone(),
-                join_authorized_via_users_server: None,
-            }),
-            |event| {
-                serde_json::from_str(event.content.get())
-                    .map(|event: RoomMemberEventContent| RoomMemberEventContent {
-                        membership: MembershipState::Ban,
-                        join_authorized_via_users_server: None,
-                        ..event
-                    })
-                    .map_err(|_| Error::bad_database("Invalid member event in database."))
-            },
-        )?;
+            third_party_invite: None,
+            is_direct: None,
+            avatar_url: event.avatar_url,
+            displayname: event.displayname,
+            blurhash: event.blurhash,
+        }
+    } else {
+        RoomMemberEventContent {
+            reason: body.reason.clone(),
+            ..RoomMemberEventContent::new(MembershipState::Ban)
+        }
+    };
 
     let mutex_state = Arc::clone(
         services()
@@ -315,6 +314,7 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
                 unsigned: None,
                 state_key: Some(body.user_id.to_string()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &body.room_id,
@@ -335,17 +335,7 @@ pub async fn unban_user_route(
 ) -> Result<unban_user::v3::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    if let Ok(Some(membership_event)) = services()
-        .rooms
-        .state_accessor
-        .get_member(&body.room_id, sender_user)
-    {
-        if membership_event.membership != MembershipState::Ban {
-            return Ok(unban_user::v3::Response {});
-        }
-    }
-
-    let mut event: RoomMemberEventContent = serde_json::from_str(
+    let event: RoomMemberEventContent = serde_json::from_str(
         services()
             .rooms
             .state_accessor
@@ -363,8 +353,19 @@ pub async fn unban_user_route(
     )
     .map_err(|_| Error::bad_database("Invalid member event in database."))?;
 
-    event.membership = MembershipState::Leave;
-    event.reason.clone_from(&body.reason);
+    // If they are already unbanned and the reason is unchanged, there isn't any point in sending a new event.
+    if event.membership == MembershipState::Leave && event.reason == body.reason {
+        return Ok(unban_user::v3::Response {});
+    }
+
+    let event = RoomMemberEventContent {
+        is_direct: None,
+        membership: MembershipState::Leave,
+        third_party_invite: None,
+        reason: body.reason.clone(),
+        join_authorized_via_users_server: None,
+        ..event
+    };
 
     let mutex_state = Arc::clone(
         services()
@@ -387,6 +388,7 @@ pub async fn unban_user_route(
                 unsigned: None,
                 state_key: Some(body.user_id.to_string()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &body.room_id,
@@ -454,7 +456,7 @@ pub async fn get_member_events_route(
         .user_can_see_state_events(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view this room.",
         ));
     }
@@ -489,7 +491,7 @@ pub async fn joined_members_route(
         .user_can_see_state_events(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view this room.",
         ));
     }
@@ -625,7 +627,7 @@ async fn join_room_by_id_helper(
         let event_id = format!(
             "${}",
             ruma::signatures::reference_hash(&join_event_stub, &room_version_id)
-                .expect("ruma can calculate reference hashes")
+                .expect("Event format validated when event was hashed")
         );
         let event_id = <&EventId>::try_from(event_id.as_str())
             .expect("ruma's reference hashes are valid event ids");
@@ -939,6 +941,7 @@ async fn join_room_by_id_helper(
                     unsigned: None,
                     state_key: Some(sender_user.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 room_id,
@@ -984,6 +987,8 @@ async fn join_room_by_id_helper(
                         .as_str()
                 })
                 .and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok());
+            let restricted_join = join_authorized_via_users_server.is_some();
+
             // TODO: Is origin needed?
             join_event_stub.insert(
                 "origin".to_owned(),
@@ -1030,7 +1035,7 @@ async fn join_room_by_id_helper(
                 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())
+            let event_id = OwnedEventId::try_from(event_id)
                 .expect("ruma's reference hashes are valid event ids");
 
             // Add event_id back
@@ -1055,46 +1060,35 @@ 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.",
-                            ));
-                        }
-                    };
+            let pdu = if let Some(signed_raw) = send_join_response.room_state.event {
+                let (signed_event_id, signed_pdu) =
+                    gen_event_id_canonical_json(&signed_raw, &room_version_id)?;
 
                 if signed_event_id != event_id {
-                    return Err(Error::BadRequest(
-                        ErrorKind::InvalidParam,
+                    return Err(Error::BadServerResponse(
                         "Server sent event with wrong event id",
                     ));
                 }
 
+                signed_pdu
+            } else if restricted_join {
+                return Err(Error::BadServerResponse(
+                    "No signed event was returned, despite just performing a restricted join",
+                ));
+            } else {
+                join_event
+            };
+
             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,
-                    )
+                .handle_incoming_pdu(&remote_server, &event_id, room_id, pdu, 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()))
@@ -1142,7 +1136,7 @@ async fn make_join_request(
 async fn validate_and_add_event_id(
     pdu: &RawJsonValue,
     room_version: &RoomVersionId,
-    pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+    pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
 ) -> Result<(OwnedEventId, CanonicalJsonObject)> {
     let mut value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
         error!("Invalid PDU in server response: {:?}: {:?}", pdu, e);
@@ -1151,7 +1145,7 @@ async fn validate_and_add_event_id(
     let event_id = EventId::parse(format!(
         "${}",
         ruma::signatures::reference_hash(&value, room_version)
-            .expect("ruma can calculate reference hashes")
+            .map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Invalid PDU format"))?
     ))
     .expect("ruma's reference hashes are valid event ids");
 
@@ -1189,8 +1183,35 @@ async fn validate_and_add_event_id(
         }
     }
 
-    if let Err(e) = ruma::signatures::verify_event(&*pub_key_map.read().await, &value, room_version)
-    {
+    let origin_server_ts = value.get("origin_server_ts").ok_or_else(|| {
+        error!("Invalid PDU, no origin_server_ts field");
+        Error::BadRequest(
+            ErrorKind::MissingParam,
+            "Invalid PDU, no origin_server_ts field",
+        )
+    })?;
+
+    let origin_server_ts: MilliSecondsSinceUnixEpoch = {
+        let ts = origin_server_ts.as_integer().ok_or_else(|| {
+            Error::BadRequest(
+                ErrorKind::InvalidParam,
+                "origin_server_ts must be an integer",
+            )
+        })?;
+
+        MilliSecondsSinceUnixEpoch(i64::from(ts).try_into().map_err(|_| {
+            Error::BadRequest(ErrorKind::InvalidParam, "Time must be after the unix epoch")
+        })?)
+    };
+
+    let unfiltered_keys = (*pub_key_map.read().await).clone();
+
+    let keys =
+        services()
+            .globals
+            .filter_keys_server_map(unfiltered_keys, origin_server_ts, room_version);
+
+    if let Err(e) = ruma::signatures::verify_event(&keys, &value, room_version) {
         warn!("Event {} failed verification {:?} {}", event_id, pdu, e);
         back_off(event_id).await;
         return Err(Error::BadServerResponse("Event failed verification."));
@@ -1243,6 +1264,7 @@ pub(crate) async fn invite_helper<'a>(
                     unsigned: None,
                     state_key: Some(user_id.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 room_id,
@@ -1319,17 +1341,14 @@ pub(crate) async fn invite_helper<'a>(
             .filter(|server| &**server != services().globals.server_name());
 
         services().sending.send_pdu(servers, &pdu_id)?;
-
-        return Ok(());
-    }
-
+    } else {
         if !services()
             .rooms
             .state_cache
             .is_joined(sender_user, room_id)?
         {
             return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+                ErrorKind::forbidden(),
                 "You don't have permission to view this room.",
             ));
         }
@@ -1365,6 +1384,7 @@ pub(crate) async fn invite_helper<'a>(
                     unsigned: None,
                     state_key: Some(user_id.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 room_id,
@@ -1372,7 +1392,9 @@ pub(crate) async fn invite_helper<'a>(
             )
             .await?;
 
+        // Critical point ends
         drop(state_lock);
+    }
 
     Ok(())
 }
@@ -1470,12 +1492,15 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId, reason: Option<Strin
             Some(e) => e,
         };
 
-        let mut event: RoomMemberEventContent = serde_json::from_str(member_event.content.get())
-            .map_err(|_| Error::bad_database("Invalid member event in database."))?;
-
-        event.membership = MembershipState::Leave;
-        event.reason = reason;
-        event.join_authorized_via_users_server = None;
+        let event = RoomMemberEventContent {
+            is_direct: None,
+            membership: MembershipState::Leave,
+            third_party_invite: None,
+            reason,
+            join_authorized_via_users_server: None,
+            ..serde_json::from_str(member_event.content.get())
+                .map_err(|_| Error::bad_database("Invalid member event in database."))?
+        };
 
         services()
             .rooms
@@ -1487,6 +1512,7 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId, reason: Option<Strin
                     unsigned: None,
                     state_key: Some(user_id.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
                 user_id,
                 room_id,
@@ -1588,7 +1614,7 @@ async fn remote_leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
     let event_id = EventId::parse(format!(
         "${}",
         ruma::signatures::reference_hash(&leave_event_stub, &room_version_id)
-            .expect("ruma can calculate reference hashes")
+            .expect("Event format validated when event was hashed")
     ))
     .expect("ruma's reference hashes are valid event ids");
 

src/api/client_server/message.rs

@@ -43,7 +43,7 @@ pub async fn send_message_event_route(
         && !services().globals.allow_encryption()
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Encryption has been disabled",
         ));
     }
@@ -84,6 +84,11 @@ pub async fn send_message_event_route(
                 unsigned: Some(unsigned),
                 state_key: None,
                 redacts: None,
+                timestamp: if body.appservice_info.is_some() {
+                    body.timestamp
+                } else {
+                    None
+                },
             },
             sender_user,
             &body.room_id,

src/api/client_server/mod.rs

@@ -11,6 +11,7 @@ mod keys;
 mod media;
 mod membership;
 mod message;
+mod openid;
 mod presence;
 mod profile;
 mod push;
@@ -47,6 +48,7 @@ pub use keys::*;
 pub use media::*;
 pub use membership::*;
 pub use message::*;
+pub use openid::*;
 pub use presence::*;
 pub use profile::*;
 pub use push::*;

src/api/client_server/openid.rs

@@ -0,0 +1,23 @@
+use std::time::Duration;
+
+use ruma::{api::client::account, authentication::TokenType};
+
+use crate::{services, Result, Ruma};
+
+/// # `POST /_matrix/client/r0/user/{userId}/openid/request_token`
+///
+/// Request an OpenID token to verify identity with third-party services.
+///
+/// - The token generated is only valid for the OpenID API.
+pub async fn create_openid_token_route(
+    body: Ruma<account::request_openid_token::v3::Request>,
+) -> Result<account::request_openid_token::v3::Response> {
+    let (access_token, expires_in) = services().users.create_openid_token(&body.user_id)?;
+
+    Ok(account::request_openid_token::v3::Response {
+        access_token,
+        token_type: TokenType::Bearer,
+        matrix_server_name: services().globals.server_name().to_owned(),
+        expires_in: Duration::from_secs(expires_in),
+    })
+}

src/api/client_server/profile.rs

@@ -65,6 +65,7 @@ pub async fn set_displayname_route(
                     unsigned: None,
                     state_key: Some(sender_user.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
                 room_id,
             ))
@@ -200,6 +201,7 @@ pub async fn set_avatar_url_route(
                     unsigned: None,
                     state_key: Some(sender_user.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
                 room_id,
             ))

src/api/client_server/redact.rs

@@ -44,6 +44,7 @@ pub async fn redact_event_route(
                 unsigned: None,
                 state_key: None,
                 redacts: Some(body.event_id.into()),
+                timestamp: None,
             },
             sender_user,
             &body.room_id,

src/api/client_server/relations.rs

@@ -3,7 +3,7 @@ use ruma::api::client::relations::{
     get_relating_events_with_rel_type_and_event_type,
 };
 
-use crate::{service::rooms::timeline::PduCount, services, Result, Ruma};
+use crate::{services, Result, Ruma};
 
 /// # `GET /_matrix/client/r0/rooms/{roomId}/relations/{eventId}/{relType}/{eventType}`
 pub async fn get_relating_events_with_rel_type_and_event_type_route(
@@ -11,27 +11,6 @@ pub async fn get_relating_events_with_rel_type_and_event_type_route(
 ) -> Result<get_relating_events_with_rel_type_and_event_type::v1::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    let from = match body.from.clone() {
-        Some(from) => PduCount::try_from_string(&from)?,
-        None => match ruma::api::Direction::Backward {
-            // TODO: fix ruma so `body.dir` exists
-            ruma::api::Direction::Forward => PduCount::min(),
-            ruma::api::Direction::Backward => PduCount::max(),
-        },
-    };
-
-    let to = body
-        .to
-        .as_ref()
-        .and_then(|t| PduCount::try_from_string(t).ok());
-
-    // Use limit or else 10, with maximum 100
-    let limit = body
-        .limit
-        .and_then(|u| u32::try_from(u).ok())
-        .map_or(10_usize, |u| u as usize)
-        .min(100);
-
     let res = services()
         .rooms
         .pdu_metadata
@@ -41,9 +20,11 @@ pub async fn get_relating_events_with_rel_type_and_event_type_route(
             &body.event_id,
             Some(body.event_type.clone()),
             Some(body.rel_type.clone()),
-            from,
-            to,
-            limit,
+            body.from.clone(),
+            body.to.clone(),
+            body.limit,
+            body.recurse,
+            &body.dir,
         )?;
 
     Ok(
@@ -51,6 +32,7 @@ pub async fn get_relating_events_with_rel_type_and_event_type_route(
             chunk: res.chunk,
             next_batch: res.next_batch,
             prev_batch: res.prev_batch,
+            recursion_depth: res.recursion_depth,
         },
     )
 }
@@ -61,27 +43,6 @@ pub async fn get_relating_events_with_rel_type_route(
 ) -> Result<get_relating_events_with_rel_type::v1::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    let from = match body.from.clone() {
-        Some(from) => PduCount::try_from_string(&from)?,
-        None => match ruma::api::Direction::Backward {
-            // TODO: fix ruma so `body.dir` exists
-            ruma::api::Direction::Forward => PduCount::min(),
-            ruma::api::Direction::Backward => PduCount::max(),
-        },
-    };
-
-    let to = body
-        .to
-        .as_ref()
-        .and_then(|t| PduCount::try_from_string(t).ok());
-
-    // Use limit or else 10, with maximum 100
-    let limit = body
-        .limit
-        .and_then(|u| u32::try_from(u).ok())
-        .map_or(10_usize, |u| u as usize)
-        .min(100);
-
     let res = services()
         .rooms
         .pdu_metadata
@@ -91,15 +52,18 @@ pub async fn get_relating_events_with_rel_type_route(
             &body.event_id,
             None,
             Some(body.rel_type.clone()),
-            from,
-            to,
-            limit,
+            body.from.clone(),
+            body.to.clone(),
+            body.limit,
+            body.recurse,
+            &body.dir,
         )?;
 
     Ok(get_relating_events_with_rel_type::v1::Response {
         chunk: res.chunk,
         next_batch: res.next_batch,
         prev_batch: res.prev_batch,
+        recursion_depth: res.recursion_depth,
     })
 }
 
@@ -109,27 +73,6 @@ pub async fn get_relating_events_route(
 ) -> Result<get_relating_events::v1::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    let from = match body.from.clone() {
-        Some(from) => PduCount::try_from_string(&from)?,
-        None => match ruma::api::Direction::Backward {
-            // TODO: fix ruma so `body.dir` exists
-            ruma::api::Direction::Forward => PduCount::min(),
-            ruma::api::Direction::Backward => PduCount::max(),
-        },
-    };
-
-    let to = body
-        .to
-        .as_ref()
-        .and_then(|t| PduCount::try_from_string(t).ok());
-
-    // Use limit or else 10, with maximum 100
-    let limit = body
-        .limit
-        .and_then(|u| u32::try_from(u).ok())
-        .map_or(10_usize, |u| u as usize)
-        .min(100);
-
     services()
         .rooms
         .pdu_metadata
@@ -139,8 +82,10 @@ pub async fn get_relating_events_route(
             &body.event_id,
             None,
             None,
-            from,
-            to,
-            limit,
+            body.from.clone(),
+            body.to.clone(),
+            body.limit,
+            body.recurse,
+            &body.dir,
         )
 }

src/api/client_server/room.rs

@@ -72,7 +72,7 @@ pub async fn create_room_route(
         && !services().users.is_admin(sender_user)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Room creation has been disabled.",
         ));
     }
@@ -230,6 +230,7 @@ pub async fn create_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &room_id,
@@ -258,6 +259,7 @@ pub async fn create_room_route(
                 unsigned: None,
                 state_key: Some(sender_user.to_string()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &room_id,
@@ -311,6 +313,7 @@ pub async fn create_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &room_id,
@@ -334,6 +337,7 @@ pub async fn create_room_route(
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 &room_id,
@@ -360,6 +364,7 @@ pub async fn create_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &room_id,
@@ -381,6 +386,7 @@ pub async fn create_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &room_id,
@@ -403,6 +409,7 @@ pub async fn create_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &room_id,
@@ -447,6 +454,7 @@ pub async fn create_room_route(
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 &room_id,
@@ -469,6 +477,7 @@ pub async fn create_room_route(
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 &room_id,
@@ -485,7 +494,10 @@ pub async fn create_room_route(
 
     // Homeserver specific stuff
     if let Some(alias) = alias {
-        services().rooms.alias.set_alias(&alias, &room_id)?;
+        services()
+            .rooms
+            .alias
+            .set_alias(&alias, &room_id, sender_user)?;
     }
 
     if body.visibility == room::Visibility::Public {
@@ -522,7 +534,7 @@ pub async fn get_room_event_route(
         &body.event_id,
     )? {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view this event.",
         ));
     }
@@ -551,7 +563,7 @@ pub async fn get_room_aliases_route(
         .is_joined(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view this room.",
         ));
     }
@@ -626,6 +638,7 @@ pub async fn upgrade_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &body.room_id,
@@ -727,6 +740,7 @@ pub async fn upgrade_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &replacement_room,
@@ -755,6 +769,7 @@ pub async fn upgrade_room_route(
                 unsigned: None,
                 state_key: Some(sender_user.to_string()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &replacement_room,
@@ -797,6 +812,7 @@ pub async fn upgrade_room_route(
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
                 sender_user,
                 &replacement_room,
@@ -815,7 +831,7 @@ pub async fn upgrade_room_route(
         services()
             .rooms
             .alias
-            .set_alias(&alias, &replacement_room)?;
+            .set_alias(&alias, &replacement_room, sender_user)?;
     }
 
     // Get the old room power levels
@@ -847,6 +863,7 @@ pub async fn upgrade_room_route(
                 unsigned: None,
                 state_key: Some("".to_owned()),
                 redacts: None,
+                timestamp: None,
             },
             sender_user,
             &body.room_id,

src/api/client_server/search.rs

@@ -43,7 +43,7 @@ pub async fn search_events_route(
             .is_joined(sender_user, &room_id)?
         {
             return Err(Error::BadRequest(
-                ErrorKind::Forbidden,
+                ErrorKind::forbidden(),
                 "You don't have permission to view this room.",
             ));
         }
@@ -89,7 +89,8 @@ pub async fn search_events_route(
                 .get_pdu_from_id(result)
                 .ok()?
                 .filter(|pdu| {
-                    services()
+                    !pdu.is_redacted()
+                        && services()
                             .rooms
                             .state_accessor
                             .user_can_see_event(sender_user, &pdu.room_id, &pdu.event_id)

src/api/client_server/session.rs

@@ -63,7 +63,7 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
                 UserId::parse(user)
             } else {
                 warn!("Bad login type: {:?}", &body.login_info);
-                return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type."));
+                return Err(Error::BadRequest(ErrorKind::forbidden(), "Bad login type."));
             }
             .map_err(|_| Error::BadRequest(ErrorKind::InvalidUsername, "Username is invalid."))?;
 
@@ -78,7 +78,7 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
                 .users
                 .password_hash(&user_id)?
                 .ok_or(Error::BadRequest(
-                    ErrorKind::Forbidden,
+                    ErrorKind::forbidden(),
                     "Wrong username or password.",
                 ))?;
 
@@ -93,7 +93,7 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
 
             if !hash_matches {
                 return Err(Error::BadRequest(
-                    ErrorKind::Forbidden,
+                    ErrorKind::forbidden(),
                     "Wrong username or password.",
                 ));
             }
@@ -143,7 +143,7 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
                 UserId::parse(user)
             } else {
                 warn!("Bad login type: {:?}", &body.login_info);
-                return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type."));
+                return Err(Error::BadRequest(ErrorKind::forbidden(), "Bad login type."));
             }
             .map_err(|_| Error::BadRequest(ErrorKind::InvalidUsername, "Username is invalid."))?;
 

src/api/client_server/state.rs

@@ -10,7 +10,7 @@ use ruma::{
         room::canonical_alias::RoomCanonicalAliasEventContent, AnyStateEventContent, StateEventType,
     },
     serde::Raw,
-    EventId, RoomId, UserId,
+    EventId, MilliSecondsSinceUnixEpoch, RoomId, UserId,
 };
 use tracing::log::warn;
 
@@ -32,6 +32,11 @@ pub async fn send_state_event_for_key_route(
         &body.event_type,
         &body.body.body, // Yes, I hate it too
         body.state_key.to_owned(),
+        if body.appservice_info.is_some() {
+            body.timestamp
+        } else {
+            None
+        },
     )
     .await?;
 
@@ -54,7 +59,7 @@ pub async fn send_state_event_for_empty_key_route(
     // Forbid m.room.encryption if encryption is disabled
     if body.event_type == StateEventType::RoomEncryption && !services().globals.allow_encryption() {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Encryption has been disabled",
         ));
     }
@@ -65,6 +70,11 @@ pub async fn send_state_event_for_empty_key_route(
         &body.event_type.to_string().into(),
         &body.body.body,
         body.state_key.to_owned(),
+        if body.appservice_info.is_some() {
+            body.timestamp
+        } else {
+            None
+        },
     )
     .await?;
 
@@ -88,7 +98,7 @@ pub async fn get_state_events_route(
         .user_can_see_state_events(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view the room state.",
         ));
     }
@@ -121,7 +131,7 @@ pub async fn get_state_events_for_key_route(
         .user_can_see_state_events(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view the room state.",
         ));
     }
@@ -160,7 +170,7 @@ pub async fn get_state_events_for_empty_key_route(
         .user_can_see_state_events(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You don't have permission to view the room state.",
         ));
     }
@@ -190,6 +200,7 @@ async fn send_state_event_for_key_helper(
     event_type: &StateEventType,
     json: &Raw<AnyStateEventContent>,
     state_key: String,
+    timestamp: Option<MilliSecondsSinceUnixEpoch>,
 ) -> Result<Arc<EventId>> {
     let sender_user = sender;
 
@@ -214,7 +225,7 @@ async fn send_state_event_for_key_helper(
                     .is_none()
             {
                 return Err(Error::BadRequest(
-                    ErrorKind::Forbidden,
+                    ErrorKind::forbidden(),
                     "You are only allowed to send canonical_alias \
                     events when it's aliases already exists",
                 ));
@@ -243,6 +254,7 @@ async fn send_state_event_for_key_helper(
                 unsigned: None,
                 state_key: Some(state_key),
                 redacts: None,
+                timestamp,
             },
             sender_user,
             room_id,

src/api/client_server/sync.rs

@@ -12,7 +12,7 @@ use ruma::{
                 Ephemeral, Filter, GlobalAccountData, InviteState, InvitedRoom, JoinedRoom,
                 LeftRoom, Presence, RoomAccountData, RoomSummary, Rooms, State, Timeline, ToDevice,
             },
-            v4::SlidingOp,
+            v4::{SlidingOp, SlidingSyncRoomHero},
             DeviceLists, UnreadNotificationsCount,
         },
         uiaa::UiaaResponse,
@@ -716,7 +716,7 @@ async fn load_joined_room(
                                         .state_cache
                                         .is_invited(&user_id, room_id)?)
                                 {
-                                    Ok::<_, Error>(Some(state_key.clone()))
+                                    Ok::<_, Error>(Some(user_id))
                                 } else {
                                     Ok(None)
                                 }
@@ -1572,7 +1572,7 @@ pub async fn sync_events_v4_route(
             sender_user.clone(),
             sender_device.clone(),
             conn_id.clone(),
-            body.room_subscriptions,
+            body.room_subscriptions.clone(),
         );
     }
 
@@ -1638,33 +1638,37 @@ pub async fn sync_events_v4_route(
                     .get_member(room_id, &member)
                     .ok()
                     .flatten()
-                    .map(|memberevent| {
-                        (
-                            memberevent
-                                .displayname
-                                .unwrap_or_else(|| member.to_string()),
-                            memberevent.avatar_url,
-                        )
+                    .map(|memberevent| SlidingSyncRoomHero {
+                        user_id: member,
+                        name: memberevent.displayname,
+                        avatar: memberevent.avatar_url,
                     })
             })
             .take(5)
             .collect::<Vec<_>>();
         let name = match &heroes[..] {
             [] => None,
-            [only] => Some(only.0.clone()),
+            [only] => Some(
+                only.name
+                    .clone()
+                    .unwrap_or_else(|| only.user_id.to_string()),
+            ),
             [firsts @ .., last] => Some(
                 firsts
                     .iter()
-                    .map(|h| h.0.clone())
+                    .map(|h| h.name.clone().unwrap_or_else(|| h.user_id.to_string()))
                     .collect::<Vec<_>>()
                     .join(", ")
                     + " and "
-                    + &last.0,
+                    + &last
+                        .name
+                        .clone()
+                        .unwrap_or_else(|| last.user_id.to_string()),
             ),
         };
 
         let avatar = if let [only] = &heroes[..] {
-            only.1.clone()
+            only.avatar.clone()
         } else {
             None
         };
@@ -1725,6 +1729,16 @@ pub async fn sync_events_v4_route(
                 ),
                 num_live: None, // Count events in timeline greater than global sync counter
                 timestamp: None,
+                heroes: if body
+                    .room_subscriptions
+                    .get(room_id)
+                    .map(|sub| sub.include_heroes.unwrap_or_default())
+                    .unwrap_or_default()
+                {
+                    Some(heroes)
+                } else {
+                    None
+                },
             },
         );
     }

src/api/client_server/typing.rs

@@ -17,7 +17,7 @@ pub async fn create_typing_event_route(
         .is_joined(sender_user, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "You are not in this room.",
         ));
     }

src/api/client_server/unversioned.rs

@@ -27,7 +27,10 @@ pub async fn get_supported_versions_route(
             "v1.4".to_owned(),
             "v1.5".to_owned(),
         ],
-        unstable_features: BTreeMap::from_iter([("org.matrix.e2e_cross_signing".to_owned(), true)]),
+        unstable_features: BTreeMap::from_iter([
+            ("org.matrix.e2e_cross_signing".to_owned(), true),
+            ("org.matrix.msc3916.stable".to_owned(), true),
+        ]),
     };
 
     Ok(resp)

src/api/ruma_wrapper/axum.rs

@@ -2,20 +2,22 @@ use std::{collections::BTreeMap, iter::FromIterator, str};
 
 use axum::{
     async_trait,
-    body::{Full, HttpBody},
-    extract::{rejection::TypedHeaderRejectionReason, FromRequest, Path, TypedHeader},
-    headers::{
-        authorization::{Bearer, Credentials},
-        Authorization,
-    },
+    body::Body,
+    extract::{FromRequest, Path},
     response::{IntoResponse, Response},
-    BoxError, RequestExt, RequestPartsExt,
+    RequestExt, RequestPartsExt,
 };
-use bytes::{Buf, BufMut, Bytes, BytesMut};
+use axum_extra::{
+    headers::{authorization::Bearer, Authorization},
+    typed_header::TypedHeaderRejectionReason,
+    TypedHeader,
+};
+use bytes::{BufMut, BytesMut};
 use http::{Request, StatusCode};
 use ruma::{
     api::{client::error::ErrorKind, AuthScheme, IncomingRequest, OutgoingResponse},
-    CanonicalJsonValue, OwnedDeviceId, OwnedServerName, OwnedUserId, UserId,
+    server_util::authorization::XMatrix,
+    CanonicalJsonValue, MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId,
 };
 use serde::Deserialize;
 use tracing::{debug, error, warn};
@@ -31,37 +33,33 @@ enum Token {
 }
 
 #[async_trait]
-impl<T, S, B> FromRequest<S, B> for Ruma<T>
+impl<T, S> FromRequest<S> for Ruma<T>
 where
     T: IncomingRequest,
-    B: HttpBody + Send + 'static,
-    B::Data: Send,
-    B::Error: Into<BoxError>,
 {
     type Rejection = Error;
 
-    async fn from_request(req: Request<B>, _state: &S) -> Result<Self, Self::Rejection> {
+    async fn from_request(req: Request<Body>, _state: &S) -> Result<Self, Self::Rejection> {
         #[derive(Deserialize)]
         struct QueryParams {
             access_token: Option<String>,
             user_id: Option<String>,
         }
 
-        let (mut parts, mut body) = match req.with_limited_body() {
-            Ok(limited_req) => {
+        let (mut parts, mut body) = {
+            let limited_req = req.with_limited_body();
             let (parts, body) = limited_req.into_parts();
-                let body = to_bytes(body)
+            let body = axum::body::to_bytes(
+                body,
+                services()
+                    .globals
+                    .max_request_size()
+                    .try_into()
+                    .unwrap_or(usize::MAX),
+            )
             .await
             .map_err(|_| Error::BadRequest(ErrorKind::MissingToken, "Missing token."))?;
             (parts, body)
-            }
-            Err(original_req) => {
-                let (parts, body) = original_req.into_parts();
-                let body = to_bytes(body)
-                    .await
-                    .map_err(|_| Error::BadRequest(ErrorKind::MissingToken, "Missing token."))?;
-                (parts, body)
-            }
         };
 
         let metadata = T::METADATA;
@@ -102,10 +100,15 @@ where
         let (sender_user, sender_device, sender_servername, appservice_info) =
             match (metadata.authentication, token) {
                 (_, Token::Invalid) => {
+                    // OpenID endpoint uses a query param with the same name, drop this once query params for user auth are removed from the spec
+                    if query_params.access_token.is_some() {
+                        (None, None, None, None)
+                    } else {
                         return Err(Error::BadRequest(
                             ErrorKind::UnknownToken { soft_logout: false },
                             "Unknown access token.",
-                    ))
+                        ));
+                    }
                 }
                 (AuthScheme::AccessToken, Token::Appservice(info)) => {
                     let user_id = query_params
@@ -132,7 +135,7 @@ where
 
                     if !services().users.exists(&user_id)? {
                         return Err(Error::BadRequest(
-                            ErrorKind::Forbidden,
+                            ErrorKind::forbidden(),
                             "User does not exist.",
                         ));
                     }
@@ -172,7 +175,7 @@ where
                                 _ => "Unknown header-related error",
                             };
 
-                            Error::BadRequest(ErrorKind::Forbidden, msg)
+                            Error::BadRequest(ErrorKind::forbidden(), msg)
                         })?;
 
                     if let Some(dest) = x_matrix.destination {
@@ -186,12 +189,17 @@ where
 
                     let origin_signatures = BTreeMap::from_iter([(
                         x_matrix.key.clone(),
-                        CanonicalJsonValue::String(x_matrix.sig),
+                        CanonicalJsonValue::String(x_matrix.sig.to_string()),
                     )]);
 
                     let signatures = BTreeMap::from_iter([(
                         x_matrix.origin.as_str().to_owned(),
-                        CanonicalJsonValue::Object(origin_signatures),
+                        CanonicalJsonValue::Object(
+                            origin_signatures
+                                .into_iter()
+                                .map(|(k, v)| (k.to_string(), v))
+                                .collect(),
+                        ),
                     )]);
 
                     let mut request_map = BTreeMap::from_iter([
@@ -226,7 +234,7 @@ where
                     let keys_result = services()
                         .rooms
                         .event_handler
-                        .fetch_signing_keys(&x_matrix.origin, vec![x_matrix.key.to_owned()])
+                        .fetch_signing_keys(&x_matrix.origin, vec![x_matrix.key.to_string()], false)
                         .await;
 
                     let keys = match keys_result {
@@ -234,14 +242,25 @@ where
                         Err(e) => {
                             warn!("Failed to fetch signing keys: {}", e);
                             return Err(Error::BadRequest(
-                                ErrorKind::Forbidden,
+                                ErrorKind::forbidden(),
                                 "Failed to fetch signing keys.",
                             ));
                         }
                     };
 
-                    let pub_key_map =
-                        BTreeMap::from_iter([(x_matrix.origin.as_str().to_owned(), keys)]);
+                    // Only verify_keys that are currently valid should be used for validating requests
+                    // as per MSC4029
+                    let pub_key_map = BTreeMap::from_iter([(
+                        x_matrix.origin.as_str().to_owned(),
+                        if keys.valid_until_ts > MilliSecondsSinceUnixEpoch::now() {
+                            keys.verify_keys
+                                .into_iter()
+                                .map(|(id, key)| (id, key.key))
+                                .collect()
+                        } else {
+                            BTreeMap::new()
+                        },
+                    )]);
 
                     match ruma::signatures::verify_json(&pub_key_map, &request_map) {
                         Ok(()) => (None, None, Some(x_matrix.origin), None),
@@ -260,7 +279,7 @@ where
                             }
 
                             return Err(Error::BadRequest(
-                                ErrorKind::Forbidden,
+                                ErrorKind::forbidden(),
                                 "Failed to verify X-Matrix signatures.",
                             ));
                         }
@@ -340,124 +359,11 @@ where
     }
 }
 
-struct XMatrix {
-    destination: Option<OwnedServerName>,
-    origin: OwnedServerName,
-    key: String, // KeyName?
-    sig: String,
-}
-
-impl Credentials for XMatrix {
-    const SCHEME: &'static str = "X-Matrix";
-
-    fn decode(value: &http::HeaderValue) -> Option<Self> {
-        debug_assert!(
-            value.as_bytes().starts_with(b"X-Matrix "),
-            "HeaderValue to decode should start with \"X-Matrix ..\", received = {value:?}",
-        );
-
-        let parameters = str::from_utf8(&value.as_bytes()["X-Matrix ".len()..])
-            .ok()?
-            .trim_start();
-
-        let mut origin = None;
-        let mut key = None;
-        let mut sig = None;
-        let mut destination = None;
-
-        for entry in parameters.split_terminator(',') {
-            let (name, value) = entry.split_once('=')?;
-
-            // It's not at all clear why some fields are quoted and others not in the spec,
-            // let's simply accept either form for every field.
-            let value = value
-                .strip_prefix('"')
-                .and_then(|rest| rest.strip_suffix('"'))
-                .unwrap_or(value);
-
-            // FIXME: Catch multiple fields of the same name
-            match name {
-                "origin" => origin = Some(value.try_into().ok()?),
-                "key" => key = Some(value.to_owned()),
-                "sig" => sig = Some(value.to_owned()),
-                "destination" => destination = Some(value.try_into().ok()?),
-                _ => debug!(
-                    "Unexpected field `{}` in X-Matrix Authorization header",
-                    name
-                ),
-            }
-        }
-
-        Some(Self {
-            destination,
-            origin: origin?,
-            key: key?,
-            sig: sig?,
-        })
-    }
-
-    fn encode(&self) -> http::HeaderValue {
-        todo!()
-    }
-}
-
 impl<T: OutgoingResponse> IntoResponse for RumaResponse<T> {
     fn into_response(self) -> Response {
         match self.0.try_into_http_response::<BytesMut>() {
-            Ok(res) => res.map(BytesMut::freeze).map(Full::new).into_response(),
+            Ok(res) => res.map(BytesMut::freeze).map(Body::from).into_response(),
             Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
         }
     }
 }
-
-// copied from hyper under the following license:
-// Copyright (c) 2014-2021 Sean McArthur
-
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-pub(crate) async fn to_bytes<T>(body: T) -> Result<Bytes, T::Error>
-where
-    T: HttpBody,
-{
-    futures_util::pin_mut!(body);
-
-    // If there's only 1 chunk, we can just return Buf::to_bytes()
-    let mut first = if let Some(buf) = body.data().await {
-        buf?
-    } else {
-        return Ok(Bytes::new());
-    };
-
-    let second = if let Some(buf) = body.data().await {
-        buf?
-    } else {
-        return Ok(first.copy_to_bytes(first.remaining()));
-    };
-
-    // With more than 1 buf, we gotta flatten into a Vec first.
-    let cap = first.remaining() + second.remaining() + body.size_hint().lower() as usize;
-    let mut vec = Vec::with_capacity(cap);
-    vec.put(first);
-    vec.put(second);
-
-    while let Some(buf) = body.data().await {
-        vec.put(buf?);
-    }
-
-    Ok(vec.into())
-}

src/api/server_server.rs

@@ -2,17 +2,25 @@
 
 use crate::{
     api::client_server::{self, claim_keys_helper, get_keys_helper},
-    service::pdu::{gen_event_id_canonical_json, PduBuilder},
+    service::{
+        globals::SigningKeys,
+        media::FileMeta,
+        pdu::{gen_event_id_canonical_json, PduBuilder},
+    },
     services, utils, Error, PduEvent, Result, Ruma,
 };
 use axum::{response::IntoResponse, Json};
+use axum_extra::headers::{CacheControl, Header};
 use get_profile_information::v1::ProfileField;
-use http::header::{HeaderValue, AUTHORIZATION};
+use http::header::AUTHORIZATION;
 
 use ruma::{
     api::{
         client::error::{Error as RumaError, ErrorKind},
         federation::{
+            authenticated_media::{
+                get_content, get_content_thumbnail, Content, ContentMetadata, FileOrLocation,
+            },
             authorization::get_event_authorization,
             backfill::get_backfill,
             device::get_devices::{self, v1::UserDevice},
@@ -24,6 +32,7 @@ use ruma::{
             event::{get_event, get_missing_events, get_room_state, get_room_state_ids},
             keys::{claim_keys, get_keys},
             membership::{create_invite, create_join_event, prepare_join_event},
+            openid::get_openid_userinfo,
             query::{get_profile_information, get_room_information},
             transactions::{
                 edu::{DeviceListUpdateContent, DirectDeviceContent, Edu, SigningKeyUpdateContent},
@@ -93,13 +102,6 @@ impl FedDest {
         }
     }
 
-    fn into_uri_string(self) -> String {
-        match self {
-            Self::Literal(addr) => addr.to_string(),
-            Self::Named(host, ref port) => host + port,
-        }
-    }
-
     fn hostname(&self) -> String {
         match &self {
             Self::Literal(addr) => addr.ip().to_string(),
@@ -135,8 +137,6 @@ where
 
     debug!("Preparing to send request to {destination}");
 
-    let mut write_destination_to_cache = false;
-
     let cached_result = services()
         .globals
         .actual_destination_cache
@@ -145,14 +145,63 @@ where
         .get(destination)
         .cloned();
 
-    let (actual_destination, host) = if let Some(result) = cached_result {
-        result
+    let actual_destination = if let Some(DestinationResponse {
+        actual_destination,
+        dest_type,
+    }) = cached_result
+    {
+        match dest_type {
+            DestType::IsIpOrHasPort => actual_destination,
+            DestType::LookupFailed {
+                well_known_retry,
+                well_known_backoff_mins,
+            } => {
+                if well_known_retry < Instant::now() {
+                    find_actual_destination(destination, None, false, Some(well_known_backoff_mins))
+                        .await
                 } else {
-        write_destination_to_cache = true;
+                    actual_destination
+                }
+            }
 
-        let result = find_actual_destination(destination).await;
-
-        (result.0, result.1.into_uri_string())
+            DestType::WellKnown { expires } => {
+                if expires < Instant::now() {
+                    find_actual_destination(destination, None, false, None).await
+                } else {
+                    actual_destination
+                }
+            }
+            DestType::WellKnownSrv {
+                srv_expires,
+                well_known_expires,
+                well_known_host,
+            } => {
+                if well_known_expires < Instant::now() {
+                    find_actual_destination(destination, None, false, None).await
+                } else if srv_expires < Instant::now() {
+                    find_actual_destination(destination, Some(well_known_host), true, None).await
+                } else {
+                    actual_destination
+                }
+            }
+            DestType::Srv {
+                well_known_retry,
+                well_known_backoff_mins,
+                srv_expires,
+            } => {
+                if well_known_retry < Instant::now() {
+                    find_actual_destination(destination, None, false, Some(well_known_backoff_mins))
+                        .await
+                } else if srv_expires < Instant::now() {
+                    find_actual_destination(destination, None, true, Some(well_known_backoff_mins))
+                        .await
+                } else {
+                    actual_destination
+                }
+            }
+        }
+    } else {
+        find_actual_destination(destination, None, false, None).await
     };
 
     let actual_destination_str = actual_destination.clone().into_https_string();
@@ -161,7 +210,7 @@ where
         .try_into_http_request::<Vec<u8>>(
             &actual_destination_str,
             SendAccessToken::IfRequired(""),
-            &[MatrixVersion::V1_4],
+            &[MatrixVersion::V1_11],
         )
         .map_err(|e| {
             warn!(
@@ -225,13 +274,14 @@ where
         for s in signature_server {
             http_request.headers_mut().insert(
                 AUTHORIZATION,
-                HeaderValue::from_str(&format!(
+                format!(
                     "X-Matrix origin=\"{}\",destination=\"{}\",key=\"{}\",sig=\"{}\"",
                     services().globals.server_name(),
                     destination,
                     s.0,
                     s.1
-                ))
+                )
+                .try_into()
                 .unwrap(),
             );
         }
@@ -289,21 +339,10 @@ where
             if status == 200 {
                 debug!("Parsing response bytes from {destination}");
                 let response = T::IncomingResponse::try_from_http_response(http_response);
-                if response.is_ok() && write_destination_to_cache {
-                    services()
-                        .globals
-                        .actual_destination_cache
-                        .write()
-                        .await
-                        .insert(
-                            OwnedServerName::from(destination),
-                            (actual_destination, host),
-                        );
-                }
 
                 response.map_err(|e| {
                     warn!(
-                        "Invalid 200 response from {} on: {} {}",
+                        "Invalid 200 response from {} on: {} {:?}",
                         &destination, url, e
                     );
                     Error::BadServerResponse("Server returned bad 200 response.")
@@ -344,42 +383,175 @@ fn add_port_to_hostname(destination_str: &str) -> FedDest {
     FedDest::Named(host.to_owned(), port.to_owned())
 }
 
-/// Returns: actual_destination, host header
-/// Implemented according to the specification at <https://matrix.org/docs/spec/server_server/r0.1.4#resolving-server-names>
+#[derive(Clone)]
+pub struct DestinationResponse {
+    pub actual_destination: FedDest,
+    pub dest_type: DestType,
+}
+
+#[derive(Clone)]
+pub enum DestType {
+    WellKnownSrv {
+        srv_expires: Instant,
+        well_known_expires: Instant,
+        well_known_host: String,
+    },
+    WellKnown {
+        expires: Instant,
+    },
+    Srv {
+        srv_expires: Instant,
+        well_known_retry: Instant,
+        well_known_backoff_mins: u16,
+    },
+    IsIpOrHasPort,
+    LookupFailed {
+        well_known_retry: Instant,
+        well_known_backoff_mins: u16,
+    },
+}
+
+/// Implemented according to the specification at <https://spec.matrix.org/v1.11/server-server-api/#resolving-server-names>
 /// Numbers in comments below refer to bullet points in linked section of specification
-async fn find_actual_destination(destination: &'_ ServerName) -> (FedDest, FedDest) {
+async fn find_actual_destination(
+    destination: &'_ ServerName,
+    // The host used to potentially lookup SRV records against, only used when only_request_srv is true
+    well_known_dest: Option<String>,
+    // Should be used when only the SRV lookup has expired
+    only_request_srv: bool,
+    // The backoff time for the last well known failure, if any
+    well_known_backoff_mins: Option<u16>,
+) -> FedDest {
     debug!("Finding actual destination for {destination}");
-    let destination_str = destination.as_str().to_owned();
-    let mut hostname = destination_str.clone();
-    let actual_destination = match get_ip_with_port(&destination_str) {
+    let destination_str = destination.to_string();
+    let next_backoff_mins = well_known_backoff_mins
+        // Errors are recommended to be cached for up to an hour
+        .map(|mins| (mins * 2).min(60))
+        .unwrap_or(1);
+
+    let (actual_destination, dest_type) = if only_request_srv {
+        let destination_str = well_known_dest.unwrap_or(destination_str);
+        let (dest, expires) = get_srv_destination(destination_str).await;
+        let well_known_retry =
+            Instant::now() + Duration::from_secs((60 * next_backoff_mins).into());
+        (
+            dest,
+            if let Some(expires) = expires {
+                DestType::Srv {
+                    well_known_backoff_mins: next_backoff_mins,
+                    srv_expires: expires,
+
+                    well_known_retry,
+                }
+            } else {
+                DestType::LookupFailed {
+                    well_known_retry,
+                    well_known_backoff_mins: next_backoff_mins,
+                }
+            },
+        )
+    } else {
+        match get_ip_with_port(&destination_str) {
             Some(host_port) => {
                 debug!("1: IP literal with provided or default port");
-            host_port
+                (host_port, DestType::IsIpOrHasPort)
             }
             None => {
                 if let Some(pos) = destination_str.find(':') {
                     debug!("2: Hostname with included port");
                     let (host, port) = destination_str.split_at(pos);
-                FedDest::Named(host.to_owned(), port.to_owned())
+                    (
+                        FedDest::Named(host.to_owned(), port.to_owned()),
+                        DestType::IsIpOrHasPort,
+                    )
                 } else {
-                debug!("Requesting well known for {destination}");
-                match request_well_known(destination.as_str()).await {
-                    Some(delegated_hostname) => {
+                    debug!("Requesting well known for {destination_str}");
+                    match request_well_known(destination_str.as_str()).await {
+                        Some((delegated_hostname, timestamp)) => {
                             debug!("3: A .well-known file is available");
-                        hostname = add_port_to_hostname(&delegated_hostname).into_uri_string();
                             match get_ip_with_port(&delegated_hostname) {
-                            Some(host_and_port) => host_and_port, // 3.1: IP literal in .well-known file
+                                // 3.1: IP literal in .well-known file
+                                Some(host_and_port) => {
+                                    (host_and_port, DestType::WellKnown { expires: timestamp })
+                                }
                                 None => {
                                     if let Some(pos) = delegated_hostname.find(':') {
                                         debug!("3.2: Hostname with port in .well-known file");
                                         let (host, port) = delegated_hostname.split_at(pos);
-                                    FedDest::Named(host.to_owned(), port.to_owned())
+                                        (
+                                            FedDest::Named(host.to_owned(), port.to_owned()),
+                                            DestType::WellKnown { expires: timestamp },
+                                        )
                                     } else {
                                         debug!("Delegated hostname has no port in this branch");
-                                    if let Some(hostname_override) =
-                                        query_srv_record(&delegated_hostname).await
-                                    {
-                                        debug!("3.3: SRV lookup successful");
+                                        let (dest, srv_expires) =
+                                            get_srv_destination(delegated_hostname.clone()).await;
+                                        (
+                                            dest,
+                                            if let Some(srv_expires) = srv_expires {
+                                                DestType::WellKnownSrv {
+                                                    srv_expires,
+                                                    well_known_expires: timestamp,
+                                                    well_known_host: delegated_hostname,
+                                                }
+                                            } else {
+                                                DestType::WellKnown { expires: timestamp }
+                                            },
+                                        )
+                                    }
+                                }
+                            }
+                        }
+                        None => {
+                            debug!("4: No .well-known or an error occured");
+                            let (dest, expires) = get_srv_destination(destination_str).await;
+                            let well_known_retry = Instant::now()
+                                + Duration::from_secs((60 * next_backoff_mins).into());
+                            (
+                                dest,
+                                if let Some(expires) = expires {
+                                    DestType::Srv {
+                                        srv_expires: expires,
+                                        well_known_retry,
+                                        well_known_backoff_mins: next_backoff_mins,
+                                    }
+                                } else {
+                                    DestType::LookupFailed {
+                                        well_known_retry,
+                                        well_known_backoff_mins: next_backoff_mins,
+                                    }
+                                },
+                            )
+                        }
+                    }
+                }
+            }
+        }
+    };
+
+    debug!("Actual destination: {actual_destination:?}");
+
+    let response = DestinationResponse {
+        actual_destination,
+        dest_type,
+    };
+
+    services()
+        .globals
+        .actual_destination_cache
+        .write()
+        .await
+        .insert(destination.to_owned(), response.clone());
+
+    response.actual_destination
+}
+
+/// Looks up the SRV records for federation usage
+///
+/// If no timestamp is returned, that means no SRV record was found
+async fn get_srv_destination(delegated_hostname: String) -> (FedDest, Option<Instant>) {
+    if let Some((hostname_override, timestamp)) = query_srv_record(&delegated_hostname).await {
+        debug!("SRV lookup successful");
         let force_port = hostname_override.port();
 
         if let Ok(override_ip) = services()
@@ -395,91 +567,41 @@ async fn find_actual_destination(destination: &'_ ServerName) -> (FedDest, FedDe
                 .unwrap()
                 .insert(
                     delegated_hostname.clone(),
-                                                    (
-                                                        override_ip.iter().collect(),
-                                                        force_port.unwrap_or(8448),
-                                                    ),
+                    (override_ip.iter().collect(), force_port.unwrap_or(8448)),
                 );
         } else {
-                                            warn!("Using SRV record, but could not resolve to IP");
-                                        }
-
-                                        if let Some(port) = force_port {
-                                            FedDest::Named(delegated_hostname, format!(":{port}"))
-                                        } else {
-                                            add_port_to_hostname(&delegated_hostname)
-                                        }
-                                    } else {
-                                        debug!("3.4: No SRV records, just use the hostname from .well-known");
-                                        add_port_to_hostname(&delegated_hostname)
-                                    }
-                                }
-                            }
-                        }
-                    }
-                    None => {
-                        debug!("4: No .well-known or an error occured");
-                        match query_srv_record(&destination_str).await {
-                            Some(hostname_override) => {
-                                debug!("4: SRV record found");
-                                let force_port = hostname_override.port();
-
-                                if let Ok(override_ip) = services()
-                                    .globals
-                                    .dns_resolver()
-                                    .lookup_ip(hostname_override.hostname())
-                                    .await
-                                {
+            // Removing in case there was previously a SRV record
             services()
                 .globals
                 .tls_name_override
                 .write()
                 .unwrap()
-                                        .insert(
-                                            hostname.clone(),
-                                            (
-                                                override_ip.iter().collect(),
-                                                force_port.unwrap_or(8448),
-                                            ),
-                                        );
-                                } else {
+                .remove(&delegated_hostname);
             warn!("Using SRV record, but could not resolve to IP");
         }
 
         if let Some(port) = force_port {
-                                    FedDest::Named(hostname.clone(), format!(":{port}"))
+            (
+                FedDest::Named(delegated_hostname, format!(":{port}")),
+                Some(timestamp),
+            )
         } else {
-                                    add_port_to_hostname(&hostname)
+            (add_port_to_hostname(&delegated_hostname), Some(timestamp))
         }
-                            }
-                            None => {
-                                debug!("5: No SRV record found");
-                                add_port_to_hostname(&destination_str)
-                            }
-                        }
-                    }
-                }
-            }
-        }
-    };
-    debug!("Actual destination: {actual_destination:?}");
-
-    // Can't use get_ip_with_port here because we don't want to add a port
-    // to an IP address if it wasn't specified
-    let hostname = if let Ok(addr) = hostname.parse::<SocketAddr>() {
-        FedDest::Literal(addr)
-    } else if let Ok(addr) = hostname.parse::<IpAddr>() {
-        FedDest::Named(addr.to_string(), ":8448".to_owned())
-    } else if let Some(pos) = hostname.find(':') {
-        let (host, port) = hostname.split_at(pos);
-        FedDest::Named(host.to_owned(), port.to_owned())
     } else {
-        FedDest::Named(hostname, ":8448".to_owned())
-    };
-    (actual_destination, hostname)
+        // Removing in case there was previously a SRV record
+        services()
+            .globals
+            .tls_name_override
+            .write()
+            .unwrap()
+            .remove(&delegated_hostname);
+        debug!("No SRV records found");
+        (add_port_to_hostname(&delegated_hostname), None)
+    }
 }
 
-async fn query_given_srv_record(record: &str) -> Option<FedDest> {
+async fn query_given_srv_record(record: &str) -> Option<(FedDest, Instant)> {
     services()
         .globals
         .dns_resolver()
@@ -487,16 +609,19 @@ async fn query_given_srv_record(record: &str) -> Option<FedDest> {
         .await
         .map(|srv| {
             srv.iter().next().map(|result| {
+                (
                     FedDest::Named(
                         result.target().to_string().trim_end_matches('.').to_owned(),
                         format!(":{}", result.port()),
+                    ),
+                    srv.as_lookup().valid_until(),
                 )
             })
         })
         .unwrap_or(None)
 }
 
-async fn query_srv_record(hostname: &'_ str) -> Option<FedDest> {
+async fn query_srv_record(hostname: &'_ str) -> Option<(FedDest, Instant)> {
     let hostname = hostname.trim_end_matches('.');
 
     if let Some(host_port) = query_given_srv_record(&format!("_matrix-fed._tcp.{hostname}.")).await
@@ -507,7 +632,7 @@ async fn query_srv_record(hostname: &'_ str) -> Option<FedDest> {
     }
 }
 
-async fn request_well_known(destination: &str) -> Option<String> {
+async fn request_well_known(destination: &str) -> Option<(String, Instant)> {
     let response = services()
         .globals
         .default_client()
@@ -515,14 +640,40 @@ async fn request_well_known(destination: &str) -> Option<String> {
         .send()
         .await;
     debug!("Got well known response");
-    if let Err(e) = &response {
+    let response = match response {
+        Err(e) => {
             debug!("Well known error: {e:?}");
             return None;
         }
-    let text = response.ok()?.text().await;
+        Ok(r) => r,
+    };
+
+    let mut headers = response.headers().values();
+
+    let cache_for = CacheControl::decode(&mut headers)
+        .ok()
+        .and_then(|cc| {
+            // Servers should respect the cache control headers present on the response, or use a sensible default when headers are not present.
+            if cc.no_store() || cc.no_cache() {
+                Some(Duration::ZERO)
+            } else {
+                cc.max_age()
+                    // Servers should additionally impose a maximum cache time for responses: 48 hours is recommended.
+                    .map(|age| age.min(Duration::from_secs(60 * 60 * 48)))
+            }
+        })
+        // The recommended sensible default is 24 hours.
+        .unwrap_or_else(|| Duration::from_secs(60 * 60 * 24));
+
+    let text = response.text().await;
     debug!("Got well known response text");
+
+    let host = || {
         let body: serde_json::Value = serde_json::from_str(&text.ok()?).ok()?;
-    Some(body.get("m.server")?.as_str()?.to_owned())
+        body.get("m.server")?.as_str().map(ToOwned::to_owned)
+    };
+
+    host().map(|host| (host, Instant::now() + cache_for))
 }
 
 /// # `GET /_matrix/federation/v1/version`
@@ -662,17 +813,78 @@ pub fn parse_incoming_pdu(
 
     let (event_id, value) = match gen_event_id_canonical_json(pdu, &room_version_id) {
         Ok(t) => t,
-        Err(_) => {
+        Err(e) => {
             // Event could not be converted to canonical json
-            return Err(Error::BadRequest(
-                ErrorKind::InvalidParam,
-                "Could not convert event to canonical json.",
-            ));
+            return Err(e);
         }
     };
     Ok((event_id, value, room_id))
 }
 
+/// Attempts to parse and append PDU to timeline.
+/// If no event ID is returned, then the PDU was failed to be parsed.
+/// If the Ok(()) is returned, then the PDU was successfully appended to the timeline.
+async fn handle_pdu_in_transaction(
+    origin: &ServerName,
+    pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
+    pdu: &RawJsonValue,
+) -> (Option<OwnedEventId>, Result<()>) {
+    let (event_id, value, room_id) = match parse_incoming_pdu(pdu) {
+        Ok(t) => t,
+        Err(e) => {
+            warn!("Could not parse PDU: {e}");
+            warn!("Full PDU: {:?}", &pdu);
+            return (None, Err(Error::BadServerResponse("Could not parse PDU")));
+        }
+    };
+
+    // Makes use of the m.room.create event. If we cannot fetch this event,
+    // we must have never been in that room.
+    if services().rooms.state.get_room_version(&room_id).is_err() {
+        debug!("Room {room_id} is not known to this server");
+        return (
+            Some(event_id),
+            Err(Error::BadServerResponse("Room is not known to this server")),
+        );
+    }
+
+    // We do not add the event_id field to the pdu here because of signature and hashes checks
+
+    let mutex = Arc::clone(
+        services()
+            .globals
+            .roomid_mutex_federation
+            .write()
+            .await
+            .entry(room_id.to_owned())
+            .or_default(),
+    );
+    let mutex_lock = mutex.lock().await;
+    let start_time = Instant::now();
+
+    if let Err(e) = services()
+        .rooms
+        .event_handler
+        .handle_incoming_pdu(origin, &event_id, &room_id, value, true, pub_key_map)
+        .await
+    {
+        warn!("Error appending PDU to timeline: {}: {:?}", e, pdu);
+        return (Some(event_id), Err(e));
+    }
+
+    drop(mutex_lock);
+
+    let elapsed = start_time.elapsed();
+    debug!(
+        "Handling transaction of event {} took {}m{}s",
+        event_id,
+        elapsed.as_secs() / 60,
+        elapsed.as_secs() % 60
+    );
+
+    (Some(event_id), Ok(()))
+}
+
 /// # `PUT /_matrix/federation/v1/send/{txnId}`
 ///
 /// Push EDUs and PDUs to this server.
@@ -697,77 +909,11 @@ pub async fn send_transaction_message_route(
     // let mut auth_cache = EventMap::new();
 
     for pdu in &body.pdus {
-        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: OwnedRoomId = value
-            .get("room_id")
-            .and_then(|id| RoomId::parse(id.as_str()?).ok())
-            .ok_or(Error::BadRequest(
-                ErrorKind::InvalidParam,
-                "Invalid room id in pdu",
-            ))?;
+        let (event_id, result) =
+            handle_pdu_in_transaction(sender_servername, &pub_key_map, pdu).await;
 
-        if services().rooms.state.get_room_version(&room_id).is_err() {
-            debug!("Server is not in room {room_id}");
-            continue;
-        }
-
-        let r = parse_incoming_pdu(pdu);
-        let (event_id, value, room_id) = match r {
-            Ok(t) => t,
-            Err(e) => {
-                warn!("Could not parse PDU: {e}");
-                warn!("Full PDU: {:?}", &pdu);
-                continue;
-            }
-        };
-        // We do not add the event_id field to the pdu here because of signature and hashes checks
-
-        let mutex = Arc::clone(
-            services()
-                .globals
-                .roomid_mutex_federation
-                .write()
-                .await
-                .entry(room_id.to_owned())
-                .or_default(),
-        );
-        let mutex_lock = mutex.lock().await;
-        let start_time = Instant::now();
-        resolved_map.insert(
-            event_id.clone(),
-            services()
-                .rooms
-                .event_handler
-                .handle_incoming_pdu(
-                    sender_servername,
-                    &event_id,
-                    &room_id,
-                    value,
-                    true,
-                    &pub_key_map,
-                )
-                .await
-                .map(|_| ()),
-        );
-        drop(mutex_lock);
-
-        let elapsed = start_time.elapsed();
-        debug!(
-            "Handling transaction of event {} took {}m{}s",
-            event_id,
-            elapsed.as_secs() / 60,
-            elapsed.as_secs() % 60
-        );
-    }
-
-    for pdu in &resolved_map {
-        if let Err(e) = pdu.1 {
-            if matches!(e, Error::BadRequest(ErrorKind::NotFound, _)) {
-                warn!("Incoming PDU failed {:?}", pdu);
-            }
+        if let Some(event_id) = event_id {
+            resolved_map.insert(event_id.clone(), result.map_err(|e| e.sanitized_error()));
         }
     }
 
@@ -781,6 +927,7 @@ pub async fn send_transaction_message_route(
             Edu::Receipt(receipt) => {
                 for (room_id, room_updates) in receipt.receipts {
                     for (user_id, user_updates) in room_updates.read {
+                        if user_id.server_name() == sender_servername {
                             if let Some((event_id, _)) = user_updates
                                 .event_ids
                                 .iter()
@@ -820,8 +967,10 @@ pub async fn send_transaction_message_route(
                         }
                     }
                 }
+            }
             Edu::Typing(typing) => {
-                if services()
+                if typing.user_id.server_name() == sender_servername
+                    && services()
                         .rooms
                         .state_cache
                         .is_joined(&typing.user_id, &typing.room_id)?
@@ -848,23 +997,23 @@ pub async fn send_transaction_message_route(
                 }
             }
             Edu::DeviceListUpdate(DeviceListUpdateContent { user_id, .. }) => {
+                if user_id.server_name() == sender_servername {
                     services().users.mark_device_key_update(&user_id)?;
                 }
+            }
             Edu::DirectToDevice(DirectDeviceContent {
                 sender,
                 ev_type,
                 message_id,
                 messages,
             }) => {
+                if sender.server_name() == sender_servername
                     // Check if this is a new transaction id
-                if services()
+                    && services()
                         .transaction_ids
                         .existing_txnid(&sender, None, &message_id)?
-                    .is_some()
+                        .is_none()
                 {
-                    continue;
-                }
-
                     for (target_user_id, map) in &messages {
                         for (target_device_id_maybe, event) in map {
                             match target_device_id_maybe {
@@ -911,14 +1060,13 @@ pub async fn send_transaction_message_route(
                         .transaction_ids
                         .add_txnid(&sender, None, &message_id, &[])?;
                 }
+            }
             Edu::SigningKeyUpdate(SigningKeyUpdateContent {
                 user_id,
                 master_key,
                 self_signing_key,
             }) => {
-                if user_id.server_name() != sender_servername {
-                    continue;
-                }
+                if user_id.server_name() == sender_servername {
                     if let Some(master_key) = master_key {
                         services().users.add_cross_signing_keys(
                             &user_id,
@@ -929,16 +1077,12 @@ pub async fn send_transaction_message_route(
                         )?;
                     }
                 }
+            }
             Edu::_Custom(_) => {}
         }
     }
 
-    Ok(send_transaction_message::v1::Response {
-        pdus: resolved_map
-            .into_iter()
-            .map(|(e, r)| (e, r.map_err(|e| e.sanitized_error())))
-            .collect(),
-    })
+    Ok(send_transaction_message::v1::Response { pdus: resolved_map })
 }
 
 /// # `GET /_matrix/federation/v1/event/{eventId}`
@@ -977,7 +1121,7 @@ pub async fn get_event_route(
         .server_in_room(sender_servername, room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not in room",
         ));
     }
@@ -988,7 +1132,7 @@ pub async fn get_event_route(
         &body.event_id,
     )? {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not allowed to see event.",
         ));
     }
@@ -1020,7 +1164,7 @@ pub async fn get_backfill_route(
         .server_in_room(sender_servername, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not in room.",
         ));
     }
@@ -1090,7 +1234,7 @@ pub async fn get_missing_events_route(
         .server_in_room(sender_servername, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not in room",
         ));
     }
@@ -1175,7 +1319,7 @@ pub async fn get_event_authorization_route(
         .server_in_room(sender_servername, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not in room.",
         ));
     }
@@ -1233,7 +1377,7 @@ pub async fn get_room_state_route(
         .server_in_room(sender_servername, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not in room.",
         ));
     }
@@ -1309,7 +1453,7 @@ pub async fn get_room_state_ids_route(
         .server_in_room(sender_servername, &body.room_id)?
     {
         return Err(Error::BadRequest(
-            ErrorKind::Forbidden,
+            ErrorKind::forbidden(),
             "Server is not in room.",
         ));
     }
@@ -1441,6 +1585,7 @@ pub async fn create_join_event_template_route(
             unsigned: None,
             state_key: Some(body.user_id.to_string()),
             redacts: None,
+            timestamp: None,
         },
         &body.user_id,
         &body.room_id,
@@ -1680,7 +1825,7 @@ pub async fn create_invite_route(
     let event_id = EventId::parse(format!(
         "${}",
         ruma::signatures::reference_hash(&signed_event, &body.room_version)
-            .expect("ruma can calculate reference hashes")
+            .expect("Event format validated when event was hashed")
     ))
     .expect("ruma's reference hashes are valid event ids");
 
@@ -1749,6 +1894,90 @@ pub async fn create_invite_route(
     })
 }
 
+/// # `GET /_matrix/federation/v1/media/download/{serverName}/{mediaId}`
+///
+/// Load media from our server.
+pub async fn get_content_route(
+    body: Ruma<get_content::v1::Request>,
+) -> Result<get_content::v1::Response> {
+    let mxc = format!(
+        "mxc://{}/{}",
+        services().globals.server_name(),
+        body.media_id
+    );
+
+    if let Some(FileMeta {
+        content_disposition,
+        content_type,
+        file,
+    }) = services().media.get(mxc.clone()).await?
+    {
+        Ok(get_content::v1::Response::new(
+            ContentMetadata::new(),
+            FileOrLocation::File(Content {
+                file,
+                content_type,
+                content_disposition: Some(content_disposition),
+            }),
+        ))
+    } else {
+        Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
+    }
+}
+
+/// # `GET /_matrix/federation/v1/media/thumbnail/{serverName}/{mediaId}`
+///
+/// Load media thumbnail from our server or over federation.
+pub async fn get_content_thumbnail_route(
+    body: Ruma<get_content_thumbnail::v1::Request>,
+) -> Result<get_content_thumbnail::v1::Response> {
+    let mxc = format!(
+        "mxc://{}/{}",
+        services().globals.server_name(),
+        body.media_id
+    );
+
+    let Some(FileMeta {
+        file,
+        content_type,
+        content_disposition,
+    }) = services()
+        .media
+        .get_thumbnail(
+            mxc.clone(),
+            body.width
+                .try_into()
+                .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Width is invalid."))?,
+            body.height
+                .try_into()
+                .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Width is invalid."))?,
+        )
+        .await?
+    else {
+        return Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."));
+    };
+
+    services()
+        .media
+        .upload_thumbnail(
+            mxc,
+            content_type.as_deref(),
+            body.width.try_into().expect("all UInts are valid u32s"),
+            body.height.try_into().expect("all UInts are valid u32s"),
+            &file,
+        )
+        .await?;
+
+    Ok(get_content_thumbnail::v1::Response::new(
+        ContentMetadata::new(),
+        FileOrLocation::File(Content {
+            file,
+            content_type,
+            content_disposition: Some(content_disposition),
+        }),
+    ))
+}
+
 /// # `GET /_matrix/federation/v1/user/devices/{userId}`
 ///
 /// Gets information on all devices of the user.
@@ -1914,6 +2143,25 @@ pub async fn claim_keys_route(
     })
 }
 
+/// # `GET /_matrix/federation/v1/openid/userinfo`
+///
+/// Get information about the user that generated the OpenID token.
+pub async fn get_openid_userinfo_route(
+    body: Ruma<get_openid_userinfo::v1::Request>,
+) -> Result<get_openid_userinfo::v1::Response> {
+    Ok(get_openid_userinfo::v1::Response::new(
+        services()
+            .users
+            .find_from_openid_token(&body.access_token)?
+            .ok_or_else(|| {
+                Error::BadRequest(
+                    ErrorKind::Unauthorized,
+                    "OpenID token has expired or does not exist.",
+                )
+            })?,
+    ))
+}
+
 /// # `GET /.well-known/matrix/server`
 ///
 /// Returns the federation server discovery information.

src/config/mod.rs

@@ -47,6 +47,8 @@ pub struct Config {
     #[serde(default = "false_fn")]
     pub allow_registration: bool,
     pub registration_token: Option<String>,
+    #[serde(default = "default_openid_token_ttl")]
+    pub openid_token_ttl: u64,
     #[serde(default = "true_fn")]
     pub allow_encryption: bool,
     #[serde(default = "false_fn")]
@@ -57,7 +59,7 @@ pub struct Config {
     pub allow_unstable_room_versions: bool,
     #[serde(default = "default_default_room_version")]
     pub default_room_version: RoomVersionId,
-    #[serde(default)]
+    #[serde(default, flatten)]
     pub well_known: WellKnownConfig,
     #[serde(default = "false_fn")]
     pub allow_jaeger: bool,
@@ -95,7 +97,9 @@ pub struct TlsConfig {
 
 #[derive(Clone, Debug, Deserialize, Default)]
 pub struct WellKnownConfig {
+    #[serde(rename = "well_known_client")]
     pub client: Option<Url>,
+    #[serde(rename = "well_known_server")]
     pub server: Option<OwnedServerName>,
 }
 
@@ -302,6 +306,10 @@ fn default_turn_ttl() -> u64 {
     60 * 60 * 24
 }
 
+fn default_openid_token_ttl() -> u64 {
+    60 * 60
+}
+
 // I know, it's a great name
 pub fn default_default_room_version() -> RoomVersionId {
     RoomVersionId::V10

src/database/key_value/globals.rs

@@ -1,15 +1,19 @@
-use std::collections::{BTreeMap, HashMap};
+use std::collections::HashMap;
 
 use async_trait::async_trait;
 use futures_util::{stream::FuturesUnordered, StreamExt};
 use lru_cache::LruCache;
 use ruma::{
-    api::federation::discovery::{ServerSigningKeys, VerifyKey},
+    api::federation::discovery::{OldVerifyKey, ServerSigningKeys},
     signatures::Ed25519KeyPair,
-    DeviceId, MilliSecondsSinceUnixEpoch, OwnedServerSigningKeyId, ServerName, UserId,
+    DeviceId, ServerName, UserId,
 };
 
-use crate::{database::KeyValueDatabase, service, services, utils, Error, Result};
+use crate::{
+    database::KeyValueDatabase,
+    service::{self, globals::SigningKeys},
+    services, utils, Error, Result,
+};
 
 pub const COUNTER: &[u8] = b"c";
 pub const LAST_CHECK_FOR_UPDATES_COUNT: &[u8] = b"u";
@@ -237,64 +241,97 @@ lasttimelinecount_cache: {lasttimelinecount_cache}\n"
         self.global.remove(b"keypair")
     }
 
-    fn add_signing_key(
+    fn add_signing_key_from_trusted_server(
         &self,
         origin: &ServerName,
         new_keys: ServerSigningKeys,
-    ) -> Result<BTreeMap<OwnedServerSigningKeyId, VerifyKey>> {
-        // Not atomic, but this is not critical
-        let signingkeys = self.server_signingkeys.get(origin.as_bytes())?;
-
-        let mut keys = signingkeys
-            .and_then(|keys| serde_json::from_slice(&keys).ok())
-            .unwrap_or_else(|| {
-                // Just insert "now", it doesn't matter
-                ServerSigningKeys::new(origin.to_owned(), MilliSecondsSinceUnixEpoch::now())
-            });
+    ) -> Result<SigningKeys> {
+        let prev_keys = self.server_signingkeys.get(origin.as_bytes())?;
 
+        Ok(
+            if let Some(mut prev_keys) =
+                prev_keys.and_then(|keys| serde_json::from_slice::<ServerSigningKeys>(&keys).ok())
+            {
                 let ServerSigningKeys {
                     verify_keys,
                     old_verify_keys,
                     ..
                 } = new_keys;
 
-        keys.verify_keys.extend(verify_keys);
-        keys.old_verify_keys.extend(old_verify_keys);
+                prev_keys.verify_keys.extend(verify_keys);
+                prev_keys.old_verify_keys.extend(old_verify_keys);
+                prev_keys.valid_until_ts = new_keys.valid_until_ts;
 
                 self.server_signingkeys.insert(
                     origin.as_bytes(),
-            &serde_json::to_vec(&keys).expect("serversigningkeys can be serialized"),
+                    &serde_json::to_vec(&prev_keys).expect("serversigningkeys can be serialized"),
                 )?;
 
-        let mut tree = keys.verify_keys;
-        tree.extend(
-            keys.old_verify_keys
-                .into_iter()
-                .map(|old| (old.0, VerifyKey::new(old.1.key))),
-        );
+                prev_keys.into()
+            } else {
+                self.server_signingkeys.insert(
+                    origin.as_bytes(),
+                    &serde_json::to_vec(&new_keys).expect("serversigningkeys can be serialized"),
+                )?;
 
-        Ok(tree)
+                new_keys.into()
+            },
+        )
+    }
+
+    fn add_signing_key_from_origin(
+        &self,
+        origin: &ServerName,
+        new_keys: ServerSigningKeys,
+    ) -> Result<SigningKeys> {
+        let prev_keys = self.server_signingkeys.get(origin.as_bytes())?;
+
+        Ok(
+            if let Some(mut prev_keys) =
+                prev_keys.and_then(|keys| serde_json::from_slice::<ServerSigningKeys>(&keys).ok())
+            {
+                let ServerSigningKeys {
+                    verify_keys,
+                    old_verify_keys,
+                    ..
+                } = new_keys;
+
+                // Moving `verify_keys` no longer present to `old_verify_keys`
+                for (key_id, key) in prev_keys.verify_keys {
+                    if !verify_keys.contains_key(&key_id) {
+                        prev_keys
+                            .old_verify_keys
+                            .insert(key_id, OldVerifyKey::new(prev_keys.valid_until_ts, key.key));
+                    }
+                }
+
+                prev_keys.verify_keys = verify_keys;
+                prev_keys.old_verify_keys.extend(old_verify_keys);
+                prev_keys.valid_until_ts = new_keys.valid_until_ts;
+
+                self.server_signingkeys.insert(
+                    origin.as_bytes(),
+                    &serde_json::to_vec(&prev_keys).expect("serversigningkeys can be serialized"),
+                )?;
+
+                prev_keys.into()
+            } else {
+                self.server_signingkeys.insert(
+                    origin.as_bytes(),
+                    &serde_json::to_vec(&new_keys).expect("serversigningkeys can be serialized"),
+                )?;
+
+                new_keys.into()
+            },
+        )
     }
 
     /// This returns an empty `Ok(BTreeMap<..>)` when there are no keys found for the server.
-    fn signing_keys_for(
-        &self,
-        origin: &ServerName,
-    ) -> Result<BTreeMap<OwnedServerSigningKeyId, VerifyKey>> {
+    fn signing_keys_for(&self, origin: &ServerName) -> Result<Option<SigningKeys>> {
         let signingkeys = self
             .server_signingkeys
             .get(origin.as_bytes())?
-            .and_then(|bytes| serde_json::from_slice(&bytes).ok())
-            .map(|keys: ServerSigningKeys| {
-                let mut tree = keys.verify_keys;
-                tree.extend(
-                    keys.old_verify_keys
-                        .into_iter()
-                        .map(|old| (old.0, VerifyKey::new(old.1.key))),
-                );
-                tree
-            })
-            .unwrap_or_else(BTreeMap::new);
+            .and_then(|bytes| serde_json::from_slice::<SigningKeys>(&bytes).ok());
 
         Ok(signingkeys)
     }

src/database/key_value/media.rs

@@ -1,4 +1,4 @@
-use ruma::api::client::error::ErrorKind;
+use ruma::{api::client::error::ErrorKind, http_headers::ContentDisposition};
 
 use crate::{database::KeyValueDatabase, service, utils, Error, Result};
 
@@ -8,7 +8,7 @@ impl service::media::Data for KeyValueDatabase {
         mxc: String,
         width: u32,
         height: u32,
-        content_disposition: Option<&str>,
+        content_disposition: &ContentDisposition,
         content_type: Option<&str>,
     ) -> Result<Vec<u8>> {
         let mut key = mxc.as_bytes().to_vec();
@@ -16,12 +16,7 @@ impl service::media::Data for KeyValueDatabase {
         key.extend_from_slice(&width.to_be_bytes());
         key.extend_from_slice(&height.to_be_bytes());
         key.push(0xff);
-        key.extend_from_slice(
-            content_disposition
-                .as_ref()
-                .map(|f| f.as_bytes())
-                .unwrap_or_default(),
-        );
+        key.extend_from_slice(content_disposition.to_string().as_bytes());
         key.push(0xff);
         key.extend_from_slice(
             content_type
@@ -40,7 +35,7 @@ impl service::media::Data for KeyValueDatabase {
         mxc: String,
         width: u32,
         height: u32,
-    ) -> Result<(Option<String>, Option<String>, Vec<u8>)> {
+    ) -> Result<(ContentDisposition, Option<String>, Vec<u8>)> {
         let mut prefix = mxc.as_bytes().to_vec();
         prefix.push(0xff);
         prefix.extend_from_slice(&width.to_be_bytes());
@@ -68,15 +63,9 @@ impl service::media::Data for KeyValueDatabase {
             .next()
             .ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
 
-        let content_disposition = if content_disposition_bytes.is_empty() {
-            None
-        } else {
-            Some(
-                utils::string_from_bytes(content_disposition_bytes).map_err(|_| {
-                    Error::bad_database("Content Disposition in mediaid_file is invalid unicode.")
-                })?,
-            )
-        };
+        let content_disposition = content_disposition_bytes.try_into().unwrap_or_else(|_| {
+            ContentDisposition::new(ruma::http_headers::ContentDispositionType::Inline)
+        });
         Ok((content_disposition, content_type, key))
     }
 }

src/database/key_value/rooms/alias.rs

@@ -1,9 +1,15 @@
-use ruma::{api::client::error::ErrorKind, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId};
+use ruma::{
+    api::client::error::ErrorKind, OwnedRoomAliasId, OwnedRoomId, OwnedUserId, RoomAliasId, RoomId,
+    UserId,
+};
 
 use crate::{database::KeyValueDatabase, service, services, utils, Error, Result};
 
 impl service::rooms::alias::Data for KeyValueDatabase {
-    fn set_alias(&self, alias: &RoomAliasId, room_id: &RoomId) -> Result<()> {
+    fn set_alias(&self, alias: &RoomAliasId, room_id: &RoomId, user_id: &UserId) -> Result<()> {
+        // Comes first as we don't want a stuck alias
+        self.alias_userid
+            .insert(alias.alias().as_bytes(), user_id.as_bytes())?;
         self.alias_roomid
             .insert(alias.alias().as_bytes(), room_id.as_bytes())?;
         let mut aliasid = room_id.as_bytes().to_vec();
@@ -22,13 +28,13 @@ impl service::rooms::alias::Data for KeyValueDatabase {
                 self.aliasid_alias.remove(&key)?;
             }
             self.alias_roomid.remove(alias.alias().as_bytes())?;
+            self.alias_userid.remove(alias.alias().as_bytes())
         } else {
-            return Err(Error::BadRequest(
+            Err(Error::BadRequest(
                 ErrorKind::NotFound,
                 "Alias does not exist.",
-            ));
+            ))
         }
-        Ok(())
     }
 
     fn resolve_local_alias(&self, alias: &RoomAliasId) -> Result<Option<OwnedRoomId>> {
@@ -57,4 +63,16 @@ impl service::rooms::alias::Data for KeyValueDatabase {
                 .map_err(|_| Error::bad_database("Invalid alias in aliasid_alias."))
         }))
     }
+
+    fn who_created_alias(&self, alias: &RoomAliasId) -> Result<Option<OwnedUserId>> {
+        self.alias_userid
+            .get(alias.alias().as_bytes())?
+            .map(|bytes| {
+                UserId::parse(utils::string_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("User ID in alias_userid is invalid unicode.")
+                })?)
+                .map_err(|_| Error::bad_database("User ID in alias_roomid is invalid."))
+            })
+            .transpose()
+    }
 }

src/database/key_value/rooms/search.rs

@@ -2,14 +2,20 @@ use ruma::RoomId;
 
 use crate::{database::KeyValueDatabase, service, services, utils, Result};
 
-impl service::rooms::search::Data for KeyValueDatabase {
-    fn index_pdu<'a>(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()> {
-        let mut batch = message_body
-            .split_terminator(|c: char| !c.is_alphanumeric())
+/// Splits a string into tokens used as keys in the search inverted index
+///
+/// This may be used to tokenize both message bodies (for indexing) or search
+/// queries (for querying).
+fn tokenize(body: &str) -> impl Iterator<Item = String> + '_ {
+    body.split_terminator(|c: char| !c.is_alphanumeric())
         .filter(|s| !s.is_empty())
         .filter(|word| word.len() <= 50)
         .map(str::to_lowercase)
-            .map(|word| {
+}
+
+impl service::rooms::search::Data for KeyValueDatabase {
+    fn index_pdu<'a>(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()> {
+        let mut batch = tokenize(message_body).map(|word| {
             let mut key = shortroomid.to_be_bytes().to_vec();
             key.extend_from_slice(word.as_bytes());
             key.push(0xff);
@@ -20,6 +26,22 @@ impl service::rooms::search::Data for KeyValueDatabase {
         self.tokenids.insert_batch(&mut batch)
     }
 
+    fn deindex_pdu(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()> {
+        let batch = tokenize(message_body).map(|word| {
+            let mut key = shortroomid.to_be_bytes().to_vec();
+            key.extend_from_slice(word.as_bytes());
+            key.push(0xFF);
+            key.extend_from_slice(pdu_id); // TODO: currently we save the room id a second time here
+            key
+        });
+
+        for token in batch {
+            self.tokenids.remove(&token)?;
+        }
+
+        Ok(())
+    }
+
     fn search_pdus<'a>(
         &'a self,
         room_id: &RoomId,
@@ -33,11 +55,7 @@ impl service::rooms::search::Data for KeyValueDatabase {
             .to_be_bytes()
             .to_vec();
 
-        let words: Vec<_> = search_string
-            .split_terminator(|c: char| !c.is_alphanumeric())
-            .filter(|s| !s.is_empty())
-            .map(str::to_lowercase)
-            .collect();
+        let words: Vec<_> = tokenize(search_string).collect();
 
         let iterators = words.clone().into_iter().map(move |word| {
             let mut prefix2 = prefix.clone();

src/database/key_value/uiaa.rs

@@ -80,7 +80,7 @@ impl service::uiaa::Data for KeyValueDatabase {
                 .userdevicesessionid_uiaainfo
                 .get(&userdevicesessionid)?
                 .ok_or(Error::BadRequest(
-                    ErrorKind::Forbidden,
+                    ErrorKind::forbidden(),
                     "UIAA session does not exist.",
                 ))?,
         )

src/database/key_value/users.rs

@@ -11,6 +11,7 @@ use ruma::{
 use tracing::warn;
 
 use crate::{
+    api::client_server::TOKEN_LENGTH,
     database::KeyValueDatabase,
     service::{self, users::clean_signatures},
     services, utils, Error, Result,
@@ -943,6 +944,52 @@ impl service::users::Data for KeyValueDatabase {
             Ok(None)
         }
     }
+
+    // Creates an OpenID token, which can be used to prove that a user has access to an account (primarily for integrations)
+    fn create_openid_token(&self, user_id: &UserId) -> Result<(String, u64)> {
+        let token = utils::random_string(TOKEN_LENGTH);
+
+        let expires_in = services().globals.config.openid_token_ttl;
+        let expires_at = utils::millis_since_unix_epoch()
+            .checked_add(expires_in * 1000)
+            .expect("time is valid");
+
+        let mut value = expires_at.to_be_bytes().to_vec();
+        value.extend_from_slice(user_id.as_bytes());
+
+        self.openidtoken_expiresatuserid
+            .insert(token.as_bytes(), value.as_slice())?;
+
+        Ok((token, expires_in))
+    }
+
+    /// Find out which user an OpenID access token belongs to.
+    fn find_from_openid_token(&self, token: &str) -> Result<Option<OwnedUserId>> {
+        let Some(value) = self.openidtoken_expiresatuserid.get(token.as_bytes())? else {
+            return Ok(None);
+        };
+        let (expires_at_bytes, user_bytes) = value.split_at(0u64.to_be_bytes().len());
+
+        let expires_at = u64::from_be_bytes(
+            expires_at_bytes
+                .try_into()
+                .map_err(|_| Error::bad_database("expires_at in openid_userid is invalid u64."))?,
+        );
+
+        if expires_at < utils::millis_since_unix_epoch() {
+            self.openidtoken_expiresatuserid.remove(token.as_bytes())?;
+
+            return Ok(None);
+        }
+
+        Some(
+            UserId::parse(utils::string_from_bytes(user_bytes).map_err(|_| {
+                Error::bad_database("User ID in openid_userid is invalid unicode.")
+            })?)
+            .map_err(|_| Error::bad_database("User ID in openid_userid is invalid.")),
+        )
+        .transpose()
+    }
 }
 
 impl KeyValueDatabase {}

src/database/mod.rs

@@ -6,6 +6,7 @@ use crate::{
     SERVICES,
 };
 use abstraction::{KeyValueDatabaseEngine, KvTree};
+use base64::{engine::general_purpose, Engine};
 use directories::ProjectDirs;
 use lru_cache::LruCache;
 
@@ -57,6 +58,7 @@ pub struct KeyValueDatabase {
     pub(super) userid_masterkeyid: Arc<dyn KvTree>,
     pub(super) userid_selfsigningkeyid: Arc<dyn KvTree>,
     pub(super) userid_usersigningkeyid: Arc<dyn KvTree>,
+    pub(super) openidtoken_expiresatuserid: Arc<dyn KvTree>, // expiresatuserid  = expiresat + userid
 
     pub(super) userfilterid_filter: Arc<dyn KvTree>, // UserFilterId = UserId + FilterId
 
@@ -100,6 +102,8 @@ pub struct KeyValueDatabase {
     pub(super) userroomid_leftstate: Arc<dyn KvTree>,
     pub(super) roomuserid_leftcount: Arc<dyn KvTree>,
 
+    pub(super) alias_userid: Arc<dyn KvTree>, // User who created the alias
+
     pub(super) disabledroomids: Arc<dyn KvTree>, // Rooms where incoming federation handling is disabled
 
     pub(super) lazyloadedids: Arc<dyn KvTree>, // LazyLoadedIds = UserId + DeviceId + RoomId + LazyLoadedUserId
@@ -290,6 +294,7 @@ impl KeyValueDatabase {
             userid_masterkeyid: builder.open_tree("userid_masterkeyid")?,
             userid_selfsigningkeyid: builder.open_tree("userid_selfsigningkeyid")?,
             userid_usersigningkeyid: builder.open_tree("userid_usersigningkeyid")?,
+            openidtoken_expiresatuserid: builder.open_tree("openidtoken_expiresatuserid")?,
             userfilterid_filter: builder.open_tree("userfilterid_filter")?,
             todeviceid_events: builder.open_tree("todeviceid_events")?,
 
@@ -325,6 +330,8 @@ impl KeyValueDatabase {
             userroomid_leftstate: builder.open_tree("userroomid_leftstate")?,
             roomuserid_leftcount: builder.open_tree("roomuserid_leftcount")?,
 
+            alias_userid: builder.open_tree("alias_userid")?,
+
             disabledroomids: builder.open_tree("disabledroomids")?,
 
             lazyloadedids: builder.open_tree("lazyloadedids")?,
@@ -404,11 +411,9 @@ impl KeyValueDatabase {
         // Matrix resource ownership is based on the server name; changing it
         // requires recreating the database from scratch.
         if services().users.count()? > 0 {
-            let conduit_user =
-                UserId::parse_with_server_name("conduit", services().globals.server_name())
-                    .expect("@conduit:server_name is valid");
+            let conduit_user = services().globals.server_user();
 
-            if !services().users.exists(&conduit_user)? {
+            if !services().users.exists(conduit_user)? {
                 error!(
                     "The {} server user does not exist, and the database is not new.",
                     conduit_user
@@ -420,7 +425,7 @@ impl KeyValueDatabase {
         }
 
         // If the database has any data, perform data migrations before starting
-        let latest_database_version = 13;
+        let latest_database_version = 16;
 
         if services().users.count()? > 0 {
             // MIGRATIONS
@@ -937,6 +942,86 @@ impl KeyValueDatabase {
                 warn!("Migration: 12 -> 13 finished");
             }
 
+            if services().globals.database_version()? < 16 {
+                // Reconstruct all media using the filesystem
+                db.mediaid_file.clear().unwrap();
+
+                for file in fs::read_dir(services().globals.get_media_folder()).unwrap() {
+                    let file = file.unwrap();
+                    let mediaid = general_purpose::URL_SAFE_NO_PAD
+                        .decode(file.file_name().into_string().unwrap())
+                        .unwrap();
+
+                    let mut parts = mediaid.rsplit(|&b| b == 0xff);
+
+                    let mut removed_bytes = 0;
+
+                    let content_type_bytes = parts.next().unwrap();
+                    removed_bytes += content_type_bytes.len() + 1;
+
+                    let content_disposition_bytes = parts
+                        .next()
+                        .ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
+                    removed_bytes += content_disposition_bytes.len();
+
+                    let mut content_disposition =
+                        utils::string_from_bytes(content_disposition_bytes).map_err(|_| {
+                            Error::bad_database("Content Disposition in mediaid_file is invalid.")
+                        })?;
+
+                    if content_disposition.contains("filename=")
+                        && !content_disposition.contains("filename=\"")
+                    {
+                        println!("{}", &content_disposition);
+                        content_disposition =
+                            content_disposition.replacen("filename=", "filename=\"", 1);
+                        content_disposition.push('"');
+                        println!("{}", &content_disposition);
+
+                        let mut new_key = mediaid[..(mediaid.len() - removed_bytes)].to_vec();
+                        assert!(*new_key.last().unwrap() == 0xff);
+
+                        let mut shorter_key = new_key.clone();
+                        shorter_key.extend(
+                            ruma::http_headers::ContentDisposition::new(
+                                ruma::http_headers::ContentDispositionType::Inline,
+                            )
+                            .to_string()
+                            .as_bytes(),
+                        );
+                        shorter_key.push(0xff);
+                        shorter_key.extend_from_slice(content_type_bytes);
+
+                        new_key.extend_from_slice(content_disposition.to_string().as_bytes());
+                        new_key.push(0xff);
+                        new_key.extend_from_slice(content_type_bytes);
+
+                        // Some file names are too long. Ignore those.
+                        match fs::rename(
+                            services().globals.get_media_file(&mediaid),
+                            services().globals.get_media_file(&new_key),
+                        ) {
+                            Ok(_) => {
+                                db.mediaid_file.insert(&new_key, &[])?;
+                            }
+                            Err(_) => {
+                                fs::rename(
+                                    services().globals.get_media_file(&mediaid),
+                                    services().globals.get_media_file(&shorter_key),
+                                )
+                                .unwrap();
+                                db.mediaid_file.insert(&shorter_key, &[])?;
+                            }
+                        }
+                    } else {
+                        db.mediaid_file.insert(&mediaid, &[])?;
+                    }
+                }
+                services().globals.bump_database_version(16)?;
+
+                warn!("Migration: 13 -> 16 finished");
+            }
+
             assert_eq!(
                 services().globals.database_version().unwrap(),
                 latest_database_version
@@ -1102,22 +1187,21 @@ impl KeyValueDatabase {
 
 /// Sets the emergency password and push rules for the @conduit account in case emergency password is set
 fn set_emergency_access() -> Result<bool> {
-    let conduit_user = UserId::parse_with_server_name("conduit", services().globals.server_name())
-        .expect("@conduit:server_name is a valid UserId");
+    let conduit_user = services().globals.server_user();
 
     services().users.set_password(
-        &conduit_user,
+        conduit_user,
         services().globals.emergency_password().as_deref(),
     )?;
 
     let (ruleset, res) = match services().globals.emergency_password() {
-        Some(_) => (Ruleset::server_default(&conduit_user), Ok(true)),
+        Some(_) => (Ruleset::server_default(conduit_user), Ok(true)),
         None => (Ruleset::new(), Ok(false)),
     };
 
     services().account_data.update(
         None,
-        &conduit_user,
+        conduit_user,
         GlobalAccountDataEventType::PushRules.to_string().into(),
         &serde_json::to_value(&GlobalAccountDataEvent {
             content: PushRulesEventContent { global: ruleset },

src/main.rs

@@ -1,8 +1,10 @@
 use std::{future::Future, io, net::SocketAddr, sync::atomic, time::Duration};
 
 use axum::{
+    body::Body,
     extract::{DefaultBodyLimit, FromRequestParts, MatchedPath},
-    response::IntoResponse,
+    middleware::map_response,
+    response::{IntoResponse, Response},
     routing::{any, get, on, MethodFilter},
     Router,
 };
@@ -13,7 +15,7 @@ use figment::{
     Figment,
 };
 use http::{
-    header::{self, HeaderName},
+    header::{self, HeaderName, CONTENT_SECURITY_POLICY},
     Method, StatusCode, Uri,
 };
 use ruma::api::{
@@ -55,7 +57,7 @@ async fn main() {
                 ))
                 .nested(),
             )
-            .merge(Env::prefixed("CONDUIT_").global());
+            .merge(Env::prefixed("CONDUIT_").global().split("__"));
 
     let config = match raw_config.extract::<Config>() {
         Ok(s) => s,
@@ -68,11 +70,13 @@ async fn main() {
     config.warn_deprecated();
 
     if config.allow_jaeger {
-        opentelemetry::global::set_text_map_propagator(opentelemetry_jaeger::Propagator::new());
-        let tracer = opentelemetry_jaeger::new_agent_pipeline()
-            .with_auto_split_batch(true)
-            .with_service_name("conduit")
-            .install_batch(opentelemetry::runtime::Tokio)
+        opentelemetry::global::set_text_map_propagator(
+            opentelemetry_jaeger_propagator::Propagator::new(),
+        );
+        let tracer = opentelemetry_otlp::new_pipeline()
+            .tracing()
+            .with_exporter(opentelemetry_otlp::new_exporter().tonic())
+            .install_batch(opentelemetry_sdk::runtime::Tokio)
             .unwrap();
         let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
 
@@ -141,6 +145,13 @@ async fn main() {
     }
 }
 
+/// Adds additional headers to prevent any potential XSS attacks via the media repo
+async fn set_csp_header(response: Response) -> impl IntoResponse {
+    (
+        [(CONTENT_SECURITY_POLICY, "sandbox; default-src 'none'; script-src 'none'; plugin-types application/pdf; style-src 'unsafe-inline'; object-src 'self';")], response
+    )
+}
+
 async fn run_server() -> io::Result<()> {
     let config = &services().globals.config;
     let addr = SocketAddr::from((config.address, config.port));
@@ -181,6 +192,7 @@ async fn run_server() -> io::Result<()> {
                 ])
                 .max_age(Duration::from_secs(86400)),
         )
+        .layer(map_response(set_csp_header))
         .layer(DefaultBodyLimit::max(
             config
                 .max_request_size
@@ -216,10 +228,10 @@ async fn run_server() -> io::Result<()> {
     Ok(())
 }
 
-async fn spawn_task<B: Send + 'static>(
-    req: http::Request<B>,
-    next: axum::middleware::Next<B>,
-) -> std::result::Result<axum::response::Response, StatusCode> {
+async fn spawn_task(
+    req: http::Request<Body>,
+    next: axum::middleware::Next,
+) -> std::result::Result<Response, StatusCode> {
     if services().globals.shutdown.load(atomic::Ordering::Relaxed) {
         return Err(StatusCode::SERVICE_UNAVAILABLE);
     }
@@ -228,10 +240,10 @@ async fn spawn_task<B: Send + 'static>(
         .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
 }
 
-async fn unrecognized_method<B: Send>(
-    req: http::Request<B>,
-    next: axum::middleware::Next<B>,
-) -> std::result::Result<axum::response::Response, StatusCode> {
+async fn unrecognized_method(
+    req: http::Request<Body>,
+    next: axum::middleware::Next,
+) -> std::result::Result<Response, StatusCode> {
     let method = req.method().clone();
     let uri = req.uri().clone();
     let inner = next.run(req).await;
@@ -277,6 +289,7 @@ fn routes(config: &Config) -> Router {
         .ruma_route(client_server::get_room_aliases_route)
         .ruma_route(client_server::get_filter_route)
         .ruma_route(client_server::create_filter_route)
+        .ruma_route(client_server::create_openid_token_route)
         .ruma_route(client_server::set_global_account_data_route)
         .ruma_route(client_server::set_room_account_data_route)
         .ruma_route(client_server::get_global_account_data_route)
@@ -366,10 +379,14 @@ fn routes(config: &Config) -> Router {
         .ruma_route(client_server::turn_server_route)
         .ruma_route(client_server::send_event_to_device_route)
         .ruma_route(client_server::get_media_config_route)
+        .ruma_route(client_server::get_media_config_auth_route)
         .ruma_route(client_server::create_content_route)
         .ruma_route(client_server::get_content_route)
+        .ruma_route(client_server::get_content_auth_route)
         .ruma_route(client_server::get_content_as_filename_route)
+        .ruma_route(client_server::get_content_as_filename_auth_route)
         .ruma_route(client_server::get_content_thumbnail_route)
+        .ruma_route(client_server::get_content_thumbnail_auth_route)
         .ruma_route(client_server::get_devices_route)
         .ruma_route(client_server::get_device_route)
         .ruma_route(client_server::update_device_route)
@@ -427,10 +444,13 @@ fn routes(config: &Config) -> Router {
             .ruma_route(server_server::create_join_event_v2_route)
             .ruma_route(server_server::create_invite_route)
             .ruma_route(server_server::get_devices_route)
+            .ruma_route(server_server::get_content_route)
+            .ruma_route(server_server::get_content_thumbnail_route)
             .ruma_route(server_server::get_room_information_route)
             .ruma_route(server_server::get_profile_information_route)
             .ruma_route(server_server::get_keys_route)
             .ruma_route(server_server::claim_keys_route)
+            .ruma_route(server_server::get_openid_userinfo_route)
             .ruma_route(server_server::well_known_server)
     } else {
         router

src/service/admin/mod.rs

@@ -1,9 +1,4 @@
-use std::{
-    collections::BTreeMap,
-    convert::{TryFrom, TryInto},
-    sync::Arc,
-    time::Instant,
-};
+use std::{collections::BTreeMap, convert::TryFrom, sync::Arc, time::Instant};
 
 use clap::Parser;
 use regex::Regex;
@@ -24,7 +19,8 @@ use ruma::{
         },
         TimelineEventType,
     },
-    EventId, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
+    EventId, MilliSecondsSinceUnixEpoch, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId,
+    RoomVersionId, ServerName, UserId,
 };
 use serde_json::value::to_raw_value;
 use tokio::sync::{mpsc, Mutex, RwLock};
@@ -77,6 +73,12 @@ enum AdminCommand {
     /// List all rooms we are currently handling an incoming pdu from
     IncomingFederation,
 
+    /// Removes an alias from the server
+    RemoveAlias {
+        /// The alias to be removed
+        alias: Box<RoomAliasId>,
+    },
+
     /// Deactivate a user
     ///
     /// User will not be removed from all rooms by default.
@@ -160,24 +162,23 @@ enum AdminCommand {
         password: Option<String>,
     },
 
+    /// Temporarily toggle user registration by passing either true or false as an argument, does not persist between restarts
+    AllowRegistration { status: Option<bool> },
+
     /// Disables incoming federation handling for a room.
     DisableRoom { room_id: Box<RoomId> },
     /// Enables incoming federation handling for a room again.
     EnableRoom { room_id: Box<RoomId> },
 
-    /// Verify json signatures
-    /// [commandbody]()
-    /// # ```
-    /// # json here
-    /// # ```
+    /// Sign a json object using Conduit's signing keys, putting the json in a codeblock
     SignJson,
 
-    /// Verify json signatures
-    /// [commandbody]()
-    /// # ```
-    /// # json here
-    /// # ```
+    /// Verify json signatures, putting the json in a codeblock
     VerifyJson,
+
+    /// Parses a JSON object as an event then creates a hash and signs it, putting a room
+    /// version as an argument, and the json in a codeblock
+    HashAndSignEvent { room_version_id: RoomVersionId },
 }
 
 #[derive(Debug)]
@@ -212,8 +213,7 @@ impl Service {
         // TODO: Use futures when we have long admin commands
         //let mut futures = FuturesUnordered::new();
 
-        let conduit_user = UserId::parse(format!("@conduit:{}", services().globals.server_name()))
-            .expect("@conduit:server_name is valid");
+        let conduit_user = services().globals.server_user();
 
         if let Ok(Some(conduit_room)) = services().admin.get_admin_room() {
             loop {
@@ -246,8 +246,9 @@ impl Service {
                                     unsigned: None,
                                     state_key: None,
                                     redacts: None,
+                                    timestamp: None,
                                 },
-                                &conduit_user,
+                                conduit_user,
                                 &conduit_room,
                                 &state_lock,
                             )
@@ -660,6 +661,24 @@ impl Service {
                     "Created user with user_id: {user_id} and password: {password}"
                 ))
             }
+            AdminCommand::AllowRegistration { status } => {
+                if let Some(status) = status {
+                    services().globals.set_registration(status).await;
+                    RoomMessageEventContent::text_plain(if status {
+                        "Registration is now enabled"
+                    } else {
+                        "Registration is now disabled"
+                    })
+                } else {
+                    RoomMessageEventContent::text_plain(
+                        if services().globals.allow_registration().await {
+                            "Registration is currently enabled"
+                        } else {
+                            "Registration is currently disabled"
+                        },
+                    )
+                }
+            }
             AdminCommand::DisableRoom { room_id } => {
                 services().rooms.metadata.disable_room(&room_id, true)?;
                 RoomMessageEventContent::text_plain("Room disabled.")
@@ -841,15 +860,46 @@ impl Service {
                             services()
                                 .rooms
                                 .event_handler
+                                // Generally we shouldn't be checking against expired keys unless required, so in the admin
+                                // room it might be best to not allow expired keys
                                 .fetch_required_signing_keys(&value, &pub_key_map)
                                 .await?;
 
-                            let pub_key_map = pub_key_map.read().await;
-                            match ruma::signatures::verify_json(&pub_key_map, &value) {
-                                Ok(_) => RoomMessageEventContent::text_plain("Signature correct"),
-                                Err(e) => RoomMessageEventContent::text_plain(format!(
+                            let mut expired_key_map = BTreeMap::new();
+                            let mut valid_key_map = BTreeMap::new();
+
+                            for (server, keys) in pub_key_map.into_inner().into_iter() {
+                                if keys.valid_until_ts > MilliSecondsSinceUnixEpoch::now() {
+                                    valid_key_map.insert(
+                                        server,
+                                        keys.verify_keys
+                                            .into_iter()
+                                            .map(|(id, key)| (id, key.key))
+                                            .collect(),
+                                    );
+                                } else {
+                                    expired_key_map.insert(
+                                        server,
+                                        keys.verify_keys
+                                            .into_iter()
+                                            .map(|(id, key)| (id, key.key))
+                                            .collect(),
+                                    );
+                                }
+                            }
+
+                            if ruma::signatures::verify_json(&valid_key_map, &value).is_ok() {
+                                RoomMessageEventContent::text_plain("Signature correct")
+                            } else if let Err(e) =
+                                ruma::signatures::verify_json(&expired_key_map, &value)
+                            {
+                                RoomMessageEventContent::text_plain(format!(
                                     "Signature verification failed: {e}"
-                                )),
+                                ))
+                            } else {
+                                RoomMessageEventContent::text_plain(
+                                    "Signature correct (with expired keys)",
+                                )
                             }
                         }
                         Err(e) => RoomMessageEventContent::text_plain(format!("Invalid json: {e}")),
@@ -860,6 +910,61 @@ impl Service {
                     )
                 }
             }
+            AdminCommand::HashAndSignEvent { room_version_id } => {
+                if body.len() > 2
+                    // Language may be specified as part of the codeblock (e.g. "```json")
+                    && body[0].trim().starts_with("```")
+                    && body.last().unwrap().trim() == "```"
+                {
+                    let string = body[1..body.len() - 1].join("\n");
+                    match serde_json::from_str(&string) {
+                        Ok(mut value) => {
+                            if let Err(e) = ruma::signatures::hash_and_sign_event(
+                                services().globals.server_name().as_str(),
+                                services().globals.keypair(),
+                                &mut value,
+                                &room_version_id,
+                            ) {
+                                RoomMessageEventContent::text_plain(format!("Invalid event: {e}"))
+                            } else {
+                                let json_text = serde_json::to_string_pretty(&value)
+                                    .expect("canonical json is valid json");
+                                RoomMessageEventContent::text_plain(json_text)
+                            }
+                        }
+                        Err(e) => RoomMessageEventContent::text_plain(format!("Invalid json: {e}")),
+                    }
+                } else {
+                    RoomMessageEventContent::text_plain(
+                        "Expected code block in command body. Add --help for details.",
+                    )
+                }
+            }
+            AdminCommand::RemoveAlias { alias } => {
+                if alias.server_name() != services().globals.server_name() {
+                    RoomMessageEventContent::text_plain(
+                        "Cannot remove alias which is not from this server",
+                    )
+                } else if services()
+                    .rooms
+                    .alias
+                    .resolve_local_alias(&alias)?
+                    .is_none()
+                {
+                    RoomMessageEventContent::text_plain("No such alias exists")
+                } else {
+                    // We execute this as the server user for two reasons
+                    // 1. If the user can execute commands in the admin room, they can always remove the alias.
+                    // 2. In the future, we are likely going to be able to allow users to execute commands via
+                    //   other methods, such as IPC, which would lead to us not knowing their user id
+
+                    services()
+                        .rooms
+                        .alias
+                        .remove_alias(&alias, services().globals.server_user())?;
+                    RoomMessageEventContent::text_plain("Alias removed sucessfully")
+                }
+            }
         };
 
         Ok(reply_message_content)
@@ -967,11 +1072,9 @@ impl Service {
         let state_lock = mutex_state.lock().await;
 
         // Create a user for the server
-        let conduit_user =
-            UserId::parse_with_server_name("conduit", services().globals.server_name())
-                .expect("@conduit:server_name is valid");
+        let conduit_user = services().globals.server_user();
 
-        services().users.create(&conduit_user, None)?;
+        services().users.create(conduit_user, None)?;
 
         let room_version = services().globals.default_room_version();
         let mut content = match room_version {
@@ -984,7 +1087,7 @@ impl Service {
             | RoomVersionId::V7
             | RoomVersionId::V8
             | RoomVersionId::V9
-            | RoomVersionId::V10 => RoomCreateEventContent::new_v1(conduit_user.clone()),
+            | RoomVersionId::V10 => RoomCreateEventContent::new_v1(conduit_user.to_owned()),
             RoomVersionId::V11 => RoomCreateEventContent::new_v11(),
             _ => unreachable!("Validity of room version already checked"),
         };
@@ -1003,8 +1106,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1031,8 +1135,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some(conduit_user.to_string()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1040,7 +1145,7 @@ impl Service {
 
         // 3. Power levels
         let mut users = BTreeMap::new();
-        users.insert(conduit_user.clone(), 100.into());
+        users.insert(conduit_user.to_owned(), 100.into());
 
         services()
             .rooms
@@ -1056,8 +1161,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1075,8 +1181,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1096,8 +1203,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1117,8 +1225,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1137,8 +1246,9 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
@@ -1157,17 +1267,16 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                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");
+        let alias: OwnedRoomAliasId = services().globals.admin_alias().to_owned();
 
         services()
             .rooms
@@ -1183,14 +1292,18 @@ impl Service {
                     unsigned: None,
                     state_key: Some("".to_owned()),
                     redacts: None,
+                    timestamp: None,
                 },
-                &conduit_user,
+                conduit_user,
                 &room_id,
                 &state_lock,
             )
             .await?;
 
-        services().rooms.alias.set_alias(&alias, &room_id)?;
+        services()
+            .rooms
+            .alias
+            .set_alias(&alias, &room_id, conduit_user)?;
 
         Ok(())
     }
@@ -1199,15 +1312,10 @@ impl Service {
     ///
     /// Errors are propagated from the database, and will have None if there is no admin room
     pub(crate) fn get_admin_room(&self) -> Result<Option<OwnedRoomId>> {
-        let admin_room_alias: Box<RoomAliasId> =
-            format!("#admins:{}", services().globals.server_name())
-                .try_into()
-                .expect("#admins:server_name is a valid alias name");
-
         services()
             .rooms
             .alias
-            .resolve_local_alias(&admin_room_alias)
+            .resolve_local_alias(services().globals.admin_alias())
     }
 
     /// Invite the user to the conduit admin room.
@@ -1231,9 +1339,7 @@ impl Service {
             let state_lock = mutex_state.lock().await;
 
             // Use the server user to grant the new admin's power level
-            let conduit_user =
-                UserId::parse_with_server_name("conduit", services().globals.server_name())
-                    .expect("@conduit:server_name is valid");
+            let conduit_user = services().globals.server_user();
 
             // Invite and join the real user
             services()
@@ -1256,8 +1362,9 @@ impl Service {
                         unsigned: None,
                         state_key: Some(user_id.to_string()),
                         redacts: None,
+                        timestamp: None,
                     },
-                    &conduit_user,
+                    conduit_user,
                     &room_id,
                     &state_lock,
                 )
@@ -1282,6 +1389,7 @@ impl Service {
                         unsigned: None,
                         state_key: Some(user_id.to_string()),
                         redacts: None,
+                        timestamp: None,
                     },
                     user_id,
                     &room_id,
@@ -1308,8 +1416,9 @@ impl Service {
                         unsigned: None,
                         state_key: Some("".to_owned()),
                         redacts: None,
+                        timestamp: None,
                     },
-                    &conduit_user,
+                    conduit_user,
                     &room_id,
                     &state_lock,
                 )
@@ -1327,14 +1436,24 @@ impl Service {
                 unsigned: None,
                 state_key: None,
                 redacts: None,
+                timestamp: None,
             },
-            &conduit_user,
+            conduit_user,
             &room_id,
             &state_lock,
         ).await?;
         }
         Ok(())
     }
+
+    /// Checks whether a given user is an admin of this server
+    pub fn user_is_admin(&self, user_id: &UserId) -> Result<bool> {
+        let Some(admin_room) = self.get_admin_room()? else {
+            return Ok(false);
+        };
+
+        services().rooms.state_cache.is_joined(user_id, &admin_room)
+    }
 }
 
 #[cfg(test)]

src/service/globals/data.rs

@@ -1,13 +1,71 @@
-use std::collections::BTreeMap;
-
-use async_trait::async_trait;
-use ruma::{
-    api::federation::discovery::{ServerSigningKeys, VerifyKey},
-    signatures::Ed25519KeyPair,
-    DeviceId, OwnedServerSigningKeyId, ServerName, UserId,
+use std::{
+    collections::BTreeMap,
+    time::{Duration, SystemTime},
 };
 
-use crate::Result;
+use crate::{services, Result};
+use async_trait::async_trait;
+use ruma::{
+    api::federation::discovery::{OldVerifyKey, ServerSigningKeys, VerifyKey},
+    serde::Base64,
+    signatures::Ed25519KeyPair,
+    DeviceId, MilliSecondsSinceUnixEpoch, ServerName, UserId,
+};
+use serde::Deserialize;
+
+/// Similar to ServerSigningKeys, but drops a few unnecessary fields we don't require post-validation
+#[derive(Deserialize, Debug, Clone)]
+pub struct SigningKeys {
+    pub verify_keys: BTreeMap<String, VerifyKey>,
+    pub old_verify_keys: BTreeMap<String, OldVerifyKey>,
+    pub valid_until_ts: MilliSecondsSinceUnixEpoch,
+}
+
+impl SigningKeys {
+    /// Creates the SigningKeys struct, using the keys of the current server
+    pub fn load_own_keys() -> Self {
+        let mut keys = Self {
+            verify_keys: BTreeMap::new(),
+            old_verify_keys: BTreeMap::new(),
+            valid_until_ts: MilliSecondsSinceUnixEpoch::from_system_time(
+                SystemTime::now() + Duration::from_secs(7 * 86400),
+            )
+            .expect("Should be valid until year 500,000,000"),
+        };
+
+        keys.verify_keys.insert(
+            format!("ed25519:{}", services().globals.keypair().version()),
+            VerifyKey {
+                key: Base64::new(services().globals.keypair.public_key().to_vec()),
+            },
+        );
+
+        keys
+    }
+}
+
+impl From<ServerSigningKeys> for SigningKeys {
+    fn from(value: ServerSigningKeys) -> Self {
+        let ServerSigningKeys {
+            verify_keys,
+            old_verify_keys,
+            valid_until_ts,
+            ..
+        } = value;
+
+        Self {
+            verify_keys: verify_keys
+                .into_iter()
+                .map(|(id, key)| (id.to_string(), key))
+                .collect(),
+            old_verify_keys: old_verify_keys
+                .into_iter()
+                .map(|(id, key)| (id.to_string(), key))
+                .collect(),
+            valid_until_ts,
+        }
+    }
+}
 
 #[async_trait]
 pub trait Data: Send + Sync {
@@ -21,17 +79,23 @@ pub trait Data: Send + Sync {
     fn clear_caches(&self, amount: u32);
     fn load_keypair(&self) -> Result<Ed25519KeyPair>;
     fn remove_keypair(&self) -> Result<()>;
-    fn add_signing_key(
+    /// Only extends the cached keys, not moving any verify_keys to old_verify_keys, as if we suddenly
+    /// recieve requests from the origin server, we want to be able to accept requests from them
+    fn add_signing_key_from_trusted_server(
         &self,
         origin: &ServerName,
         new_keys: ServerSigningKeys,
-    ) -> Result<BTreeMap<OwnedServerSigningKeyId, VerifyKey>>;
-
-    /// This returns an empty `Ok(BTreeMap<..>)` when there are no keys found for the server.
-    fn signing_keys_for(
+    ) -> Result<SigningKeys>;
+    /// Extends cached keys, as well as moving verify_keys that are not present in these new keys to
+    /// old_verify_keys, so that potnetially comprimised keys cannot be used to make requests
+    fn add_signing_key_from_origin(
         &self,
         origin: &ServerName,
-    ) -> Result<BTreeMap<OwnedServerSigningKeyId, VerifyKey>>;
+        new_keys: ServerSigningKeys,
+    ) -> Result<SigningKeys>;
+
+    /// This returns an empty `Ok(BTreeMap<..>)` when there are no keys found for the server.
+    fn signing_keys_for(&self, origin: &ServerName) -> Result<Option<SigningKeys>>;
     fn database_version(&self) -> Result<u64>;
     fn bump_database_version(&self, new_version: u64) -> Result<()>;
 }

src/service/globals/mod.rs

@@ -1,25 +1,19 @@
 mod data;
-pub use data::Data;
+pub use data::{Data, SigningKeys};
 use ruma::{
-    serde::Base64, OwnedDeviceId, OwnedEventId, OwnedRoomId, OwnedServerName,
-    OwnedServerSigningKeyId, OwnedUserId,
+    serde::Base64, MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedEventId, OwnedRoomAliasId,
+    OwnedRoomId, OwnedServerName, OwnedUserId, RoomAliasId,
 };
 
-use crate::api::server_server::FedDest;
+use crate::api::server_server::DestinationResponse;
 
 use crate::{services, Config, Error, Result};
 use futures_util::FutureExt;
 use hickory_resolver::TokioAsyncResolver;
-use hyper::{
-    client::connect::dns::{GaiResolver, Name},
-    service::Service as HyperService,
-};
-use reqwest::dns::{Addrs, Resolve, Resolving};
+use hyper_util::client::legacy::connect::dns::{GaiResolver, Name as HyperName};
+use reqwest::dns::{Addrs, Name, Resolve, Resolving};
 use ruma::{
-    api::{
-        client::sync::sync_events,
-        federation::discovery::{ServerSigningKeys, VerifyKey},
-    },
+    api::{client::sync::sync_events, federation::discovery::ServerSigningKeys},
     DeviceId, RoomVersionId, ServerName, UserId,
 };
 use std::{
@@ -30,6 +24,7 @@ use std::{
     iter,
     net::{IpAddr, SocketAddr},
     path::PathBuf,
+    str::FromStr,
     sync::{
         atomic::{self, AtomicBool},
         Arc, RwLock as StdRwLock,
@@ -37,11 +32,12 @@ use std::{
     time::{Duration, Instant},
 };
 use tokio::sync::{broadcast, watch::Receiver, Mutex, RwLock, Semaphore};
+use tower_service::Service as TowerService;
 use tracing::{error, info};
 
 use base64::{engine::general_purpose, Engine as _};
 
-type WellKnownMap = HashMap<OwnedServerName, (FedDest, String)>;
+type WellKnownMap = HashMap<OwnedServerName, DestinationResponse>;
 type TlsNameMap = HashMap<String, (Vec<IpAddr>, u16)>;
 type RateLimitState = (Instant, u32); // Time if last failed try, number of failed tries
 type SyncHandle = (
@@ -55,6 +51,7 @@ pub struct Service {
     pub actual_destination_cache: Arc<RwLock<WellKnownMap>>, // actual_destination, host
     pub tls_name_override: Arc<StdRwLock<TlsNameMap>>,
     pub config: Config,
+    allow_registration: RwLock<bool>,
     keypair: Arc<ruma::signatures::Ed25519KeyPair>,
     dns_resolver: TokioAsyncResolver,
     jwt_decoding_key: Option<jsonwebtoken::DecodingKey>,
@@ -71,6 +68,8 @@ pub struct Service {
     pub roomid_mutex_state: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>,
     pub roomid_mutex_federation: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>, // this lock will be held longer
     pub roomid_federationhandletime: RwLock<HashMap<OwnedRoomId, (OwnedEventId, Instant)>>,
+    server_user: OwnedUserId,
+    admin_alias: OwnedRoomAliasId,
     pub stateres_mutex: Arc<Mutex<()>>,
     pub rotate: RotationHandler,
 
@@ -137,11 +136,19 @@ impl Resolve for Resolver {
             })
             .unwrap_or_else(|| {
                 let this = &mut self.inner.clone();
-                Box::pin(HyperService::<Name>::call(this, name).map(|result| {
+                Box::pin(
+                    TowerService::<HyperName>::call(
+                        this,
+                        // Beautiful hack, please remove this in the future.
+                        HyperName::from_str(name.as_str())
+                            .expect("reqwest Name is just wrapper for hyper-util Name"),
+                    )
+                    .map(|result| {
                         result
                             .map(|addrs| -> Addrs { Box::new(addrs) })
                             .map_err(|err| -> Box<dyn StdError + Send + Sync> { Box::new(err) })
-                }))
+                    }),
+                )
             })
     }
 }
@@ -184,6 +191,11 @@ impl Service {
         let unstable_room_versions = vec![RoomVersionId::V3, RoomVersionId::V4, RoomVersionId::V5];
 
         let mut s = Self {
+            allow_registration: RwLock::new(config.allow_registration),
+            admin_alias: RoomAliasId::parse(format!("#admins:{}", &config.server_name))
+                .expect("#admins:server_name is a valid alias name"),
+            server_user: UserId::parse(format!("@conduit:{}", &config.server_name))
+                .expect("@conduit:server_name is valid"),
             db,
             config,
             keypair: Arc::new(keypair),
@@ -277,6 +289,14 @@ impl Service {
         self.config.server_name.as_ref()
     }
 
+    pub fn server_user(&self) -> &UserId {
+        self.server_user.as_ref()
+    }
+
+    pub fn admin_alias(&self) -> &RoomAliasId {
+        self.admin_alias.as_ref()
+    }
+
     pub fn max_request_size(&self) -> u32 {
         self.config.max_request_size
     }
@@ -285,8 +305,15 @@ impl Service {
         self.config.max_fetch_prev_events
     }
 
-    pub fn allow_registration(&self) -> bool {
-        self.config.allow_registration
+    /// Allows for the temporary (non-persistant) toggling of registration
+    pub async fn set_registration(&self, status: bool) {
+        let mut lock = self.allow_registration.write().await;
+        *lock = status;
+    }
+
+    /// Checks whether user registration is allowed
+    pub async fn allow_registration(&self) -> bool {
+        *self.allow_registration.read().await
     }
 
     pub fn allow_encryption(&self) -> bool {
@@ -362,36 +389,89 @@ impl Service {
         room_versions
     }
 
-    /// TODO: the key valid until timestamp is only honored in room version > 4
-    /// Remove the outdated keys and insert the new ones.
-    ///
     /// This doesn't actually check that the keys provided are newer than the old set.
-    pub fn add_signing_key(
+    pub fn add_signing_key_from_trusted_server(
         &self,
         origin: &ServerName,
         new_keys: ServerSigningKeys,
-    ) -> Result<BTreeMap<OwnedServerSigningKeyId, VerifyKey>> {
-        self.db.add_signing_key(origin, new_keys)
+    ) -> Result<SigningKeys> {
+        self.db
+            .add_signing_key_from_trusted_server(origin, new_keys)
     }
 
-    /// This returns an empty `Ok(BTreeMap<..>)` when there are no keys found for the server.
-    pub fn signing_keys_for(
+    /// Same as from_trusted_server, except it will move active keys not present in `new_keys` to old_signing_keys
+    pub fn add_signing_key_from_origin(
         &self,
         origin: &ServerName,
-    ) -> Result<BTreeMap<OwnedServerSigningKeyId, VerifyKey>> {
-        let mut keys = self.db.signing_keys_for(origin)?;
-        if origin == self.server_name() {
-            keys.insert(
-                format!("ed25519:{}", services().globals.keypair().version())
-                    .try_into()
-                    .expect("found invalid server signing keys in DB"),
-                VerifyKey {
-                    key: Base64::new(self.keypair.public_key().to_vec()),
-                },
-            );
+        new_keys: ServerSigningKeys,
+    ) -> Result<SigningKeys> {
+        self.db.add_signing_key_from_origin(origin, new_keys)
     }
 
-        Ok(keys)
+    /// This returns Ok(None) when there are no keys found for the server.
+    pub fn signing_keys_for(&self, origin: &ServerName) -> Result<Option<SigningKeys>> {
+        Ok(self.db.signing_keys_for(origin)?.or_else(|| {
+            if origin == self.server_name() {
+                Some(SigningKeys::load_own_keys())
+            } else {
+                None
+            }
+        }))
+    }
+
+    /// Filters the key map of multiple servers down to keys that should be accepted given the expiry time,
+    /// room version, and timestamp of the paramters
+    pub fn filter_keys_server_map(
+        &self,
+        keys: BTreeMap<String, SigningKeys>,
+        timestamp: MilliSecondsSinceUnixEpoch,
+        room_version_id: &RoomVersionId,
+    ) -> BTreeMap<String, BTreeMap<String, Base64>> {
+        keys.into_iter()
+            .filter_map(|(server, keys)| {
+                self.filter_keys_single_server(keys, timestamp, room_version_id)
+                    .map(|keys| (server, keys))
+            })
+            .collect()
+    }
+
+    /// Filters the keys of a single server down to keys that should be accepted given the expiry time,
+    /// room version, and timestamp of the paramters
+    pub fn filter_keys_single_server(
+        &self,
+        keys: SigningKeys,
+        timestamp: MilliSecondsSinceUnixEpoch,
+        room_version_id: &RoomVersionId,
+    ) -> Option<BTreeMap<String, Base64>> {
+        if keys.valid_until_ts > timestamp
+            // valid_until_ts MUST be ignored in room versions 1, 2, 3, and 4.
+            // https://spec.matrix.org/v1.10/server-server-api/#get_matrixkeyv2server
+                || matches!(room_version_id, RoomVersionId::V1
+                    | RoomVersionId::V2
+                    | RoomVersionId::V4
+                    | RoomVersionId::V3)
+        {
+            // Given that either the room version allows stale keys, or the valid_until_ts is
+            // in the future, all verify_keys are valid
+            let mut map: BTreeMap<_, _> = keys
+                .verify_keys
+                .into_iter()
+                .map(|(id, key)| (id, key.key))
+                .collect();
+
+            map.extend(keys.old_verify_keys.into_iter().filter_map(|(id, key)| {
+                // Even on old room versions, we don't allow old keys if they are expired
+                if key.expired_ts > timestamp {
+                    Some((id, key.key))
+                } else {
+                    None
+                }
+            }));
+
+            Some(map)
+        } else {
+            None
+        }
     }
 
     pub fn database_version(&self) -> Result<u64> {

src/service/media/data.rs

@@ -1,3 +1,5 @@
+use ruma::http_headers::ContentDisposition;
+
 use crate::Result;
 
 pub trait Data: Send + Sync {
@@ -6,7 +8,7 @@ pub trait Data: Send + Sync {
         mxc: String,
         width: u32,
         height: u32,
-        content_disposition: Option<&str>,
+        content_disposition: &ContentDisposition,
         content_type: Option<&str>,
     ) -> Result<Vec<u8>>;
 
@@ -16,5 +18,5 @@ pub trait Data: Send + Sync {
         mxc: String,
         width: u32,
         height: u32,
-    ) -> Result<(Option<String>, Option<String>, Vec<u8>)>;
+    ) -> Result<(ContentDisposition, Option<String>, Vec<u8>)>;
 }

src/service/media/mod.rs

@@ -2,6 +2,7 @@ mod data;
 use std::io::Cursor;
 
 pub use data::Data;
+use ruma::http_headers::{ContentDisposition, ContentDispositionType};
 
 use crate::{services, Result};
 use image::imageops::FilterType;
@@ -12,7 +13,7 @@ use tokio::{
 };
 
 pub struct FileMeta {
-    pub content_disposition: Option<String>,
+    pub content_disposition: ContentDisposition,
     pub content_type: Option<String>,
     pub file: Vec<u8>,
 }
@@ -26,14 +27,17 @@ impl Service {
     pub async fn create(
         &self,
         mxc: String,
-        content_disposition: Option<&str>,
+        content_disposition: Option<ContentDisposition>,
         content_type: Option<&str>,
         file: &[u8],
     ) -> Result<()> {
+        let content_disposition =
+            content_disposition.unwrap_or(ContentDisposition::new(ContentDispositionType::Inline));
+
         // Width, Height = 0 if it's not a thumbnail
         let key = self
             .db
-            .create_file_metadata(mxc, 0, 0, content_disposition, content_type)?;
+            .create_file_metadata(mxc, 0, 0, &content_disposition, content_type)?;
 
         let path = services().globals.get_media_file(&key);
         let mut f = File::create(path).await?;
@@ -46,15 +50,18 @@ impl Service {
     pub async fn upload_thumbnail(
         &self,
         mxc: String,
-        content_disposition: Option<&str>,
         content_type: Option<&str>,
         width: u32,
         height: u32,
         file: &[u8],
     ) -> Result<()> {
-        let key =
-            self.db
-                .create_file_metadata(mxc, width, height, content_disposition, content_type)?;
+        let key = self.db.create_file_metadata(
+            mxc,
+            width,
+            height,
+            &ContentDisposition::new(ContentDispositionType::Inline),
+            content_type,
+        )?;
 
         let path = services().globals.get_media_file(&key);
         let mut f = File::create(path).await?;
@@ -166,22 +173,20 @@ impl Service {
                                 / u64::from(original_height)
                         };
                         if use_width {
-                            if intermediate <= u64::from(::std::u32::MAX) {
+                            if intermediate <= u64::from(u32::MAX) {
                                 (width, intermediate as u32)
                             } else {
                                 (
-                                    (u64::from(width) * u64::from(::std::u32::MAX) / intermediate)
-                                        as u32,
-                                    ::std::u32::MAX,
+                                    (u64::from(width) * u64::from(u32::MAX) / intermediate) as u32,
+                                    u32::MAX,
                                 )
                             }
-                        } else if intermediate <= u64::from(::std::u32::MAX) {
+                        } else if intermediate <= u64::from(u32::MAX) {
                             (intermediate as u32, height)
                         } else {
                             (
-                                ::std::u32::MAX,
-                                (u64::from(height) * u64::from(::std::u32::MAX) / intermediate)
-                                    as u32,
+                                u32::MAX,
+                                (u64::from(height) * u64::from(u32::MAX) / intermediate) as u32,
                             )
                         }
                     };
@@ -200,7 +205,7 @@ impl Service {
                     mxc,
                     width,
                     height,
-                    content_disposition.as_deref(),
+                    &content_disposition,
                     content_type.as_deref(),
                 )?;
 

src/service/pdu.rs

@@ -1,5 +1,6 @@
 use crate::Error;
 use ruma::{
+    api::client::error::ErrorKind,
     canonical_json::redact_content_in_place,
     events::{
         room::{member::RoomMemberEventContent, redaction::RoomRedactionEventContent},
@@ -72,6 +73,23 @@ impl PduEvent {
         Ok(())
     }
 
+    pub fn is_redacted(&self) -> bool {
+        #[derive(Deserialize)]
+        struct ExtractRedactedBecause {
+            redacted_because: Option<serde::de::IgnoredAny>,
+        }
+
+        let Some(unsigned) = &self.unsigned else {
+            return false;
+        };
+
+        let Ok(unsigned) = ExtractRedactedBecause::deserialize(&**unsigned) else {
+            return false;
+        };
+
+        unsigned.redacted_because.is_some()
+    }
+
     pub fn remove_transaction_id(&mut self) -> crate::Result<()> {
         if let Some(unsigned) = &self.unsigned {
             let mut unsigned: BTreeMap<String, Box<RawJsonValue>> =
@@ -426,7 +444,7 @@ pub(crate) fn gen_event_id_canonical_json(
         "${}",
         // Anything higher than version3 behaves the same
         ruma::signatures::reference_hash(&value, room_version_id)
-            .expect("ruma can calculate reference hashes")
+            .map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Invalid PDU format"))?
     )
     .try_into()
     .expect("ruma's reference hashes are valid event ids");
@@ -443,4 +461,8 @@ pub struct PduBuilder {
     pub unsigned: Option<BTreeMap<String, serde_json::Value>>,
     pub state_key: Option<String>,
     pub redacts: Option<Arc<EventId>>,
+    /// For timestamped messaging, should only be used for appservices
+    ///
+    /// Will be set to current time if None
+    pub timestamp: Option<MilliSecondsSinceUnixEpoch>,
 }

src/service/rooms/alias/data.rs

@@ -1,9 +1,12 @@
 use crate::Result;
-use ruma::{OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId};
+use ruma::{OwnedRoomAliasId, OwnedRoomId, OwnedUserId, RoomAliasId, RoomId, UserId};
 
 pub trait Data: Send + Sync {
     /// Creates or updates the alias to the given room id.
-    fn set_alias(&self, alias: &RoomAliasId, room_id: &RoomId) -> Result<()>;
+    fn set_alias(&self, alias: &RoomAliasId, room_id: &RoomId, user_id: &UserId) -> Result<()>;
+
+    /// Finds the user who assigned the given alias to a room
+    fn who_created_alias(&self, alias: &RoomAliasId) -> Result<Option<OwnedUserId>>;
 
     /// Forgets about an alias. Returns an error if the alias did not exist.
     fn remove_alias(&self, alias: &RoomAliasId) -> Result<()>;

src/service/rooms/alias/mod.rs

@@ -1,9 +1,17 @@
 mod data;
 
 pub use data::Data;
+use tracing::error;
 
-use crate::Result;
-use ruma::{OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId};
+use crate::{services, Error, Result};
+use ruma::{
+    api::client::error::ErrorKind,
+    events::{
+        room::power_levels::{RoomPowerLevels, RoomPowerLevelsEventContent},
+        StateEventType,
+    },
+    OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId, UserId,
+};
 
 pub struct Service {
     pub db: &'static dyn Data,
@@ -11,13 +19,71 @@ pub struct Service {
 
 impl Service {
     #[tracing::instrument(skip(self))]
-    pub fn set_alias(&self, alias: &RoomAliasId, room_id: &RoomId) -> Result<()> {
-        self.db.set_alias(alias, room_id)
+    pub fn set_alias(&self, alias: &RoomAliasId, room_id: &RoomId, user_id: &UserId) -> Result<()> {
+        if alias == services().globals.admin_alias() && user_id != services().globals.server_user()
+        {
+            Err(Error::BadRequest(
+                ErrorKind::forbidden(),
+                "Only the server user can set this alias",
+            ))
+        } else {
+            self.db.set_alias(alias, room_id, user_id)
+        }
     }
 
     #[tracing::instrument(skip(self))]
-    pub fn remove_alias(&self, alias: &RoomAliasId) -> Result<()> {
+    fn user_can_remove_alias(&self, alias: &RoomAliasId, user_id: &UserId) -> Result<bool> {
+        let Some(room_id) = self.resolve_local_alias(alias)? else {
+            return Err(Error::BadRequest(ErrorKind::NotFound, "Alias not found."));
+        };
+
+        // The creator of an alias can remove it
+        if self
+            .db
+            .who_created_alias(alias)?
+            .map(|user| user == user_id)
+            .unwrap_or_default()
+            // Server admins can remove any local alias
+            || services().admin.user_is_admin(user_id)?
+            // Always allow the Conduit user to remove the alias, since there may not be an admin room
+            || services().globals.server_user ()== user_id
+        {
+            Ok(true)
+            // Checking whether the user is able to change canonical aliases of the room
+        } else if let Some(event) = services().rooms.state_accessor.room_state_get(
+            &room_id,
+            &StateEventType::RoomPowerLevels,
+            "",
+        )? {
+            serde_json::from_str(event.content.get())
+                .map_err(|_| Error::bad_database("Invalid event content for m.room.power_levels"))
+                .map(|content: RoomPowerLevelsEventContent| {
+                    RoomPowerLevels::from(content)
+                        .user_can_send_state(user_id, StateEventType::RoomCanonicalAlias)
+                })
+        // If there is no power levels event, only the room creator can change canonical aliases
+        } else if let Some(event) = services().rooms.state_accessor.room_state_get(
+            &room_id,
+            &StateEventType::RoomCreate,
+            "",
+        )? {
+            Ok(event.sender == user_id)
+        } else {
+            error!("Room {} has no m.room.create event (VERY BAD)!", room_id);
+            Err(Error::bad_database("Room has no m.room.create event"))
+        }
+    }
+
+    #[tracing::instrument(skip(self))]
+    pub fn remove_alias(&self, alias: &RoomAliasId, user_id: &UserId) -> Result<()> {
+        if self.user_can_remove_alias(alias, user_id)? {
             self.db.remove_alias(alias)
+        } else {
+            Err(Error::BadRequest(
+                ErrorKind::forbidden(),
+                "User is not permitted to remove this alias.",
+            ))
+        }
     }
 
     #[tracing::instrument(skip(self))]

src/service/rooms/auth_chain/mod.rs

@@ -133,7 +133,10 @@ impl Service {
             match services().rooms.timeline.get_pdu(&event_id) {
                 Ok(Some(pdu)) => {
                     if pdu.room_id != room_id {
-                        return Err(Error::BadRequest(ErrorKind::Forbidden, "Evil event in db"));
+                        return Err(Error::BadRequest(
+                            ErrorKind::forbidden(),
+                            "Evil event in db",
+                        ));
                     }
                     for auth_event in &pdu.auth_events {
                         let sauthevent = services()

src/service/rooms/event_handler/mod.rs

@@ -9,6 +9,7 @@ use std::{
 };
 
 use futures_util::{stream::FuturesUnordered, Future, StreamExt};
+use globals::SigningKeys;
 use ruma::{
     api::{
         client::error::ErrorKind,
@@ -30,7 +31,6 @@ use ruma::{
         StateEventType, TimelineEventType,
     },
     int,
-    serde::Base64,
     state_res::{self, RoomVersion, StateMap},
     uint, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
     OwnedServerName, OwnedServerSigningKeyId, RoomId, RoomVersionId, ServerName,
@@ -78,7 +78,7 @@ impl Service {
         room_id: &'a RoomId,
         value: BTreeMap<String, CanonicalJsonValue>,
         is_timeline_event: bool,
-        pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &'a RwLock<BTreeMap<String, SigningKeys>>,
     ) -> Result<Option<Vec<u8>>> {
         // 0. Check the server is in the room
         if !services().rooms.metadata.exists(room_id)? {
@@ -90,7 +90,7 @@ impl Service {
 
         if services().rooms.metadata.is_disabled(room_id)? {
             return Err(Error::BadRequest(
-                ErrorKind::Forbidden,
+                ErrorKind::forbidden(),
                 "Federation of this room is currently disabled on this server.",
             ));
         }
@@ -162,7 +162,7 @@ impl Service {
             // Check for disabled again because it might have changed
             if services().rooms.metadata.is_disabled(room_id)? {
                 return Err(Error::BadRequest(
-                    ErrorKind::Forbidden,
+                    ErrorKind::forbidden(),
                     "Federation of this room is currently disabled on this server.",
                 ));
             }
@@ -304,19 +304,12 @@ impl Service {
         room_id: &'a RoomId,
         mut value: BTreeMap<String, CanonicalJsonValue>,
         auth_events_known: bool,
-        pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &'a RwLock<BTreeMap<String, SigningKeys>>,
     ) -> AsyncRecursiveType<'a, Result<(Arc<PduEvent>, BTreeMap<String, CanonicalJsonValue>)>> {
         Box::pin(async move {
             // 1.1. Remove unsigned field
             value.remove("unsigned");
 
-            // TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?: https://matrix.org/docs/spec/rooms/v6#canonical-json
-
-            // We go through all the signatures we see on the value and fetch the corresponding signing
-            // keys
-            self.fetch_required_signing_keys(&value, pub_key_map)
-                .await?;
-
             // 2. Check signatures, otherwise drop
             // 3. check content hash, redact if doesn't match
             let create_event_content: RoomCreateEventContent =
@@ -329,8 +322,47 @@ impl Service {
             let room_version =
                 RoomVersion::new(room_version_id).expect("room version is supported");
 
+            // TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?: https://matrix.org/docs/spec/rooms/v6#canonical-json
+
+            // We go through all the signatures we see on the value and fetch the corresponding signing
+            // keys
+            self.fetch_required_signing_keys(&value, pub_key_map)
+                .await?;
+
+            let origin_server_ts = value.get("origin_server_ts").ok_or_else(|| {
+                error!("Invalid PDU, no origin_server_ts field");
+                Error::BadRequest(
+                    ErrorKind::MissingParam,
+                    "Invalid PDU, no origin_server_ts field",
+                )
+            })?;
+
+            let origin_server_ts: MilliSecondsSinceUnixEpoch = {
+                let ts = origin_server_ts.as_integer().ok_or_else(|| {
+                    Error::BadRequest(
+                        ErrorKind::InvalidParam,
+                        "origin_server_ts must be an integer",
+                    )
+                })?;
+
+                MilliSecondsSinceUnixEpoch(i64::from(ts).try_into().map_err(|_| {
+                    Error::BadRequest(ErrorKind::InvalidParam, "Time must be after the unix epoch")
+                })?)
+            };
+
             let guard = pub_key_map.read().await;
-            let mut val = match ruma::signatures::verify_event(&guard, &value, room_version_id) {
+
+            let pkey_map = (*guard).clone();
+
+            // Removing all the expired keys, unless the room version allows stale keys
+            let filtered_keys = services().globals.filter_keys_server_map(
+                pkey_map,
+                origin_server_ts,
+                room_version_id,
+            );
+
+            let mut val =
+                match ruma::signatures::verify_event(&filtered_keys, &value, room_version_id) {
                     Err(e) => {
                         // Drop
                         warn!("Dropping bad event {}: {}", event_id, e,);
@@ -487,7 +519,7 @@ impl Service {
         create_event: &PduEvent,
         origin: &ServerName,
         room_id: &RoomId,
-        pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
     ) -> Result<Option<Vec<u8>>> {
         // Skip the PDU if we already have it as a timeline event
         if let Ok(Some(pduid)) = services().rooms.timeline.get_pdu_id(&incoming_pdu.event_id) {
@@ -1097,7 +1129,7 @@ impl Service {
         create_event: &'a PduEvent,
         room_id: &'a RoomId,
         room_version_id: &'a RoomVersionId,
-        pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &'a RwLock<BTreeMap<String, SigningKeys>>,
     ) -> AsyncRecursiveType<'a, Vec<(Arc<PduEvent>, Option<BTreeMap<String, CanonicalJsonValue>>)>>
     {
         Box::pin(async move {
@@ -1280,7 +1312,7 @@ impl Service {
         create_event: &PduEvent,
         room_id: &RoomId,
         room_version_id: &RoomVersionId,
-        pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
         initial_set: Vec<Arc<EventId>>,
     ) -> Result<(
         Vec<Arc<EventId>>,
@@ -1378,7 +1410,7 @@ impl Service {
     pub(crate) async fn fetch_required_signing_keys(
         &self,
         event: &BTreeMap<String, CanonicalJsonValue>,
-        pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
     ) -> Result<()> {
         let signatures = event
             .get("signatures")
@@ -1407,6 +1439,7 @@ impl Service {
                         )
                     })?,
                     signature_ids,
+                    true,
                 )
                 .await;
 
@@ -1434,7 +1467,7 @@ impl Service {
         pdu: &RawJsonValue,
         servers: &mut BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>,
         room_version: &RoomVersionId,
-        pub_key_map: &mut RwLockWriteGuard<'_, BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &mut RwLockWriteGuard<'_, BTreeMap<String, SigningKeys>>,
     ) -> Result<()> {
         let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
             error!("Invalid PDU in server response: {:?}: {:?}", pdu, e);
@@ -1444,7 +1477,7 @@ impl Service {
         let event_id = format!(
             "${}",
             ruma::signatures::reference_hash(&value, room_version)
-                .expect("ruma can calculate reference hashes")
+                .map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Invalid PDU format"))?
         );
         let event_id = <&EventId>::try_from(event_id.as_str())
             .expect("ruma's reference hashes are valid event ids");
@@ -1485,8 +1518,18 @@ impl Service {
 
             let signature_ids = signature_object.keys().cloned().collect::<Vec<_>>();
 
-            let contains_all_ids = |keys: &BTreeMap<String, Base64>| {
-                signature_ids.iter().all(|id| keys.contains_key(id))
+            let contains_all_ids = |keys: &SigningKeys| {
+                signature_ids.iter().all(|id| {
+                    keys.verify_keys
+                        .keys()
+                        .map(ToString::to_string)
+                        .any(|key_id| id == &key_id)
+                        || keys
+                            .old_verify_keys
+                            .keys()
+                            .map(ToString::to_string)
+                            .any(|key_id| id == &key_id)
+                })
             };
 
             let origin = <&ServerName>::try_from(signature_server.as_str()).map_err(|_| {
@@ -1499,13 +1542,7 @@ impl Service {
 
             trace!("Loading signing keys for {}", origin);
 
-            let result: BTreeMap<_, _> = services()
-                .globals
-                .signing_keys_for(origin)?
-                .into_iter()
-                .map(|(k, v)| (k.to_string(), v.key))
-                .collect();
-
+            if let Some(result) = services().globals.signing_keys_for(origin)? {
                 if !contains_all_ids(&result) {
                     trace!("Signing key not loaded for {}", origin);
                     servers.insert(origin.to_owned(), BTreeMap::new());
@@ -1513,6 +1550,7 @@ impl Service {
 
                 pub_key_map.insert(origin.to_string(), result);
             }
+        }
 
         Ok(())
     }
@@ -1521,7 +1559,7 @@ impl Service {
         &self,
         event: &create_join_event::v2::Response,
         room_version: &RoomVersionId,
-        pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
     ) -> Result<()> {
         let mut servers: BTreeMap<
             OwnedServerName,
@@ -1584,10 +1622,7 @@ impl Service {
 
                     let result = services()
                         .globals
-                        .add_signing_key(&k.server_name, k.clone())?
-                        .into_iter()
-                        .map(|(k, v)| (k.to_string(), v.key))
-                        .collect::<BTreeMap<_, _>>();
+                        .add_signing_key_from_trusted_server(&k.server_name, k.clone())?;
 
                     pkm.insert(k.server_name.to_string(), result);
                 }
@@ -1618,12 +1653,9 @@ impl Service {
             if let (Ok(get_keys_response), origin) = result {
                 info!("Result is from {origin}");
                 if let Ok(key) = get_keys_response.server_key.deserialize() {
-                    let result: BTreeMap<_, _> = services()
+                    let result = services()
                         .globals
-                        .add_signing_key(&origin, key)?
-                        .into_iter()
-                        .map(|(k, v)| (k.to_string(), v.key))
-                        .collect();
+                        .add_signing_key_from_origin(&origin, key)?;
                     pub_key_map.write().await.insert(origin.to_string(), result);
                 }
             }
@@ -1655,11 +1687,6 @@ impl Service {
                 }
             };
 
-        if acl_event_content.allow.is_empty() {
-            // Ignore broken acl events
-            return Ok(());
-        }
-
         if acl_event_content.is_allowed(server_name) {
             Ok(())
         } else {
@@ -1668,7 +1695,7 @@ impl Service {
                 server_name, room_id
             );
             Err(Error::BadRequest(
-                ErrorKind::Forbidden,
+                ErrorKind::forbidden(),
                 "Server was denied by room ACL",
             ))
         }
@@ -1681,9 +1708,23 @@ impl Service {
         &self,
         origin: &ServerName,
         signature_ids: Vec<String>,
-    ) -> Result<BTreeMap<String, Base64>> {
-        let contains_all_ids =
-            |keys: &BTreeMap<String, Base64>| signature_ids.iter().all(|id| keys.contains_key(id));
+        // Whether to ask for keys from trusted servers. Should be false when getting
+        // keys for validating requests, as per MSC4029
+        query_via_trusted_servers: bool,
+    ) -> Result<SigningKeys> {
+        let contains_all_ids = |keys: &SigningKeys| {
+            signature_ids.iter().all(|id| {
+                keys.verify_keys
+                    .keys()
+                    .map(ToString::to_string)
+                    .any(|key_id| id == &key_id)
+                    || keys
+                        .old_verify_keys
+                        .keys()
+                        .map(ToString::to_string)
+                        .any(|key_id| id == &key_id)
+            })
+        };
 
         let permit = services()
             .globals
@@ -1744,48 +1785,91 @@ impl Service {
 
         trace!("Loading signing keys for {}", origin);
 
-        let mut result: BTreeMap<_, _> = services()
-            .globals
-            .signing_keys_for(origin)?
-            .into_iter()
-            .map(|(k, v)| (k.to_string(), v.key))
-            .collect();
+        let result = services().globals.signing_keys_for(origin)?;
+
+        let mut expires_soon_or_has_expired = false;
+
+        if let Some(result) = result.clone() {
+            let ts_threshold = MilliSecondsSinceUnixEpoch::from_system_time(
+                SystemTime::now() + Duration::from_secs(30 * 60),
+            )
+            .expect("Should be valid until year 500,000,000");
+
+            debug!(
+                "The treshhold is {:?}, found time is {:?} for server {}",
+                ts_threshold, result.valid_until_ts, origin
+            );
 
             if contains_all_ids(&result) {
+                // We want to ensure that the keys remain valid by the time the other functions that handle signatures reach them
+                if result.valid_until_ts > ts_threshold {
+                    debug!(
+                        "Keys for {} are deemed as valid, as they expire at {:?}",
+                        &origin, &result.valid_until_ts
+                    );
                     return Ok(result);
                 }
 
+                expires_soon_or_has_expired = true;
+            }
+        }
+
+        let mut keys = result.unwrap_or_else(|| SigningKeys {
+            verify_keys: BTreeMap::new(),
+            old_verify_keys: BTreeMap::new(),
+            valid_until_ts: MilliSecondsSinceUnixEpoch::now(),
+        });
+
+        // We want to set this to the max, and then lower it whenever we see older keys
+        keys.valid_until_ts = MilliSecondsSinceUnixEpoch::from_system_time(
+            SystemTime::now() + Duration::from_secs(7 * 86400),
+        )
+        .expect("Should be valid until year 500,000,000");
+
         debug!("Fetching signing keys for {} over federation", origin);
 
-        if let Some(server_key) = services()
+        if let Some(mut server_key) = services()
             .sending
             .send_federation_request(origin, get_server_keys::v2::Request::new())
             .await
             .ok()
             .and_then(|resp| resp.server_key.deserialize().ok())
         {
+            // Keys should only be valid for a maximum of seven days
+            server_key.valid_until_ts = server_key.valid_until_ts.min(
+                MilliSecondsSinceUnixEpoch::from_system_time(
+                    SystemTime::now() + Duration::from_secs(7 * 86400),
+                )
+                .expect("Should be valid until year 500,000,000"),
+            );
+
             services()
                 .globals
-                .add_signing_key(origin, server_key.clone())?;
+                .add_signing_key_from_origin(origin, server_key.clone())?;
 
-            result.extend(
+            if keys.valid_until_ts > server_key.valid_until_ts {
+                keys.valid_until_ts = server_key.valid_until_ts;
+            }
+
+            keys.verify_keys.extend(
                 server_key
                     .verify_keys
                     .into_iter()
-                    .map(|(k, v)| (k.to_string(), v.key)),
+                    .map(|(id, key)| (id.to_string(), key)),
             );
-            result.extend(
+            keys.old_verify_keys.extend(
                 server_key
                     .old_verify_keys
                     .into_iter()
-                    .map(|(k, v)| (k.to_string(), v.key)),
+                    .map(|(id, key)| (id.to_string(), key)),
             );
 
-            if contains_all_ids(&result) {
-                return Ok(result);
+            if contains_all_ids(&keys) {
+                return Ok(keys);
             }
         }
 
+        if query_via_trusted_servers {
             for server in services().globals.trusted_servers() {
                 debug!("Asking {} for {}'s signing key", server, origin);
                 if let Some(server_keys) = services()
@@ -1812,25 +1896,60 @@ impl Service {
                     })
                 {
                     trace!("Got signing keys: {:?}", server_keys);
-                for k in server_keys {
-                    services().globals.add_signing_key(origin, k.clone())?;
-                    result.extend(
+                    for mut k in server_keys {
+                        if k.valid_until_ts
+                        // Half an hour should give plenty of time for the server to respond with keys that are still
+                        // valid, given we requested keys which are valid at least an hour from now
+                            < MilliSecondsSinceUnixEpoch::from_system_time(
+                                SystemTime::now() + Duration::from_secs(30 * 60),
+                            )
+                            .expect("Should be valid until year 500,000,000")
+                        {
+                            // Keys should only be valid for a maximum of seven days
+                            k.valid_until_ts = k.valid_until_ts.min(
+                                MilliSecondsSinceUnixEpoch::from_system_time(
+                                    SystemTime::now() + Duration::from_secs(7 * 86400),
+                                )
+                                .expect("Should be valid until year 500,000,000"),
+                            );
+
+                            if keys.valid_until_ts > k.valid_until_ts {
+                                keys.valid_until_ts = k.valid_until_ts;
+                            }
+
+                            services()
+                                .globals
+                                .add_signing_key_from_trusted_server(origin, k.clone())?;
+                            keys.verify_keys.extend(
                                 k.verify_keys
                                     .into_iter()
-                            .map(|(k, v)| (k.to_string(), v.key)),
+                                    .map(|(id, key)| (id.to_string(), key)),
                             );
-                    result.extend(
+                            keys.old_verify_keys.extend(
                                 k.old_verify_keys
                                     .into_iter()
-                            .map(|(k, v)| (k.to_string(), v.key)),
+                                    .map(|(id, key)| (id.to_string(), key)),
+                            );
+                        } else {
+                            warn!(
+                                "Server {} gave us keys older than we requested, valid until: {:?}",
+                                origin, k.valid_until_ts
                             );
                         }
 
-                if contains_all_ids(&result) {
-                    return Ok(result);
+                        if contains_all_ids(&keys) {
+                            return Ok(keys);
                         }
                     }
                 }
+            }
+        }
+
+        // We should return these keys if fresher keys were not found
+        if expires_soon_or_has_expired {
+            info!("Returning stale keys for {}", origin);
+            return Ok(keys);
+        }
 
         drop(permit);
 

src/service/rooms/pdu_metadata/mod.rs

@@ -3,9 +3,9 @@ use std::sync::Arc;
 
 pub use data::Data;
 use ruma::{
-    api::client::relations::get_relating_events,
+    api::{client::relations::get_relating_events, Direction},
     events::{relation::RelationType, TimelineEventType},
-    EventId, RoomId, UserId,
+    EventId, RoomId, UInt, UserId,
 };
 use serde::Deserialize;
 
@@ -48,26 +48,48 @@ impl Service {
         target: &EventId,
         filter_event_type: Option<TimelineEventType>,
         filter_rel_type: Option<RelationType>,
-        from: PduCount,
-        to: Option<PduCount>,
-        limit: usize,
+        from: Option<String>,
+        to: Option<String>,
+        limit: Option<UInt>,
+        recurse: bool,
+        dir: &Direction,
     ) -> Result<get_relating_events::v1::Response> {
+        let from = match from {
+            Some(from) => PduCount::try_from_string(&from)?,
+            None => match dir {
+                Direction::Forward => PduCount::min(),
+                Direction::Backward => PduCount::max(),
+            },
+        };
+
+        let to = to.as_ref().and_then(|t| PduCount::try_from_string(t).ok());
+
+        // Use limit or else 10, with maximum 100
+        let limit = limit
+            .and_then(|u| u32::try_from(u).ok())
+            .map_or(10_usize, |u| u as usize)
+            .min(100);
+
         let next_token;
 
-        //TODO: Fix ruma: match body.dir {
-        match ruma::api::Direction::Backward {
-            ruma::api::Direction::Forward => {
-                let events_after: Vec<_> = services()
-                    .rooms
-                    .pdu_metadata
-                    .relations_until(sender_user, room_id, target, from)? // TODO: should be relations_after
-                    .filter(|r| {
-                        r.as_ref().map_or(true, |(_, pdu)| {
+        // Spec (v1.10) recommends depth of at least 3
+        let depth: u8 = if recurse { 3 } else { 1 };
+
+        match dir {
+            Direction::Forward => {
+                let relations_until = &services().rooms.pdu_metadata.relations_until(
+                    sender_user,
+                    room_id,
+                    target,
+                    from,
+                    depth,
+                )?;
+                let events_after: Vec<_> = relations_until // TODO: should be relations_after
+                    .iter()
+                    .filter(|(_, pdu)| {
                         filter_event_type.as_ref().map_or(true, |t| &pdu.kind == t)
                             && if let Ok(content) =
-                                    serde_json::from_str::<ExtractRelatesToEventId>(
-                                        pdu.content.get(),
-                                    )
+                                serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get())
                             {
                                 filter_rel_type
                                     .as_ref()
@@ -76,9 +98,7 @@ impl Service {
                                 false
                             }
                     })
-                    })
                     .take(limit)
-                    .filter_map(|r| r.ok()) // Filter out buggy events
                     .filter(|(_, pdu)| {
                         services()
                             .rooms
@@ -86,7 +106,7 @@ impl Service {
                             .user_can_see_event(sender_user, room_id, &pdu.event_id)
                             .unwrap_or(false)
                     })
-                    .take_while(|&(k, _)| Some(k) != to) // Stop at `to`
+                    .take_while(|(k, _)| Some(k) != to.as_ref()) // Stop at `to`
                     .collect();
 
                 next_token = events_after.last().map(|(count, _)| count).copied();
@@ -101,20 +121,23 @@ impl Service {
                     chunk: events_after,
                     next_batch: next_token.map(|t| t.stringify()),
                     prev_batch: Some(from.stringify()),
+                    recursion_depth: if recurse { Some(depth.into()) } else { None },
                 })
             }
-            ruma::api::Direction::Backward => {
-                let events_before: Vec<_> = services()
-                    .rooms
-                    .pdu_metadata
-                    .relations_until(sender_user, room_id, target, from)?
-                    .filter(|r| {
-                        r.as_ref().map_or(true, |(_, pdu)| {
+            Direction::Backward => {
+                let relations_until = &services().rooms.pdu_metadata.relations_until(
+                    sender_user,
+                    room_id,
+                    target,
+                    from,
+                    depth,
+                )?;
+                let events_before: Vec<_> = relations_until
+                    .iter()
+                    .filter(|(_, pdu)| {
                         filter_event_type.as_ref().map_or(true, |t| &pdu.kind == t)
                             && if let Ok(content) =
-                                    serde_json::from_str::<ExtractRelatesToEventId>(
-                                        pdu.content.get(),
-                                    )
+                                serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get())
                             {
                                 filter_rel_type
                                     .as_ref()
@@ -123,9 +146,7 @@ impl Service {
                                 false
                             }
                     })
-                    })
                     .take(limit)
-                    .filter_map(|r| r.ok()) // Filter out buggy events
                     .filter(|(_, pdu)| {
                         services()
                             .rooms
@@ -133,7 +154,7 @@ impl Service {
                             .user_can_see_event(sender_user, room_id, &pdu.event_id)
                             .unwrap_or(false)
                     })
-                    .take_while(|&(k, _)| Some(k) != to) // Stop at `to`
+                    .take_while(|&(k, _)| Some(k) != to.as_ref()) // Stop at `to`
                     .collect();
 
                 next_token = events_before.last().map(|(count, _)| count).copied();
@@ -147,6 +168,7 @@ impl Service {
                     chunk: events_before,
                     next_batch: next_token.map(|t| t.stringify()),
                     prev_batch: Some(from.stringify()),
+                    recursion_depth: if recurse { Some(depth.into()) } else { None },
                 })
             }
         }
@@ -158,14 +180,44 @@ impl Service {
         room_id: &'a RoomId,
         target: &'a EventId,
         until: PduCount,
-    ) -> Result<impl Iterator<Item = Result<(PduCount, PduEvent)>> + 'a> {
+        max_depth: u8,
+    ) -> Result<Vec<(PduCount, PduEvent)>> {
         let room_id = services().rooms.short.get_or_create_shortroomid(room_id)?;
         let target = match services().rooms.timeline.get_pdu_count(target)? {
             Some(PduCount::Normal(c)) => c,
             // TODO: Support backfilled relations
             _ => 0, // This will result in an empty iterator
         };
-        self.db.relations_until(user_id, room_id, target, until)
+
+        self.db
+            .relations_until(user_id, room_id, target, until)
+            .map(|mut relations| {
+                let mut pdus: Vec<_> = (*relations).into_iter().filter_map(Result::ok).collect();
+                let mut stack: Vec<_> =
+                    pdus.clone().iter().map(|pdu| (pdu.to_owned(), 1)).collect();
+
+                while let Some(stack_pdu) = stack.pop() {
+                    let target = match stack_pdu.0 .0 {
+                        PduCount::Normal(c) => c,
+                        // TODO: Support backfilled relations
+                        PduCount::Backfilled(_) => 0, // This will result in an empty iterator
+                    };
+
+                    if let Ok(relations) = self.db.relations_until(user_id, room_id, target, until)
+                    {
+                        for relation in relations.flatten() {
+                            if stack_pdu.1 < max_depth {
+                                stack.push((relation.clone(), stack_pdu.1 + 1));
+                            }
+
+                            pdus.push(relation);
+                        }
+                    }
+                }
+
+                pdus.sort_by(|a, b| a.0.cmp(&b.0));
+                pdus
+            })
     }
 
     #[tracing::instrument(skip(self, room_id, event_ids))]

src/service/rooms/search/data.rs

@@ -4,6 +4,8 @@ use ruma::RoomId;
 pub trait Data: Send + Sync {
     fn index_pdu(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()>;
 
+    fn deindex_pdu(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()>;
+
     #[allow(clippy::type_complexity)]
     fn search_pdus<'a>(
         &'a self,

src/service/rooms/search/mod.rs

@@ -15,6 +15,16 @@ impl Service {
         self.db.index_pdu(shortroomid, pdu_id, message_body)
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn deindex_pdu<'a>(
+        &self,
+        shortroomid: u64,
+        pdu_id: &[u8],
+        message_body: &str,
+    ) -> Result<()> {
+        self.db.deindex_pdu(shortroomid, pdu_id, message_body)
+    }
+
     #[tracing::instrument(skip(self))]
     pub fn search_pdus<'a>(
         &'a self,

src/service/rooms/spaces/mod.rs

@@ -408,7 +408,7 @@ impl Service {
                     debug!("User is not allowed to see room {room_id}");
                     // This error will be caught later
                     return Err(Error::BadRequest(
-                        ErrorKind::Forbidden,
+                        ErrorKind::forbidden(),
                         "User is not allowed to see the room",
                     ));
                 }

src/service/rooms/state_accessor/mod.rs

@@ -321,6 +321,7 @@ impl Service {
             unsigned: None,
             state_key: Some(target_user.into()),
             redacts: None,
+            timestamp: None,
         };
 
         Ok(services()

src/service/rooms/timeline/mod.rs

@@ -21,10 +21,9 @@ use ruma::{
         GlobalAccountDataEventType, StateEventType, TimelineEventType,
     },
     push::{Action, Ruleset, Tweak},
-    serde::Base64,
     state_res::{self, Event, RoomVersion},
-    uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
-    OwnedServerName, RoomId, RoomVersionId, ServerName, UserId,
+    uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
+    OwnedEventId, OwnedRoomId, OwnedServerName, RoomId, RoomVersionId, ServerName, UserId,
 };
 use serde::Deserialize;
 use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
@@ -33,7 +32,10 @@ use tracing::{error, info, warn};
 
 use crate::{
     api::server_server,
-    service::pdu::{EventHash, PduBuilder},
+    service::{
+        globals::SigningKeys,
+        pdu::{EventHash, PduBuilder},
+    },
     services, utils, Error, PduEvent, Result,
 };
 
@@ -399,7 +401,7 @@ impl Service {
                                 &pdu.room_id,
                                 false,
                             )? {
-                                self.redact_pdu(redact_id, pdu)?;
+                                self.redact_pdu(redact_id, pdu, shortroomid)?;
                             }
                         }
                     }
@@ -416,7 +418,7 @@ impl Service {
                                 &pdu.room_id,
                                 false,
                             )? {
-                                self.redact_pdu(redact_id, pdu)?;
+                                self.redact_pdu(redact_id, pdu, shortroomid)?;
                             }
                         }
                     }
@@ -483,20 +485,27 @@ impl Service {
                         .search
                         .index_pdu(shortroomid, &pdu_id, &body)?;
 
-                    let server_user = format!("@conduit:{}", services().globals.server_name());
+                    let server_user = services().globals.server_user();
 
                     let to_conduit = body.starts_with(&format!("{server_user}: "))
                         || body.starts_with(&format!("{server_user} "))
                         || body == format!("{server_user}:")
-                        || body == server_user;
+                        || body == server_user.as_str();
 
                     // This will evaluate to false if the emergency password is set up so that
                     // the administrator can execute commands as conduit
-                    let from_conduit = pdu.sender == server_user
+                    let from_conduit = pdu.sender == *server_user
                         && services().globals.emergency_password().is_none();
 
                     if let Some(admin_room) = services().admin.get_admin_room()? {
-                        if to_conduit && !from_conduit && admin_room == pdu.room_id {
+                        if to_conduit
+                            && !from_conduit
+                            && admin_room == pdu.room_id
+                            && services()
+                                .rooms
+                                .state_cache
+                                .is_joined(server_user, &admin_room)?
+                        {
                             services().admin.process_message(body);
                         }
                     }
@@ -656,6 +665,7 @@ impl Service {
             unsigned,
             state_key,
             redacts,
+            timestamp,
         } = pdu_builder;
 
         let prev_events: Vec<_> = services()
@@ -725,9 +735,9 @@ impl Service {
             event_id: ruma::event_id!("$thiswillbefilledinlater").into(),
             room_id: room_id.to_owned(),
             sender: sender.to_owned(),
-            origin_server_ts: utils::millis_since_unix_epoch()
-                .try_into()
-                .expect("time is valid"),
+            origin_server_ts: timestamp
+                .map(|ts| ts.get())
+                .unwrap_or_else(|| MilliSecondsSinceUnixEpoch::now().get()),
             kind: event_type,
             content,
             state_key,
@@ -762,7 +772,7 @@ impl Service {
 
         if !auth_check {
             return Err(Error::BadRequest(
-                ErrorKind::Forbidden,
+                ErrorKind::forbidden(),
                 "Event is not authorized.",
             ));
         }
@@ -805,7 +815,7 @@ impl Service {
         pdu.event_id = EventId::parse_arc(format!(
             "${}",
             ruma::signatures::reference_hash(&pdu_json, &room_version_id)
-                .expect("ruma can calculate reference hashes")
+                .expect("Event format validated when event was hashed")
         ))
         .expect("ruma's reference hashes are valid event ids");
 
@@ -842,7 +852,7 @@ impl Service {
                     TimelineEventType::RoomEncryption => {
                         warn!("Encryption is not allowed in the admins room");
                         return Err(Error::BadRequest(
-                            ErrorKind::Forbidden,
+                            ErrorKind::forbidden(),
                             "Encryption is not allowed in the admins room.",
                         ));
                     }
@@ -857,7 +867,7 @@ impl Service {
                             .filter(|v| v.starts_with('@'))
                             .unwrap_or(sender.as_str());
                         let server_name = services().globals.server_name();
-                        let server_user = format!("@conduit:{}", server_name);
+                        let server_user = services().globals.server_user().as_str();
                         let content = serde_json::from_str::<ExtractMembership>(pdu.content.get())
                             .map_err(|_| Error::bad_database("Invalid content in pdu."))?;
 
@@ -865,7 +875,7 @@ impl Service {
                             if target == server_user {
                                 warn!("Conduit user cannot leave from admins room");
                                 return Err(Error::BadRequest(
-                                    ErrorKind::Forbidden,
+                                    ErrorKind::forbidden(),
                                     "Conduit user cannot leave from admins room.",
                                 ));
                             }
@@ -881,7 +891,7 @@ impl Service {
                             if count < 2 {
                                 warn!("Last admin cannot leave from admins room");
                                 return Err(Error::BadRequest(
-                                    ErrorKind::Forbidden,
+                                    ErrorKind::forbidden(),
                                     "Last admin cannot leave from admins room.",
                                 ));
                             }
@@ -891,7 +901,7 @@ impl Service {
                             if target == server_user {
                                 warn!("Conduit user cannot be banned in admins room");
                                 return Err(Error::BadRequest(
-                                    ErrorKind::Forbidden,
+                                    ErrorKind::forbidden(),
                                     "Conduit user cannot be banned in admins room.",
                                 ));
                             }
@@ -907,7 +917,7 @@ impl Service {
                             if count < 2 {
                                 warn!("Last admin cannot be banned in admins room");
                                 return Err(Error::BadRequest(
-                                    ErrorKind::Forbidden,
+                                    ErrorKind::forbidden(),
                                     "Last admin cannot be banned in admins room.",
                                 ));
                             }
@@ -939,7 +949,7 @@ impl Service {
                             false,
                         )? {
                             return Err(Error::BadRequest(
-                                ErrorKind::Forbidden,
+                                ErrorKind::forbidden(),
                                 "User cannot redact this event.",
                             ));
                         }
@@ -960,7 +970,7 @@ impl Service {
                             false,
                         )? {
                             return Err(Error::BadRequest(
-                                ErrorKind::Forbidden,
+                                ErrorKind::forbidden(),
                                 "User cannot redact this event.",
                             ));
                         }
@@ -1100,14 +1110,33 @@ impl Service {
 
     /// Replace a PDU with the redacted form.
     #[tracing::instrument(skip(self, reason))]
-    pub fn redact_pdu(&self, event_id: &EventId, reason: &PduEvent) -> Result<()> {
+    pub fn redact_pdu(
+        &self,
+        event_id: &EventId,
+        reason: &PduEvent,
+        shortroomid: u64,
+    ) -> Result<()> {
         // TODO: Don't reserialize, keep original json
         if let Some(pdu_id) = self.get_pdu_id(event_id)? {
             let mut pdu = self
                 .get_pdu_from_id(&pdu_id)?
                 .ok_or_else(|| Error::bad_database("PDU ID points to invalid PDU."))?;
+
+            #[derive(Deserialize)]
+            struct ExtractBody {
+                body: String,
+            }
+
+            if let Ok(content) = serde_json::from_str::<ExtractBody>(pdu.content.get()) {
+                services()
+                    .rooms
+                    .search
+                    .deindex_pdu(shortroomid, &pdu_id, &content.body)?;
+            }
+
             let room_version_id = services().rooms.state.get_room_version(&pdu.room_id)?;
             pdu.redact(room_version_id, reason)?;
+
             self.replace_pdu(
                 &pdu_id,
                 &utils::to_canonical_object(&pdu).expect("PDU is an object"),
@@ -1188,7 +1217,7 @@ impl Service {
         &self,
         origin: &ServerName,
         pdu: Box<RawJsonValue>,
-        pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
+        pub_key_map: &RwLock<BTreeMap<String, SigningKeys>>,
     ) -> Result<()> {
         let (event_id, value, room_id) = server_server::parse_incoming_pdu(&pdu)?;
 

src/service/uiaa/mod.rs

@@ -86,7 +86,7 @@ impl Service {
 
                     if !hash_matches {
                         uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {
-                            kind: ErrorKind::Forbidden,
+                            kind: ErrorKind::forbidden(),
                             message: "Invalid username or password.".to_owned(),
                         });
                         return Ok((false, uiaainfo));
@@ -101,7 +101,7 @@ impl Service {
                     uiaainfo.completed.push(AuthType::RegistrationToken);
                 } else {
                     uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {
-                        kind: ErrorKind::Forbidden,
+                        kind: ErrorKind::forbidden(),
                         message: "Invalid registration token.".to_owned(),
                     });
                     return Ok((false, uiaainfo));

src/service/users/data.rs

@@ -211,4 +211,10 @@ pub trait Data: Send + Sync {
     fn create_filter(&self, user_id: &UserId, filter: &FilterDefinition) -> Result<String>;
 
     fn get_filter(&self, user_id: &UserId, filter_id: &str) -> Result<Option<FilterDefinition>>;
+
+    // Creates an OpenID token, which can be used to prove that a user has access to an account (primarily for integrations)
+    fn create_openid_token(&self, user_id: &UserId) -> Result<(String, u64)>;
+
+    /// Find out which user an OpenID access token belongs to.
+    fn find_from_openid_token(&self, token: &str) -> Result<Option<OwnedUserId>>;
 }

src/service/users/mod.rs

@@ -9,7 +9,6 @@ pub use data::Data;
 use ruma::{
     api::client::{
         device::Device,
-        error::ErrorKind,
         filter::FilterDefinition,
         sync::sync_events::{
             self,
@@ -20,7 +19,7 @@ use ruma::{
     events::AnyToDeviceEvent,
     serde::Raw,
     DeviceId, DeviceKeyAlgorithm, DeviceKeyId, OwnedDeviceId, OwnedDeviceKeyId, OwnedMxcUri,
-    OwnedRoomId, OwnedUserId, RoomAliasId, UInt, UserId,
+    OwnedRoomId, OwnedUserId, UInt, UserId,
 };
 
 use crate::{services, Error, Result};
@@ -262,19 +261,14 @@ impl Service {
 
     /// Check if a user is an admin
     pub fn is_admin(&self, user_id: &UserId) -> Result<bool> {
-        let admin_room_alias_id =
-            RoomAliasId::parse(format!("#admins:{}", services().globals.server_name()))
-                .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Invalid alias."))?;
-        let admin_room_id = services()
-            .rooms
-            .alias
-            .resolve_local_alias(&admin_room_alias_id)?
-            .unwrap();
-
+        if let Some(admin_room_id) = services().admin.get_admin_room()? {
             services()
                 .rooms
                 .state_cache
                 .is_joined(user_id, &admin_room_id)
+        } else {
+            Ok(false)
+        }
     }
 
     /// Create a new user account on this homeserver.
@@ -598,6 +592,16 @@ impl Service {
     ) -> Result<Option<FilterDefinition>> {
         self.db.get_filter(user_id, filter_id)
     }
+
+    // Creates an OpenID token, which can be used to prove that a user has access to an account (primarily for integrations)
+    pub fn create_openid_token(&self, user_id: &UserId) -> Result<(String, u64)> {
+        self.db.create_openid_token(user_id)
+    }
+
+    /// Find out which user an OpenID access token belongs to.
+    pub fn find_from_openid_token(&self, token: &str) -> Result<Option<OwnedUserId>> {
+        self.db.find_from_openid_token(token)
+    }
 }
 
 /// Ensure that a user only sees signatures from themselves and the target user

src/utils/error.rs

@@ -128,7 +128,7 @@ impl Error {
                 kind.clone(),
                 match kind {
                     WrongRoomKeysVersion { .. }
-                    | Forbidden
+                    | Forbidden { .. }
                     | GuestAccessForbidden
                     | ThreepidAuthFailed
                     | ThreepidDenied => StatusCode::FORBIDDEN,