Compare commits

...

28 Commits

Author SHA1 Message Date
timedout 985b45d038 style: Remove redundant length check for dag build 2026-07-11 07:40:26 +01:00
timedout 4f7436a58f style: Accept event id reference in pdu_format_check_1 2026-07-11 05:09:04 +01:00
timedout 1328d45c2d fix: Remove redundant "correct room ID" filter 2026-07-11 05:04:53 +01:00
timedout 5328055c8e feat: Move PDU validation out of parsing and into check 1 2026-07-11 05:00:39 +01:00
timedout 367a747344 style: Remove needless async def for parse_incoming_pdu_with_known_room 2026-07-11 04:35:50 +01:00
timedout eb9a478f00 fix: Address review feedback 2026-07-11 04:30:14 +01:00
timedout 8688a66df7 fix: Invert boolean returned by policy server check 2026-07-11 02:31:32 +01:00
timedout c8fff77229 feat: Add more debug logging to PDU handle funcs
Tracking down weird soft-fails
2026-07-11 02:30:50 +01:00
timedout beed83855f perf: Change signature of build_local_dag to remove refmap hacks 2026-07-10 22:59:12 +01:00
timedout d65e560e87 style: Remove duplicate pdu check 1 call in handle_incoming_pdu 2026-07-10 20:59:38 +01:00
timedout 82fef3ce79 fix: Perform checks 1-3 on auth chain PDUs 2026-07-10 20:50:22 +01:00
timedout 5dca3be349 refactor: Move PDU checks 1, 2, and 3 into pdu_checks.rs 2026-07-10 20:50:22 +01:00
timedout b90ceee9f0 style: Unify and document state_before_incoming 2026-07-10 20:50:22 +01:00
timedout 11799fa315 fix: Don't reject auth events when they're innocent in auth chain response 2026-07-10 20:50:22 +01:00
timedout 0f90072caa style: Move PDU check functions into their own module 2026-07-10 20:50:22 +01:00
timedout 794d429367 refactor: Reduce complexity of incoming PDU handling
* Ignores any events pushed without a room ID
* Removes needless clones for PDU size checking
* Removes incorrect ACL check in incoming handler
* Fast-path handling already handled outlier events
* Remove redundant same-room checks, replace with useful ones
* Combine event rejection and persistence in a single function
* Additional safety assertions in upgrade task
* Split upgrade task into multiple subroutines for reduced cognitive complexity
* Only mutate current state and forward extremities in tandem
* Improve code documentation to better explain the logic flow
2026-07-10 20:50:22 +01:00
Henry-Hiles 03bde0e5fa feat: add commit info to Nix build 2026-07-10 19:04:01 +00:00
Jade Ellis 5e306f160a chore: Admin room announcement 2026-07-10 19:04:03 +01:00
Ginger fb2eedd5fd chore: Update changelog again 2026-07-10 13:09:49 -04:00
Ginger a76c5c95aa chore: Release 2026-07-10 13:01:48 -04:00
Ginger 7f83ec3883 chore: Update changelog 2026-07-10 12:59:50 -04:00
Ginger 9b14ab4c0b fix: Parse OIDC discovery URL more leniently 2026-07-10 16:52:04 +00:00
timedout b915692ae1 fix: Remove broken federated invite revoke handling
It turns out it never worked in the first place :D
2026-07-10 16:32:38 +01:00
theS1LV3R bed80964fd chore: Update changelog 2026-07-10 13:44:59 +00:00
theS1LV3R bc7e3185ef chore(deps): Update to resolvematrix 1.1.0 2026-07-10 13:44:59 +00:00
theS1LV3R 7082914d7d chore: Update resolvematrix dependency rev hash 2026-07-10 13:44:59 +00:00
theS1LV3R 8fb14f666f chore: Add changelog 2026-07-10 13:44:59 +00:00
theS1LV3R 8b6a176901 fix: Use correct DNS caching for SNI to function properly
This removes the old (broken) DNS caching method for reqwest, and replaces
it with resolvematrix' `MatrixDnsResolver`, as it correctly implements
caching and overrides for correct SNI behavior.
2026-07-10 13:44:59 +00:00
76 changed files with 1018 additions and 983 deletions
+66
View File
@@ -1,3 +1,69 @@
# Continuwuity 26.6.0 (2026-07-10)
## Features
- Added support for linking an external identity provider with OIDC. Contributed by @ginger. (#765)
- Updated [MSC4284: Policy Servers](https://github.com/matrix-org/matrix-spec-proposals/pull/4284) implementation to support the newly stabilised proposal. Contributed by @nex. (#1487)
- Added config option for default room ACLs. Contributed by @eve. (#1691)
- Added support for fallback encryption keys. (#1710)
- Add `!admin users reject-all-invites` to clean invite spam (#1741)
- Implemented event rejection, which should resolve and prevent future netsplits of the kinds observed
within some Continuwuity rooms.
Also resolved several bugs related to both soft-failing events, and event backfilling, which should
improve state resolution stability.
The `!admin debug get-pdu` command was updated to disambiguate event acceptance status, and
`!admin debug show-auth-chain` was added to visually display event auth chains, which may assist
developers in debugging strangely complex events.
Contributed by @nex. (#1747)
- Added full support for [MSC4168: Update `m.space.*` state on room upgrade](https://github.com/matrix-org/matrix-spec-proposals/pull/4168). Contributed by @nex. (#1807)
- Improved the performance and reliability of fetching missing events, improving network partition recovery. Contributed
by @nex. (#1818)
- Added static builds using Nix, allowing for Continuwuity on musl. During this, we also introduced a `max-perf-haswell` package, separating it from `max-perf`, so you may want to swap to this if you are on NixOS. Contributed by @Henry-Hiles (QuadRadical). (#1853)
- Added support for MSC4380 invite blocking, which has become part of the Matrix specification in v1.18. Contributed by @nex. (#1875)
- Added `!admin debug get-state-at` command (#1877)
- Added a configuration option to allow choosing a client IP source that is not the TCP connecting IP. Contributed by @nex. (#1931)
- Added support for MSC4466, which allows clients to customize how changes to a user's global profile are propagated. Contributed by @ginger.
- Added support for Matrix 1.16's `state_after` feature, allowing clients which understand it to sync room state changes more reliably. Contributed by @ginger.
- Added support for authenticating clients using the new OAuth 2.0 login API. Contributed by @ginger.
- Appservice device management as outlined in MSC4190 (part of Matrix 1.17) is now fully supported. Contributed by @ginger.
- Users may now be forbidden from deactivating their own accounts with the new `allow_deactivation` config option. Contributed by @ginger.
## Bugfixes
- Adjusted legacy sync logic to allow the `roomsynctoken_shortstatehash` database column to be dropped, massively reducing database sizes, especially for old deployments. Contributed by @ginger. (#917)
- Fixed a bug that caused the server to drop events during processing if several events for the same room were sent in a singular transaction. Contributed by @nex. (#1711)
- fix `!admin query account-data account-data-get` not returning the content (#1742)
- Fixed an issue where Continuwuity would only advertise support for the unstable endpoint for Mutual Rooms (MSC2666), despite only supporting the stable endpoint. Contributed by @Henry-Hiles (QuadRadical) (#1752)
- Fixed admin commands being ignored when they had leading whitespace before admin commands. Contributed by @kitvonsnookerz. (#1804)
- Fixed several bugs in the `POST /_matrix/client/v3/rooms/{roomId}/upgrade` endpoint. Contributed by @nex. (#1807)
- Devices which set their presence as "offline" will no longer be considered for presence updates. Contributed by @timedout.
- Improved invite and join reliability in clients using legacy sync. Contributed by @ginger
- The invite recipient's membership event is now included in invite stripped state, which should fix flaky invite display in some clients. Contributed by @ginger
## Improved Documentation
- Add performance tuning documentation. Contributed by @stratself. (#1498)
- Explain accessing Continuwuity's server console when deployed via Docker. (#1671)
- Clarified in the config that `max_request_size` affects federated media as well. (#1706)
- Added example configuration using caddy-docker-proxy in the livekit setup section of the docs. Contributed by @Cease (#1762)
- Updated deployment docs to account for new RPM package availability across more distros. Contributed by @julian45. (#1912)
## Deprecations and Removals
- Removed support for LDAP. (#1701)
- Removed support for guest user registration, a little-used and deprecated approach to room previews.
- Removed the `/_conduwuit/` versions of the `local_user_count` and `version` routes. These routes are still accessible under the `/_continuwuity` prefix.
- Support for server-side blurhashing (part of MSC2448) has been removed.
- The deprecated `well_known.rtc_focus_server_urls` config option has been removed. MatrixRTC foci should be configured using the `matrix_rtc.foci` config option.
## Misc
- #1505, #1829, #1927, #1933, #1934
- Switched from Continuwuity's fork of Ruma back to upstream Ruma. Contributed by @ginger.
- The version of Debian that the Docker-based build process uses has been upgraded from Bookworm to Trixie, meaning that standalone binaries now have a minimum glibc of 2.41, and can no longer be used on distro versions from before 2025-01-30
# Continuwuity 0.5.8 (2026-04-24)
## Features
Generated
+14 -14
View File
@@ -826,7 +826,7 @@ dependencies = [
[[package]]
name = "conduwuit"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"aws-lc-rs",
"clap",
@@ -864,7 +864,7 @@ dependencies = [
[[package]]
name = "conduwuit_admin"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"assign",
"clap",
@@ -890,7 +890,7 @@ dependencies = [
[[package]]
name = "conduwuit_api"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"assign",
"async-trait",
@@ -928,7 +928,7 @@ dependencies = [
[[package]]
name = "conduwuit_build_metadata"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"built",
"cargo_metadata",
@@ -936,7 +936,7 @@ dependencies = [
[[package]]
name = "conduwuit_core"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"argon2",
"arrayvec",
@@ -1004,7 +1004,7 @@ dependencies = [
[[package]]
name = "conduwuit_database"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"async-channel",
"conduwuit_core",
@@ -1025,7 +1025,7 @@ dependencies = [
[[package]]
name = "conduwuit_macros"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"cargo_toml",
"itertools 0.15.0",
@@ -1036,7 +1036,7 @@ dependencies = [
[[package]]
name = "conduwuit_router"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"assign",
"axum",
@@ -1074,7 +1074,7 @@ dependencies = [
[[package]]
name = "conduwuit_service"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"askama",
"assign",
@@ -1126,7 +1126,7 @@ dependencies = [
[[package]]
name = "conduwuit_web"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"askama",
"assign",
@@ -4539,9 +4539,9 @@ checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7"
[[package]]
name = "resolvematrix"
version = "1.0.0"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3121df844d2f59c801c703af7200b8dab5d1c04c5032cd57fa9347440346fe87"
checksum = "a574e37f06e219dcfdb14a4f0abf56e806c4551e030aae53802d2c42b29aea33"
dependencies = [
"futures",
"hickory-resolver",
@@ -4799,7 +4799,7 @@ dependencies = [
[[package]]
name = "ruminuwuity"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"assign",
"ruma",
@@ -6972,7 +6972,7 @@ dependencies = [
[[package]]
name = "xtask"
version = "26.6.0-rc.1"
version = "26.6.0"
dependencies = [
"askama",
"cargo_metadata",
+2 -2
View File
@@ -12,7 +12,7 @@ license = "Apache-2.0"
# See also `rust-toolchain.toml`
readme = "README.md"
repository = "https://forgejo.ellis.link/continuwuation/continuwuity"
version = "26.6.0-rc.1"
version = "26.6.0"
[workspace.metadata.crane]
name = "conduwuit"
@@ -570,7 +570,7 @@ features = ["std"]
version = "0.3.0"
[workspace.dependencies.resolvematrix]
version = "1.0.0"
version = "1.1.0"
[workspace.dependencies.serde_urlencoded]
version = "0.7.1"
-1
View File
@@ -1 +0,0 @@
Appservice device management as outlined in MSC4190 (part of Matrix 1.17) is now fully supported. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Improved invite and join reliability in clients using legacy sync. Contributed by @ginger
-1
View File
@@ -1 +0,0 @@
Removed the `/_conduwuit/` versions of the `local_user_count` and `version` routes. These routes are still accessible under the `/_continuwuity` prefix.
-1
View File
@@ -1 +0,0 @@
The invite recipient's membership event is now included in invite stripped state, which should fix flaky invite display in some clients. Contributed by @ginger
-1
View File
@@ -1 +0,0 @@
Switched from Continuwuity's fork of Ruma back to upstream Ruma. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Users may now be forbidden from deactivating their own accounts with the new `allow_deactivation` config option. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Added support for Matrix 1.16's `state_after` feature, allowing clients which understand it to sync room state changes more reliably. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Added support for authenticating clients using the new OAuth 2.0 login API. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Added support for MSC4466, which allows clients to customize how changes to a user's global profile are propagated. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Devices which set their presence as "offline" will no longer be considered for presence updates. Contributed by @timedout.
-1
View File
@@ -1 +0,0 @@
Removed support for guest user registration, a little-used and deprecated approach to room previews.
-1
View File
@@ -1 +0,0 @@
The deprecated `well_known.rtc_focus_server_urls` config option has been removed. MatrixRTC foci should be configured using the `matrix_rtc.foci` config option.
-1
View File
@@ -1 +0,0 @@
The version of Debian that the Docker-based build process uses has been upgraded from Bookworm to Trixie, meaning that standalone binaries now have a minimum glibc of 2.41, and can no longer be used on distro versions from before 2025-01-30
-1
View File
@@ -1 +0,0 @@
Support for server-side blurhashing (part of MSC2448) has been removed.
-1
View File
@@ -1 +0,0 @@
Updated [MSC4284: Policy Servers](https://github.com/matrix-org/matrix-spec-proposals/pull/4284) implementation to support the newly stabilised proposal. Contributed by @nex.
-1
View File
@@ -1 +0,0 @@
Add performance tuning documentation. Contributed by @stratself.
-1
View File
@@ -1 +0,0 @@
Rewrite the resolver service to use [resolvematrix](https://forgejo.ellis.link/continuwuation/resolvematrix) for server resolution. Rewrite by @s1lv3r, crate by @Jade
-1
View File
@@ -1 +0,0 @@
Explain accessing Continuwuity's server console when deployed via Docker.
-1
View File
@@ -1 +0,0 @@
Added config option for default room ACLs. Contributed by @eve.
-1
View File
@@ -1 +0,0 @@
Removed support for LDAP.
-1
View File
@@ -1 +0,0 @@
Clarified in the config that `max_request_size` affects federated media as well.
-1
View File
@@ -1 +0,0 @@
Added support for fallback encryption keys.
-1
View File
@@ -1 +0,0 @@
Fixed a bug that caused the server to drop events during processing if several events for the same room were sent in a singular transaction. Contributed by @nex.
-1
View File
@@ -1 +0,0 @@
Add `!admin users reject-all-invites` to clean invite spam
-1
View File
@@ -1 +0,0 @@
fix `!admin query account-data account-data-get` not returning the content
-9
View File
@@ -1,9 +0,0 @@
Implemented event rejection, which should resolve and prevent future netsplits of the kinds observed
within some Continuwuity rooms.
Also resolved several bugs related to both soft-failing events, and event backfilling, which should
improve state resolution stability.
The `!admin debug get-pdu` command was updated to disambiguate event acceptance status, and
`!admin debug show-auth-chain` was added to visually display event auth chains, which may assist
developers in debugging strangely complex events.
Contributed by @nex.
-1
View File
@@ -1 +0,0 @@
Fixed an issue where Continuwuity would only advertise support for the unstable endpoint for Mutual Rooms (MSC2666), despite only supporting the stable endpoint. Contributed by @Henry-Hiles (QuadRadical)
-1
View File
@@ -1 +0,0 @@
Added example configuration using caddy-docker-proxy in the livekit setup section of the docs. Contributed by @Cease
-1
View File
@@ -1 +0,0 @@
Fixed admin commands being ignored when they had leading whitespace before admin commands. Contributed by @kitvonsnookerz.
-1
View File
@@ -1 +0,0 @@
Fixed several bugs in the `POST /_matrix/client/v3/rooms/{roomId}/upgrade` endpoint. Contributed by @nex.
-1
View File
@@ -1 +0,0 @@
Added full support for [MSC4168: Update `m.space.*` state on room upgrade](https://github.com/matrix-org/matrix-spec-proposals/pull/4168). Contributed by @nex.
-2
View File
@@ -1,2 +0,0 @@
Improved the performance and reliability of fetching missing events, improving network partition recovery. Contributed
by @nex.
-1
View File
@@ -1 +0,0 @@
Remove support for MSC4373, as the MSC is now closed. Contributed by @vel.
-1
View File
@@ -1 +0,0 @@
Added static builds using Nix, allowing for Continuwuity on musl. During this, we also introduced a `max-perf-haswell` package, separating it from `max-perf`, so you may want to swap to this if you are on NixOS. Contributed by @Henry-Hiles (QuadRadical).
-1
View File
@@ -1 +0,0 @@
Added support for MSC4380 invite blocking, which has become part of the Matrix specification in v1.18. Contributed by @nex.
-1
View File
@@ -1 +0,0 @@
Added `!admin debug get-state-at` command
-1
View File
@@ -1 +0,0 @@
Updated deployment docs to account for new RPM package availability across more distros. Contributed by @julian45.
-1
View File
@@ -1 +0,0 @@
Move resolver service into client service, and remove deprecated and unused code. Contributed by @s1lv3r.
-1
View File
@@ -1 +0,0 @@
Added a configuration option to allow choosing a client IP source that is not the TCP connecting IP. Contributed by @nex.
-1
View File
@@ -1 +0,0 @@
Enable `direct_tls` and more logging by default
-1
View File
@@ -1 +0,0 @@
Added support for linking an external identity provider with OIDC. Contributed by @ginger.
-1
View File
@@ -1 +0,0 @@
Adjusted legacy sync logic to allow the `roomsynctoken_shortstatehash` database column to be dropped, massively reducing database sizes, especially for old deployments. Contributed by @ginger.
@@ -6,10 +6,10 @@
"message": "Welcome to Continuwuity! Important announcements about the project will appear here."
},
{
"id": 14,
"id": 15,
"mention_room": true,
"date": "2026-06-20",
"message": "[v0.5.10](https://forgejo.ellis.link/continuwuation/continuwuity/releases/tag/v0.5.10) has been released. It is a security release, so we suggest you update as soon as possible. Don't forget to also join [our announcements room](https://matrix.to/#/!jIdNjSM5X-V5JVx2h2kAhUZIIQ08GyzPL55NFZAH1vM/%24K1ISNKIqfNiZzsNVCaTt2E7ZtNeP6Dsy6sbz9l3rO0A?via=ellis.link&via=gingershaped.computer&via=matrix.org)."
"date": "2026-07-10",
"message": "[Continuwuity 26.6.0](https://forgejo.ellis.link/continuwuation/continuwuity/releases/tag/v26.6.0) is finally out! This is by far our largest release yet, and it wouldn't have been possible without our community. Thank you so much for all of your patience, support, and/or development time over the last three months <3"
}
]
}
+2
View File
@@ -43,6 +43,8 @@ let
env = {
CARGO_PROFILE = profile;
RUSTFLAGS = rustflags;
GIT_COMMIT_HASH = self.rev or self.dirtyRev;
GIT_COMMIT_HASH_SHORT = self.shortRev or self.dirtyShortRev;
}
// (lib.optionalAttrs (rocksdb != null) {
ROCKSDB_INCLUDE_DIR = "${rocksdb}/include";
+1 -1
View File
@@ -1017,7 +1017,7 @@ pub(super) async fn resolve_true_destination(
.http_client(self.services.client.default.clone())
.build()?
} else {
&self.services.client.resolver
&self.services.client.matrix_resolver
};
let actual = resolver.resolve_server(server_name.as_str()).await?;
+4 -4
View File
@@ -47,7 +47,7 @@ async fn cache(
writeln!(self, "| Server Name | Destination | SNI | Override | Step | Expires |").await?;
writeln!(self, "| ----------- | ----------- | --- | -------- | ---- | ------- |").await?;
let entries = self.services.client.resolver.get_all_cache_entries();
let entries = self.services.client.matrix_resolver.get_all_cache_entries();
for (host, entry) in entries {
if let Some(ors) = overrides
@@ -81,13 +81,13 @@ async fn cache(
async fn flush_cache(&self, name: Option<OwnedServerName>, all: bool) -> Result {
if all {
self.services.client.resolver.clear_cache();
self.services.client.dns.clear_cache();
self.services.client.matrix_resolver.clear_cache();
self.services.client.dns_resolver.clear_cache();
writeln!(self, "Resolver and DNS caches cleared!").await
} else if let Some(name) = name {
self.services
.client
.resolver
.matrix_resolver
.remove_cache_entry(name.as_str());
self.write_str(&format!("Cleared {name} from resolver caches!"))
.await
+1 -1
View File
@@ -221,7 +221,7 @@ pub(crate) async fn invite_helper(
let pdu_id = services
.rooms
.event_handler
.handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, true)
.handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, false)
.boxed()
.await?
.ok_or_else(|| {
+10 -18
View File
@@ -241,7 +241,7 @@ async fn handle(
pdus: impl Stream<Item = Pdu> + Send,
edus: impl Stream<Item = Edu> + Send,
) -> std::result::Result<ResolvedMap, TransactionError> {
// group pdus by room
// Group PDUs by room for parallel processing
let pdus = pdus
.collect()
.map(|mut pdus: Vec<_>| {
@@ -252,7 +252,7 @@ async fn handle(
})
.await;
// we can evaluate rooms concurrently
// Evaluate rooms in parallel
let results: ResolvedMap = pdus
.into_iter()
.try_stream()
@@ -266,7 +266,7 @@ async fn handle(
.boxed()
.await?;
// evaluate edus after pdus, at least for now.
// Evaluate EDUs after PDUs in case some of the PDUs then forbid some EDUs.
edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
.boxed()
.await;
@@ -296,20 +296,12 @@ async fn handle_room(
// Try to sort PDUs by their dependencies, but fall back to arbitrary order on
// failure (e.g., cycles). This is best-effort; proper ordering is the sender's
// responsibility.
let sorted_event_ids = if pdu_map.len() >= 2 {
let refmap = pdu_map
.iter()
.map(|(event_id, obj)| (event_id.clone(), obj))
.collect();
build_local_dag(&refmap, DagBuilderTree::PrevEvents)
.await
.unwrap_or_else(|e| {
debug_warn!("Failed to build local DAG for room {room_id}: {e}");
pdu_map.keys().cloned().collect()
})
} else {
pdu_map.keys().cloned().collect()
};
let sorted_event_ids = build_local_dag(&pdu_map, DagBuilderTree::PrevEvents)
.await
.unwrap_or_else(|e| {
debug_warn!("Failed to build local DAG for room {room_id}: {e}");
pdu_map.keys().cloned().collect()
});
let mut results = Vec::with_capacity(sorted_event_ids.len());
for event_id in sorted_event_ids {
let value = pdu_map
@@ -322,7 +314,7 @@ async fn handle_room(
let result = services
.rooms
.event_handler
.handle_incoming_pdu(origin, room_id, &event_id, value.clone(), true)
.handle_incoming_pdu(origin, room_id, &event_id, value, false)
.boxed()
.await
.map(|_| ());
+1 -1
View File
@@ -200,7 +200,7 @@ async fn create_join_event(
let pdu_id = services
.rooms
.event_handler
.handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), true)
.handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), false)
.boxed()
.await?
.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;
+1 -1
View File
@@ -166,7 +166,7 @@ pub(crate) async fn create_knock_event_v1_route(
let pdu_id = services
.rooms
.event_handler
.handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), true)
.handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), false)
.boxed()
.await?
.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;
+1 -1
View File
@@ -153,7 +153,7 @@ async fn create_leave_event(
let pdu_id = services
.rooms
.event_handler
.handle_incoming_pdu(origin, room_id, &event_id, value, true)
.handle_incoming_pdu(origin, room_id, &event_id, value, false)
.boxed()
.await?
.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;
+1 -1
View File
@@ -2497,7 +2497,7 @@ pub struct OidcConfig {
/// automatically fetch the identity provider's metadata from this URL.
/// Generally you should set this to the base domain your identity provider
/// runs on.
pub discovery_url: Url,
pub discovery_url: String,
/// The OAuth client ID for Continuwuity to use when communicating with the
/// identity provider.
+8
View File
@@ -149,6 +149,14 @@ macro_rules! err_log {
#[macro_export]
#[collapse_debuginfo(yes)]
macro_rules! err_lev {
(debug_info) => {
if $crate::debug::logging() {
$crate::tracing::Level::INFO
} else {
$crate::tracing::Level::DEBUG
}
};
(debug_warn) => {
if $crate::debug::logging() {
$crate::tracing::Level::WARN
-108
View File
@@ -1,108 +0,0 @@
use std::{net::SocketAddr, sync::Arc, time::Duration};
use conduwuit::{Result, Server, err};
use futures::FutureExt;
use hickory_resolver::{
TokioResolver, config::ConnectionConfig, lookup_ip::LookupIp,
net::runtime::TokioRuntimeProvider,
};
use reqwest::dns::{Addrs, Name, Resolve, Resolving};
pub struct Resolver {
pub(crate) resolver: Arc<TokioResolver>,
server: Arc<Server>,
}
type ResolvingResult = Result<Addrs, Box<dyn std::error::Error + Send + Sync>>;
impl Resolver {
#[allow(clippy::as_conversions, clippy::cast_sign_loss, clippy::cast_possible_truncation)]
pub(super) fn build(server: &Arc<Server>) -> Result<Arc<Self>> {
let config = &server.config;
let (sys_conf, mut opts) = hickory_resolver::system_conf::read_system_conf()
.map_err(|e| err!(error!("Failed to configure DNS resolver from system: {e}")))?;
let mut conf = hickory_resolver::config::ResolverConfig::default();
if let Some(domain) = sys_conf.domain() {
conf.set_domain(domain.clone());
}
for sys_conf in sys_conf.search() {
conf.add_search(sys_conf.clone());
}
for sys_conf in sys_conf.name_servers() {
let mut ns = sys_conf.clone();
if config.query_over_tcp_only {
ns.connections = vec![ConnectionConfig::tcp()];
}
ns.trust_negative_responses = !config.query_all_nameservers;
conf.add_name_server(ns);
}
opts.cache_size = u64::from(config.dns_cache_entries);
opts.preserve_intermediates = true;
opts.negative_min_ttl = Some(Duration::from_secs(config.dns_min_ttl_nxdomain));
opts.negative_max_ttl = Some(Duration::from_hours(720));
opts.positive_min_ttl = Some(Duration::from_secs(config.dns_min_ttl));
opts.positive_max_ttl = Some(Duration::from_hours(168));
opts.timeout = Duration::from_secs(config.dns_timeout);
opts.attempts = config.dns_attempts as usize;
opts.try_tcp_on_error = config.dns_tcp_fallback;
opts.num_concurrent_reqs = 1;
opts.edns0 = true;
opts.case_randomization = true;
opts.ip_strategy = match config.ip_lookup_strategy {
| 1 => hickory_resolver::config::LookupIpStrategy::Ipv4Only,
| 2 => hickory_resolver::config::LookupIpStrategy::Ipv6Only,
| 3 => hickory_resolver::config::LookupIpStrategy::Ipv4AndIpv6,
| 4 => hickory_resolver::config::LookupIpStrategy::Ipv6thenIpv4,
| _ => hickory_resolver::config::LookupIpStrategy::Ipv4thenIpv6,
};
let runtime_provider = TokioRuntimeProvider::new();
let mut builder = TokioResolver::builder_with_config(conf, runtime_provider);
*builder.options_mut() = opts;
let resolver = Arc::new(builder.build().expect("failed to build resolver :("));
Ok(Arc::new(Self { resolver, server: server.clone() }))
}
/// Clear the in-memory hickory-dns caches
#[inline]
pub fn clear_cache(&self) { self.resolver.clear_cache(); }
}
impl Resolve for Resolver {
fn resolve(&self, name: Name) -> Resolving {
resolve_to_reqwest(self.server.clone(), self.resolver.clone(), name).boxed()
}
}
async fn resolve_to_reqwest(
server: Arc<Server>,
resolver: Arc<TokioResolver>,
name: Name,
) -> ResolvingResult {
use std::{io, io::ErrorKind::Interrupted};
let handle_shutdown = || Box::new(io::Error::new(Interrupted, "Server shutting down"));
let handle_results = |results: LookupIp| {
Box::new(
results
.iter()
.collect::<Vec<_>>()
.into_iter()
.map(|ip| SocketAddr::new(ip, 0)),
)
};
tokio::select! {
results = resolver.lookup_ip(name.as_str()) => Ok(handle_results(results?)),
() = server.until_shutdown() => Err(handle_shutdown()),
}
}
+84 -23
View File
@@ -1,19 +1,20 @@
mod dns;
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
use conduwuit::{Config, Result, err, trace};
use conduwuit::{Config, Result, Server, err, trace};
use either::Either;
use hickory_resolver::{
TokioResolver, config::ConnectionConfig, net::runtime::TokioRuntimeProvider,
};
use ipaddress::IPAddress;
use reqwest::redirect;
use resolvematrix::server::{MatrixResolver, MatrixResolverBuilder};
use crate::{client::dns::Resolver, service};
use crate::service;
pub struct Service {
pub resolver: MatrixResolver,
pub dns: Arc<Resolver>,
pub matrix_resolver: Arc<MatrixResolver>,
pub dns_resolver: Arc<TokioResolver>,
pub default: reqwest::Client,
pub url_preview: reqwest::Client,
@@ -31,8 +32,10 @@ pub struct Service {
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
let config = &args.server.config;
let dns = Resolver::build(args.server)?;
let resolver = MatrixResolverBuilder::new()
let dns_resolver = get_dns_resolver(args.server)?;
let matrix_resolver = Arc::new(MatrixResolverBuilder::new()
.dangerous_tls_accept_invalid_certs(args.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
.http_client(
base(&args.server.config)?
@@ -43,8 +46,9 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.redirect(redirect::Policy::limited(4))
.build()?
)
.dns_resolver(dns.resolver.clone())
.build()?;
.dns_resolver(dns_resolver.clone())
.build()?);
let matrix_dns_resolver = matrix_resolver.create_dns_resolver();
let url_preview_bind_addr = config
.url_preview_bound_interface
@@ -62,29 +66,31 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.unwrap_or_else(|| conduwuit::user_agent_media().to_owned());
Ok(Arc::new(Self {
resolver,
dns: dns.clone(),
matrix_resolver,
dns_resolver,
default: base(config)?.dns_resolver(dns.clone()).build()?,
default: base(config)?
.dns_resolver(matrix_dns_resolver.clone())
.build()?,
url_preview: base(config)
.and_then(|builder| {
builder_interface(builder, url_preview_bind_iface.as_deref())
})?
.local_address(url_preview_bind_addr)
.dns_resolver(dns.clone())
.dns_resolver(matrix_dns_resolver.clone())
.timeout(Duration::from_secs(config.url_preview_timeout))
.redirect(redirect::Policy::limited(3))
.user_agent(url_preview_user_agent)
.build()?,
extern_media: base(config)?
.dns_resolver(dns.clone())
.dns_resolver(matrix_dns_resolver.clone())
.redirect(redirect::Policy::limited(3))
.build()?,
federation: base(config)?
.dns_resolver(dns.clone())
.dns_resolver(matrix_dns_resolver.clone())
.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
.read_timeout(Duration::from_secs(config.federation_timeout))
.timeout(Duration::from_secs(
@@ -98,7 +104,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
federation_slow: base(config)?
.dns_resolver(dns.clone())
.dns_resolver(matrix_dns_resolver.clone())
.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
.read_timeout(Duration::from_secs(config.federation_timeout.saturating_mul(6)))
.timeout(Duration::from_secs(
@@ -112,7 +118,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
sender: base(config)?
.dns_resolver(dns.clone())
.dns_resolver(matrix_dns_resolver.clone())
.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
.read_timeout(Duration::from_secs(config.sender_timeout))
.timeout(Duration::from_secs(config.sender_timeout))
@@ -122,7 +128,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
appservice: base(config)?
.dns_resolver(dns.clone())
.dns_resolver(matrix_dns_resolver.clone())
.connect_timeout(Duration::from_secs(5))
.read_timeout(Duration::from_secs(config.appservice_timeout))
.timeout(Duration::from_secs(config.appservice_timeout))
@@ -132,7 +138,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
pusher: base(config)?
.dns_resolver(dns)
.dns_resolver(matrix_dns_resolver)
.connect_timeout(Duration::from_secs(config.pusher_conn_timeout))
.timeout(Duration::from_secs(config.pusher_timeout))
.pool_max_idle_per_host(1)
@@ -151,11 +157,11 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
}
async fn clear_cache(&self) {
self.resolver.clear_cache();
self.dns.resolver.clear_cache();
self.matrix_resolver.clear_cache();
self.dns_resolver.clear_cache();
}
fn name(&self) -> &str { service::make_name(std::module_path!()) }
fn name(&self) -> &str { service::make_name(module_path!()) }
}
impl Service {
@@ -254,3 +260,58 @@ fn builder_interface(
Ok(builder)
}
}
fn get_dns_resolver(server: &Arc<Server>) -> Result<Arc<TokioResolver>> {
let config = &server.config;
let (sys_conf, mut opts) = hickory_resolver::system_conf::read_system_conf()
.map_err(|e| err!(error!("Failed to configure DNS resolver from system: {e}")))?;
let mut conf = hickory_resolver::config::ResolverConfig::default();
if let Some(domain) = sys_conf.domain() {
conf.set_domain(domain.clone());
}
for sys_conf in sys_conf.search() {
conf.add_search(sys_conf.clone());
}
for sys_conf in sys_conf.name_servers() {
let mut ns = sys_conf.clone();
if config.query_over_tcp_only {
ns.connections = vec![ConnectionConfig::tcp()];
}
ns.trust_negative_responses = !config.query_all_nameservers;
conf.add_name_server(ns);
}
opts.cache_size = u64::from(config.dns_cache_entries);
opts.preserve_intermediates = true;
opts.negative_min_ttl = Some(Duration::from_secs(config.dns_min_ttl_nxdomain));
opts.negative_max_ttl = Some(Duration::from_hours(720));
opts.positive_min_ttl = Some(Duration::from_secs(config.dns_min_ttl));
opts.positive_max_ttl = Some(Duration::from_hours(168));
opts.timeout = Duration::from_secs(config.dns_timeout);
opts.attempts = config.dns_attempts.into();
opts.try_tcp_on_error = config.dns_tcp_fallback;
opts.num_concurrent_reqs = 1;
opts.edns0 = true;
opts.case_randomization = true;
opts.ip_strategy = match config.ip_lookup_strategy {
| 1 => hickory_resolver::config::LookupIpStrategy::Ipv4Only,
| 2 => hickory_resolver::config::LookupIpStrategy::Ipv6Only,
| 3 => hickory_resolver::config::LookupIpStrategy::Ipv4AndIpv6,
| 4 => hickory_resolver::config::LookupIpStrategy::Ipv6thenIpv4,
| _ => hickory_resolver::config::LookupIpStrategy::Ipv4thenIpv6,
};
let runtime_provider = TokioRuntimeProvider::new();
let mut builder = TokioResolver::builder_with_config(conf, runtime_provider);
*builder.options_mut() = opts;
let resolver = Arc::new(builder.build().expect("failed to build resolver :("));
Ok(resolver)
}
+1 -1
View File
@@ -134,7 +134,7 @@ pub async fn execute_on<'i, T, PathBuilderInput>(
let actual = self
.services
.client
.resolver
.matrix_resolver
.resolve_server(dest.as_str())
.await?;
+2 -1
View File
@@ -172,7 +172,8 @@ async fn worker(self: Arc<Self>) -> Result {
.expect("redirect url should be valid");
let provider_metadata = CoreProviderMetadata::discover_async(
IssuerUrl::from_url(config.discovery_url.clone()),
IssuerUrl::new(config.discovery_url.clone())
.map_err(|err| err!("Failed to parse OIDC discovery URL: {err}"))?,
client,
)
.await
@@ -14,6 +14,7 @@ pub async fn acl_check(&self, server_name: &ServerName, room_id: &RoomId) -> Res
.await
.map(|c: RoomServerAclEventContent| c)
else {
trace!("Room has no ACL, allowing");
return Ok(());
};
@@ -1,5 +1,6 @@
use std::{
collections::{HashMap, HashSet, VecDeque},
fmt::{Display, Formatter},
time::Instant,
};
@@ -33,6 +34,20 @@ pub enum DagBuilderTree {
AuthEvents,
}
impl DagBuilderTree {
#[must_use]
pub fn as_str(&self) -> &str {
match self {
| Self::AuthEvents => "auth_events",
| Self::PrevEvents => "prev_events",
}
}
}
impl Display for DagBuilderTree {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(self.as_str()) }
}
/// Attempts to build a localised directed acyclic graph out of the given PDUs,
/// returning them in a topologically sorted order.
///
@@ -42,23 +57,22 @@ pub enum DagBuilderTree {
/// not account for power levels or other tie breaks.
#[allow(clippy::implicit_hasher)]
pub async fn build_local_dag(
pdu_map: &HashMap<OwnedEventId, &CanonicalJsonObject>,
pdu_map: &HashMap<OwnedEventId, CanonicalJsonObject>,
tree: DagBuilderTree,
) -> Result<Vec<OwnedEventId>> {
debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs");
if pdu_map.len() <= 1 {
return Ok(pdu_map.keys().cloned().collect());
}
let mut dag: HashMap<OwnedEventId, HashSet<OwnedEventId>> =
HashMap::with_capacity(pdu_map.len());
let mut id_origin_ts: HashMap<OwnedEventId, _> = HashMap::with_capacity(pdu_map.len());
let tree = match tree {
| DagBuilderTree::AuthEvents => "auth_events",
| DagBuilderTree::PrevEvents => "prev_events",
};
for (event_id, value) in pdu_map {
// Parse all prev events as event IDs - if they are missing, return an error (we
// can't sanely continue in this case), otherwise skip invalid prev events.
let prev_events = value
.get(tree)
.get(tree.as_str())
.and_then(CanonicalJsonValue::as_array)
.ok_or_else(|| err!(Request(BadJson("event JSON for {event_id} is missing {tree}"))))?
.iter()
@@ -370,9 +384,8 @@ async fn fetch_event_via(
.send_federation_request(&remote, get_event::v1::Request::new(event_id.clone()))
.await?;
let (calculated_event_id, value) = self
.parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)
.await?;
let (calculated_event_id, value) =
Self::parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)?;
if calculated_event_id != event_id {
Err!(Request(BadJson(warn!(
@@ -538,11 +551,7 @@ pub(super) async fn fetch_and_handle_auth_events<Pdu>(
}
}
let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> = discovered_events
.iter()
.map(|(id, data)| (id.clone(), data))
.collect();
let seeded_ordered = build_local_dag(&refmap, DagBuilderTree::AuthEvents)
let seeded_ordered = build_local_dag(&discovered_events, DagBuilderTree::AuthEvents)
.await
.expect("failed to build local DAG");
let mut pdus = HashMap::with_capacity(seeded_ordered.len());
+49 -20
View File
@@ -1,9 +1,8 @@
use std::collections::{HashMap, hash_map};
use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err};
use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err, warn};
use ruma::{
CanonicalJsonObject, OwnedEventId, ServerName,
api::federation::authorization::get_event_authorization,
OwnedEventId, ServerName, api::federation::authorization::get_event_authorization,
room_version_rules::RoomVersionRules,
};
use tokio::join;
@@ -50,16 +49,49 @@ pub(super) async fn fetch_and_persist_event_auth<Pdu>(
for auth_pdu_json in event_auth.auth_chain {
let (auth_event_room_id, auth_event_id, auth_pdu_json) =
self.parse_incoming_pdu(&auth_pdu_json).await?;
match self.parse_incoming_pdu(&auth_pdu_json).await {
| Ok(parsed) => parsed,
| Err(e) => {
warn!(error=?e, "Dropping auth chain event as it could not be parsed");
continue;
},
};
if let Err(e) = Self::pdu_format_check_1(
&auth_pdu_json,
room_version_rules,
create_event.event_id(),
) {
// drop this PDU
warn!(%auth_event_id, error=?e, "Dropping auth chain event as it violates the room event format");
continue;
}
let auth_pdu_json = match self
.signature_hash_check_2_3(auth_pdu_json, room_version_rules)
.await
{
| Ok(pdu_json) => pdu_json,
| Err(e) => {
// drop this PDU
warn!(
%auth_event_id,
error=?e,
"Dropping auth chain event as it has an invalid signature"
);
continue;
},
};
// PDU check 4 is done when we've finished aggregating
if auth_event_room_id != incoming_event.room_id_or_hash() {
return Err!(Request(Forbidden(
"Auth event {auth_event_id} is in {auth_event_room_id}, not {}.",
"Auth chain event {auth_event_id} is in {auth_event_room_id}, not {}.",
incoming_event.room_id_or_hash()
)));
}
let auth_pdu = PduEvent::from_id_val(&auth_event_id, auth_pdu_json).map_err(|e| {
err!(Request(BadJson("Invalid PDU {auth_event_id} in auth chain: {e}")))
})?;
if auth_pdu.state_key().is_none() {
return Err!(Request(BadJson(
"Invalid PDU {auth_event_id} in auth_chain: not a state event"
@@ -97,17 +129,11 @@ async fn authorise_remote_auth_chain<Pdu>(
where
Pdu: Event + Send + Sync,
{
// TODO(nex): build_local_dag's signature needs changing because all callsites
// seem to do this refmap hack.
let pdu_objects = auth_chain_map
.iter()
.map(|(event_id, pdu)| (event_id, pdu.to_canonical_object()))
.map(|(event_id, pdu)| (event_id.clone(), pdu.to_canonical_object()))
.collect::<HashMap<_, _>>();
let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> = pdu_objects
.iter()
.map(|(id, data)| ((*id).to_owned(), data))
.collect();
let auth_chain_topo = build_local_dag(&refmap, DagBuilderTree::AuthEvents)
let auth_chain_topo = build_local_dag(&pdu_objects, DagBuilderTree::AuthEvents)
.await?
.into_iter()
.map(|event_id| (event_id.clone(), auth_chain_map.get(&event_id).unwrap().to_owned()))
@@ -127,6 +153,12 @@ async fn authorise_remote_auth_chain<Pdu>(
continue;
}
// IMPORTANT: We can't use the handy dandy `handle_outlier_pdu` function here
// because it may then try to fetch missing auth events, resulting in deep
// recursion. We will do the minimum required steps to validate the PDU here.
// Checks 1-3 were already done before this function is called, so we only need
// to do check 4.
let mut auth_events_by_key: HashMap<_, _> =
HashMap::with_capacity(pdu.auth_events.len());
@@ -150,24 +182,21 @@ async fn authorise_remote_auth_chain<Pdu>(
},
| hash_map::Entry::Occupied(_) => {
// Duplicate auth events by key are not allowed.
self.services
.pdu_metadata
.mark_event_rejected(auth_event_id);
self.services.pdu_metadata.mark_event_rejected(&event_id);
self.reject_and_persist(&event_id, &pdu.to_canonical_object());
continue 'outer;
},
}
}
if !self
.is_event_self_authorised(
.auth_state_check_4(
&pdu,
room_version_rules,
create_event.as_pdu(),
&auth_events_by_key,
)
.await
.await?
{
self.services.pdu_metadata.mark_event_rejected(&event_id);
self.reject_and_persist(&event_id, &pdu.to_canonical_object());
}
}
@@ -5,13 +5,13 @@
utils::{BoolExt, IterStream, stream::BroadbandExt},
};
use futures::StreamExt;
use ruma::{CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName};
use ruma::{MilliSecondsSinceUnixEpoch, RoomId, ServerName};
use crate::rooms::event_handler::{build_local_dag, fetch_and_handle_outliers::DagBuilderTree};
impl super::Service {
/// Fetches any missing prev_events for this event and persists them before
/// returning.
/// returning. The caller is responsible for then handling the incoming PDU.
pub(super) async fn fetch_prevs(
&self,
room_id: &RoomId,
@@ -84,13 +84,7 @@ pub(super) async fn fetch_prevs(
})
.collect::<HashMap<_, _>>();
let to_persist = if mapped.len() <= 1 {
mapped.keys().map(ToOwned::to_owned).collect()
} else {
let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> =
mapped.iter().map(|(id, data)| (id.clone(), data)).collect();
build_local_dag(&refmap, DagBuilderTree::PrevEvents).await?
};
let to_persist = build_local_dag(&mapped, DagBuilderTree::PrevEvents).await?;
let job_start = Instant::now();
trace!("Starting to persist {} prev events", to_persist.len());
@@ -4,111 +4,37 @@
};
use conduwuit::{
Err, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, defer, err, error,
info, matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
};
use futures::{
FutureExt, StreamExt,
future::{OptionFuture, try_join4},
};
use ruma::{
CanonicalJsonValue, EventId, OwnedUserId, RoomId, ServerName, UserId,
events::{
TimelineEventType,
room::member::{MembershipState, RoomMemberEventContent},
},
Err, Event, Result, debug, debug_error, debug_warn, defer, matrix::PartialPdu, trace,
utils::time::jitter,
};
use futures::{FutureExt, StreamExt, future::try_join3};
use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UserId};
use tokio::sync::mpsc;
use crate::rooms::timeline::{RawPduId, pdu_fits};
async fn should_rescind_invite(
services: &crate::rooms::event_handler::Services,
content: &mut BTreeMap<String, CanonicalJsonValue>,
sender: &UserId,
room_id: &RoomId,
) -> Result<Option<PduEvent>> {
// We insert a bogus event ID since we can't actually calculate the right one
content.insert("event_id".to_owned(), CanonicalJsonValue::String("$rescind".to_owned()));
let pdu_event = serde_json::from_value::<PduEvent>(
serde_json::to_value(&content).expect("CanonicalJsonObj is a valid JsonValue"),
)
.map_err(|e| err!("invalid PDU: {e}"))?;
if pdu_event.room_id().is_none_or(|r| r != room_id)
&& pdu_event.sender() != sender
&& pdu_event.event_type() != &TimelineEventType::RoomMember
&& pdu_event.state_key().is_none_or(|v| v == sender.as_str())
{
return Ok(None);
}
let target_user_id = UserId::parse(pdu_event.state_key().unwrap())?;
if pdu_event
.get_content::<RoomMemberEventContent>()?
.membership
!= MembershipState::Leave
{
return Ok(None); // Not a leave event
}
// Does the target user have a pending invite?
let Ok(pending_invite_state) = services
.state_cache
.invite_state(&target_user_id, room_id)
.await
else {
return Ok(None); // No pending invite, so nothing to rescind
};
for event in pending_invite_state {
if event
.get_field::<String>("type")?
.is_some_and(|t| t == "m.room.member")
|| event
.get_field::<OwnedUserId>("state_key")?
.is_some_and(|s| s == *target_user_id)
|| event
.get_field::<OwnedUserId>("sender")?
.is_some_and(|s| s == *sender)
|| event
.get_field::<RoomMemberEventContent>("content")?
.is_some_and(|c| c.membership == MembershipState::Invite)
{
return Ok(Some(pdu_event));
}
}
Ok(None)
}
use crate::rooms::timeline::RawPduId;
impl super::Service {
/// When receiving an event one needs to:
/// 0. Check the server is in the room
/// 1. Skip the PDU if we already know about it
/// 1.1. Remove unsigned field
/// 2. Check signatures, otherwise drop
/// 3. Check content hash, redact if doesn't match
/// 4. Fetch any missing auth events doing all checks listed here starting
/// at 1. These are not timeline events
/// 5. Reject "due to auth events" if can't get all the auth events or some
/// of the auth events are also rejected "due to auth events"
/// 6. Reject "due to auth events" if the event doesn't pass auth based on
/// the auth events
/// 7. Persist this event as an outlier
/// 8. If not timeline event: stop
/// 9. Fetch any missing prev events doing all checks listed here starting
/// at 1. These are timeline events
/// 10. Fetch missing state and auth chain events by calling `/state_ids` at
/// backwards extremities doing all the checks in this list starting at
/// 1. These are not timeline events
/// 11. Check the auth of the event passes based on the state of the event
/// 12. Ensure that the state is derived from the previous current state
/// (i.e. we calculated by doing state res where one of the inputs was a
/// previously trusted set of state, don't just trust a set of state we
/// got from a remote)
/// 13. Use state resolution to find new room state
/// 14. Check if the event passes auth based on the "current state" of the
/// room, if not soft fail it
/// Handles an incoming PDU from federation.
///
/// First checks that we want to receive this PDU. If we already have it as
/// a timeline PDU, or we don't want to receive the PDU (e.g. origin ACL'd,
/// room disabled/unknown), abort.
///
/// The PDU is then handled as an outlier event, which performs [PDU checks]
/// 1 through 4. See: `handle_outlier_pdu`.
///
/// Once handled as an outlier, any missing prev events are fetched, and
/// then the PDU will be promoted/upgraded from an outlier to a timeline
/// event clients can see. See: `upgrade_outlier_to_timeline_pdu`. After
/// this finishes, the PDU is either accepted or left as an outlier.
///
/// If the PDU is successfully upgraded, the remaining extremity count of
/// the room is checked. If there are a potentially problematic number of
/// forward extremities, a squasher task is started with a debounce period,
/// which will eventually send a dummy event that ties up as many DAG forks
/// as possible.
///
/// [PDU checks]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
#[tracing::instrument(
name = "pdu",
skip_all,
@@ -120,51 +46,33 @@ pub async fn handle_incoming_pdu<'a>(
room_id: &'a RoomId,
event_id: &'a EventId,
value: BTreeMap<String, CanonicalJsonValue>,
is_timeline_event: bool,
is_backfilled_event: bool,
) -> Result<Option<RawPduId>> {
// 1. Skip the PDU if we already have it as a timeline event
// Skip the PDU if we already have it as a timeline event. We still re-process
// outliers in this scenario.
if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await {
debug!("Database hit for incoming PDU, skipping processing");
return Ok(Some(pdu_id));
}
if !pdu_fits(&mut value.clone()) {
warn!(
"dropping incoming PDU {event_id} in room {room_id} from {origin} because it \
exceeds 65535 bytes or is otherwise too large."
);
return Err!(Request(TooLarge("PDU is too large")));
}
trace!(
"processing incoming PDU from {origin} for room {room_id} with event id {event_id}"
);
// 1.1 Check we even know about the room
let meta_exists = self.services.metadata.exists(room_id).map(Ok);
// If this is a membership state event for a local user, we will be interested
// in this event even if we aren't in the room. This allows us to process things
// like revoking invites over federation, without being in the room.
let is_interesting_member_event = value.get("type").and_then(|t| t.as_str())
== Some("m.room.member")
&& value
.get("state_key")
.and_then(|s| s.as_str())
.and_then(|s| UserId::parse(s).ok())
.is_some_and(|u| self.services.globals.user_is_local(&u));
// 1.2 Check if the room is disabled
let is_disabled = self.services.metadata.is_disabled(room_id).map(Ok);
// 1.3.1 Check room ACL on origin field/server
let origin_acl_check = self.acl_check(origin, room_id);
// 1.3.2 Check room ACL on sender's server name
let sender: OwnedUserId = value
.get("sender")
.and_then(|v| v.as_str())
.ok_or_else(|| err!("No sender in object"))
.and_then(|v| Ok(UserId::parse(v)?))
.map_err(|e| err!(Request(BadJson("PDU does not have a valid sender key: {e}"))))?;
let sender_acl_check: OptionFuture<_> = sender
.server_name()
.ne(origin)
.then(|| self.acl_check(sender.server_name(), room_id))
.into();
let (meta_exists, is_disabled, (), ()) = try_join4(
meta_exists,
is_disabled,
origin_acl_check,
sender_acl_check.map(|o| o.unwrap_or(Ok(()))),
let (room_exists, is_disabled, ()) = try_join3(
self.services.metadata.exists(room_id).map(Ok),
self.services.metadata.is_disabled(room_id).map(Ok),
self.acl_check(origin, room_id),
)
.await
.inspect_err(
@@ -177,43 +85,11 @@ pub async fn handle_incoming_pdu<'a>(
)));
}
if !self
.services
.state_cache
.server_in_room(self.services.globals.server_name(), room_id)
.await
{
// Is this a federated invite rescind?
// copied from https://github.com/element-hq/synapse/blob/7e4588a/synapse/handlers/federation_event.py#L255-L300
if value.get("type").and_then(|t| t.as_str()) == Some("m.room.member") {
if let Some(pdu) =
should_rescind_invite(&self.services, &mut value.clone(), &sender, room_id)
.await?
{
debug_info!(
"Invite to {room_id} appears to have been rescinded by {sender}, \
marking as left"
);
self.services
.state_cache
.mark_as_left(&sender, room_id, Some(pdu))
.await;
self.services.sync.wake(&sender).await;
return Ok(None);
}
if !room_exists {
if is_interesting_member_event {
// TODO: handle interesting membership events where we aren't in
// the room
}
info!(
%origin,
%room_id,
"Dropping inbound PDU for room we aren't participating in"
);
return Err!(Request(NotFound("This server is not participating in that room.")));
}
if !meta_exists {
return Err!(Request(NotFound("Room is unknown to this server")));
}
@@ -239,8 +115,10 @@ pub async fn handle_incoming_pdu<'a>(
.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
.await?;
// 8. if not timeline event: stop
if !is_timeline_event {
// If this event is being processed as part of backfill, we don't want to end up
// *appending* it during the upgrade process, so we return early.
if is_backfilled_event {
debug!("Not promoting incoming event as it is being backfilled");
return Ok(None);
}
@@ -253,11 +131,16 @@ pub async fn handle_incoming_pdu<'a>(
.await?
.origin_server_ts();
if incoming_pdu.origin_server_ts() < first_ts_in_room {
debug_warn!(
"Not promoting incoming event as it is sent before we joined the room (but was \
not backfilled)"
);
return Ok(None);
}
// 9. Fetch any missing prev events doing all checks listed here starting at 1.
// These are timeline events
// Fetch any missing prev events doing all checks listed here starting at 1.
// These are timeline events.
// TODO: This part needs to be done in a background queue somewhere.
debug!("Fetching and persisting any missing prev events");
Box::pin(self.fetch_prevs(
@@ -268,13 +151,14 @@ pub async fn handle_incoming_pdu<'a>(
first_ts_in_room,
))
.await
.debug_inspect_err(|e| {
error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
.inspect_err(|e| {
debug_error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
})?;
let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event";
let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event"
&& incoming_pdu.state_key().is_none();
// Done with prev events, now handling the incoming event
// Done with prev events, now we can handle promoting the PDU
let pdu_id = Box::pin(self.upgrade_outlier_to_timeline_pdu(
incoming_pdu,
val,
@@ -1,23 +1,22 @@
use std::collections::{BTreeMap, HashMap, hash_map};
use conduwuit::{
Err, Event, EventTypeExt, PduEvent, Result, debug, debug_info, debug_warn, err, info,
matrix::StateKey, state_res::auth_check, trace, warn,
};
use futures::future::ready;
use ruma::{
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName,
canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules,
Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace,
};
use ruma::{CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName};
use super::{check_room_id, get_room_version_rules};
use crate::rooms::timeline::pdu_fits;
use super::get_room_version_rules;
impl super::Service {
/// Handles a PDU as an outlier, performing basic checks like signatures and
/// hashes, proclaimed event auth, and then adding it to the outlier tree.
///
/// This performs steps 1 through 4 of [S-S section 5.1][spec], returning
/// the parsed PDU and modified JSON object.
///
/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(name="handle_outlier", skip_all, fields(%event_id))]
#[tracing::instrument(name = "handle_outlier", skip_all)]
pub(super) async fn handle_outlier_pdu<'a, Pdu>(
&self,
origin: &'a ServerName,
@@ -29,61 +28,39 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
where
Pdu: Event + Send + Sync,
{
if !pdu_fits(&mut value.clone()) {
warn!(
"dropping incoming PDU {event_id} in room {room_id} from {origin} because it \
exceeds 65535 bytes or is otherwise too large."
// Skip outlier handling if we already have this event as either a timeline or
// outlier PDU.
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Database hit for {event_id} (event is either an outlier or already promoted), \
skipping outlier handling"
);
return Err!(Request(TooLarge("PDU is too large")));
value.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
return Ok((pdu_event, value));
}
// 1. Remove unsigned field
value.remove("unsigned");
// 2. Check signatures, otherwise drop
// 3. check content hash, redact if doesn't match
// 1. Check that the PDU follows the format for the room version
// (in this case, just size check)
let room_version_rules = get_room_version_rules(create_event)?;
let mut incoming_pdu = match self
.services
.server_keys
.verify_event(&value, &room_version_rules)
.await
{
| Ok(ruma::signatures::Verified::All) => {
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Already have event {event_id} as an outlier or timeline event, not \
re-processing"
);
value.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
check_room_id(room_id, &pdu_event)?;
return Ok((pdu_event, value));
}
value
},
| Ok(ruma::signatures::Verified::Signatures) => {
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Received a redacted copy of {event_id}, but we already knew about it. \
Re-using known content instead."
);
check_room_id(room_id, &pdu_event)?;
let obj = pdu_event.to_canonical_object();
return Ok((pdu_event, obj));
}
Self::pdu_format_check_1(&value, &room_version_rules, create_event.event_id())
.inspect_err(|e| {
info!(
err=?e,
"Dropping incoming PDU from {origin} because it violates the room event format"
);
})?;
debug_info!("Calculated hash does not match (redaction): {event_id}");
redact(value, &room_version_rules.redaction, None)
.map_err(|e| err!(Request(BadJson("Failed to redact {event_id}: {e}"))))?
},
| Err(e) => {
return Err!(Request(Forbidden(debug_error!(
"Signature verification failed for {event_id}: {e}"
))));
},
};
value.remove("unsigned");
let room_version_rules = get_room_version_rules(create_event)?;
// 2. Check signatures, otherwise drop.
// 3. Check content hash, redacting the event if it fails.
let mut incoming_pdu = self
.signature_hash_check_2_3(value, &room_version_rules)
.await?;
// Now that we have checked the signature and hashes we can add the eventID and
// convert to our PduEvent type
@@ -91,14 +68,11 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
let pdu_event = serde_json::from_value::<PduEvent>(
serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"),
)
.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
check_room_id(room_id, &pdu_event)?;
// TODO(nex): From hereon the event is not dropped, and thus always added as an
// outlier. However, we only do that at the end of this function, which means
// several duplicated calls to add_pdu_outlier. Shouldn't we just do it here
@@ -116,7 +90,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
for auth_event_id in pdu_event.auth_events() {
if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await {
check_room_id(room_id, &auth_event)?;
trace!("Found auth event {auth_event_id} for outlier event {event_id} locally");
auth_events.insert(auth_event_id.to_owned(), auth_event);
} else {
@@ -154,7 +127,7 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
}
// Ensure none of the auth events are rejected - if they are, reject too.
for auth_event_id in auth_events.keys() {
for (auth_event_id, auth_event) in &auth_events {
if self
.services
.pdu_metadata
@@ -166,15 +139,27 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
{auth_event_id}",
event_id,
);
self.services.pdu_metadata.mark_event_rejected(event_id);
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(
"Event has rejected auth event: {auth_event_id}"
)));
}
if auth_event.room_id_or_hash() != room_id {
debug_warn!(
%auth_event_id,
auth_event_room_id=%auth_event.room_id_or_hash(),
expected_room_id=%room_id,
"Rejecting incoming event which depends on an auth event in another room.",
);
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(
"Event depends on a cross-room auth event: {auth_event_id}"
)));
}
}
// 6. Reject "due to auth events" if the event doesn't pass auth based on the
// auth events
// 4. Reject "due to auth events" if the event doesn't pass auth based on the
// claimed auth events
debug!("Checking based on auth events");
let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len());
// Build map of auth events
@@ -184,8 +169,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
.expect("we just checked that we have all auth events")
.to_owned();
check_room_id(room_id, &auth_event)?;
let key = auth_event.kind().with_state_key(
auth_event
.state_key
@@ -197,98 +180,45 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
v.insert(auth_event);
},
| hash_map::Entry::Occupied(_) => {
self.services
.outlier
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
self.services.pdu_metadata.mark_event_rejected(event_id);
return Err!(Request(Forbidden(
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(debug_warn!(
"Auth event's type and state_key combination exists multiple times: {}, \
{}",
auth_event.kind,
auth_event.state_key().unwrap_or("")
)));
))));
},
}
}
if !self
.is_event_self_authorised(
.auth_state_check_4(
&pdu_event,
&room_version_rules,
create_event.as_pdu(),
&auth_events_by_key,
)
.await
.await?
{
self.services.pdu_metadata.mark_event_rejected(event_id);
self.services
.outlier
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
return Err!(Request(Forbidden(
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(debug_warn!(
"Event authorisation fails based on event's claimed auth events"
)));
))));
}
trace!("Validation successful.");
// 7. Persist the event as an outlier.
self.services
.outlier
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
trace!("Added pdu as outlier.");
debug!("PDU passed checks and has been persisted as an outlier");
Ok((pdu_event, incoming_pdu))
}
/// Helper method that turns the return value of `is_event_self_authorised`
/// into a `Result` depending on the value.
///
/// If the event is not authorised, a Forbidden error is returned.
/// Otherwise, an empty `Ok`.
pub(super) async fn is_event_self_authorised(
&self,
pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
) -> bool {
self.expect_event_is_self_authorised(
pdu,
room_version_rules,
create_event,
auth_events_by_key,
)
.await
.is_ok()
}
/// Checks PDU check 4: Passes authorisation rules based on the event's auth
/// events ([spec]).
///
/// If the auth check fails, false is returned, otherwise true.
///
/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
pub(super) async fn expect_event_is_self_authorised(
&self,
pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
) -> Result<bool> {
let state_fetch = |ty: &StateEventType, sk: &str| {
let key = (ty.to_owned(), sk.into());
ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))
};
auth_check(
room_version_rules,
pdu,
None, // TODO: third party invite
state_fetch,
create_event,
)
.await
.map_err(|e| err!("Event self-authentication failed: {e:?}"))
/// Marks the event as rejected and then saves it as an outlier.
pub(super) fn reject_and_persist(&self, event_id: &EventId, pdu: &CanonicalJsonObject) {
self.services.pdu_metadata.mark_event_rejected(event_id);
self.services.outlier.add_pdu_outlier(event_id, pdu);
}
}
+3 -20
View File
@@ -6,6 +6,7 @@
mod handle_incoming_pdu;
mod handle_outlier_pdu;
mod parse_incoming_pdu;
pub mod pdu_checks;
mod policy_server;
mod resolve_state;
mod state_at_incoming;
@@ -19,12 +20,12 @@
DagBuilderTree, GET_MISSING_EVENTS_MAX_BATCH_SIZE, build_local_dag,
};
use ruma::{
OwnedEventId, OwnedRoomId, RoomId, events::room::create::RoomCreateEventContent,
OwnedEventId, OwnedRoomId, events::room::create::RoomCreateEventContent,
room_version_rules::RoomVersionRules,
};
use tokio::sync::{Notify, mpsc};
use crate::{Dep, globals, rooms, sending, server_keys, sync};
use crate::{Dep, globals, rooms, sending, server_keys};
pub struct Service {
pub mutex_federation: RoomMutexMap,
pub federation_handletime: SyncRwLock<HandleTimeMap>,
@@ -47,7 +48,6 @@ struct Services {
state_cache: Dep<rooms::state_cache::Service>,
state_accessor: Dep<rooms::state_accessor::Service>,
state_compressor: Dep<rooms::state_compressor::Service>,
sync: Dep<sync::Service>,
timeline: Dep<rooms::timeline::Service>,
server: Arc<Server>,
}
@@ -78,7 +78,6 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
state_compressor: args
.depend::<rooms::state_compressor::Service>("rooms::state_compressor"),
sync: args.depend::<sync::Service>("sync"),
timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
server: args.server.clone(),
},
@@ -116,22 +115,6 @@ async fn event_fetch(&self, event_id: OwnedEventId) -> Option<PduEvent> {
}
}
fn check_room_id<Pdu: Event>(room_id: &RoomId, pdu: &Pdu) -> Result {
if pdu
.room_id()
.is_some_and(|claimed_room_id| claimed_room_id != room_id)
{
return Err!(Request(InvalidParam(error!(
pdu_event_id = %pdu.event_id(),
pdu_room_id = pdu.room_id().map(tracing::field::display),
%room_id,
"Found event from room in room",
))));
}
Ok(())
}
fn get_room_version_rules<Pdu: Event>(create_event: &Pdu) -> Result<RoomVersionRules> {
let content: RoomCreateEventContent = create_event.get_content()?;
let Some(room_version_rules) = content.room_version.rules() else {
@@ -23,7 +23,12 @@ fn extract_room_id(event_type: &str, pdu: &CanonicalJsonObject) -> Result<OwnedR
.map_err(|e| err!(Request(BadJson("Invalid room_id {room_id:?} in pdu: {e}"))));
}
// If there's no room ID, and this is not a create event, it is illegal.
if event_type != "m.room.create" || pdu.get("state_key").is_none() {
let is_create = event_type == "m.room.create"
&& pdu
.get("state_key")
.and_then(|v| v.as_str())
.is_some_and(str::is_empty);
if !is_create {
return Err!(Request(BadJson("Missing room_id in pdu")));
}
@@ -42,10 +47,7 @@ fn extract_room_id(event_type: &str, pdu: &CanonicalJsonObject) -> Result<OwnedR
return Err!(Request(BadJson("Unknown room version in pdu")));
};
if !room_version_rules
.authorization
.room_create_event_id_as_room_id
{
if room_version_rules.event_format.require_room_create_room_id {
return Err!(Request(BadJson("Missing room_id in pdu")));
}
@@ -80,33 +82,10 @@ pub(super) fn expect_event_id_array(
}
impl super::Service {
/// Performs some basic validation on the PDU to make sure it's not
/// obviously malformed. This is not a full validation, but guards against
/// extreme errors.
///
/// Currently, this just validates that prev/auth events are within
/// acceptable ranges. Other servers do some additional things like
/// checking depth range, but serde will do that later when converting the
/// object to a PduEvent.
pub fn validate_pdu(&self, pdu: &CanonicalJsonObject) -> Result {
// Since v3:
// `event_id` should not be present on the PDU.
// NOTE: The above is ignored since technically it's still allowed to be
// included, but should be ignored instead.
// `auth_events` and `prev_events` must be an array of event IDs
let auth_events = expect_event_id_array(pdu, "auth_events")?;
if auth_events.len() > 10 {
return Err!(Request(BadJson("PDU has too many auth events")));
}
let prev_events = expect_event_id_array(pdu, "prev_events")?;
if prev_events.len() > 20 {
return Err!(Request(BadJson("PDU has too many prev events")));
}
Ok(())
}
pub async fn parse_incoming_pdu_with_known_room(
&self,
/// Parses an incoming PDU JSON object, generating an event ID for it. Does
/// not insert the event ID into the returned object. Does not discover the
/// room ID.
pub(super) fn parse_incoming_pdu_with_known_room(
pdu: &RawJsonValue,
room_version_rules: &RoomVersionRules,
) -> Result<(OwnedEventId, CanonicalJsonObject)> {
@@ -114,10 +93,13 @@ pub async fn parse_incoming_pdu_with_known_room(
gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| {
err!(Request(InvalidParam("Could not convert event to canonical json: {e}")))
})?;
self.validate_pdu(&value)?;
// NOTE: validation checks are now performed by `pdu_format_check_1`.
Ok((event_id, value))
}
/// Parses an incoming PDU JSON object, generating an event ID for it and
/// attempts to discover the associated room ID. Does not insert the event
/// ID into the returned object.
pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
let value = serde_json::from_str::<CanonicalJsonObject>(pdu.get()).map_err(|e| {
err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}")))
@@ -142,7 +124,7 @@ pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
gen_event_id_canonical_json(pdu, &room_version_rules).map_err(|e| {
err!(Request(InvalidParam("Could not convert event to canonical json: {e}")))
})?;
self.validate_pdu(&value)?;
// NOTE: validation checks are now performed by `pdu_format_check_1`.
Ok((room_id, event_id, value))
}
}
@@ -0,0 +1,289 @@
use std::collections::HashMap;
use conduwuit::{
Err, Event, EventTypeExt, PduEvent, Result, debug, debug::DebugInspect, debug_error,
debug_info, err, info, matrix::StateKey, state_res, trace,
};
use futures::future::ready;
use ruma::{
CanonicalJsonObject, EventId, OwnedEventId, ServerName, api::error::ErrorKind,
canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules,
};
use crate::rooms::{
event_handler::parse_incoming_pdu::expect_event_id_array, timeline::pdu_fits,
};
impl super::Service {
/// Checks that the PDU conforms to the PDU format (check 1). This is
/// already mostly done during deserialisation, so this function just checks
/// that the PDU isn't a too large.
pub(super) fn pdu_format_check_1(
pdu_json: &CanonicalJsonObject,
room_version_rules: &RoomVersionRules,
create_event_id: &EventId,
) -> Result<()> {
let event_format = &room_version_rules.event_format;
// NOTE: if we do any more validation outside of deserialisation, it has to be
// done here.
if !pdu_fits(pdu_json) {
return Err!(Request(TooLarge("PDU is too large")));
}
if event_format.require_room_create_room_id {
if pdu_json.get("room_id").is_none() {
return Err!(Request(BadJson("Missing required PDU field: `room_id`")));
}
}
let auth_events = expect_event_id_array(pdu_json, "auth_events")?;
if auth_events.len() > 10 {
return Err!(Request(BadJson("PDU has too many auth events")));
}
let create_event_in_auth_events = auth_events.iter().any(|id| id == create_event_id);
if !event_format.allow_room_create_in_auth_events && create_event_in_auth_events {
return Err!(Request(BadJson("PDU references a create event")));
}
let prev_events = expect_event_id_array(pdu_json, "prev_events")?;
if prev_events.len() > 20 {
return Err!(Request(BadJson("PDU has too many prev events")));
}
Ok(())
}
/// Checks that the PDU has a valid signature (check 2), and redacts it if
/// the content hash verification fails (check 3), returning the
/// potentially modified JSON. Returns an error if the PDU cannot be
/// redacted, or fails signature verification.
pub(super) async fn signature_hash_check_2_3(
&self,
pdu_json: CanonicalJsonObject,
room_version_rules: &RoomVersionRules,
) -> Result<CanonicalJsonObject> {
match self
.services
.server_keys
.verify_event(&pdu_json, room_version_rules)
.await
{
| Ok(ruma::signatures::Verified::All) => {
trace!("Signatures and hashes verified successfully");
Ok(pdu_json)
},
| Ok(ruma::signatures::Verified::Signatures) => {
debug_info!("Content hash mismatch, redacting event and continuing");
let redacted = redact(pdu_json, &room_version_rules.redaction, None)
.map_err(|e| err!(Request(BadJson("Unable to redact event: {e}"))))?;
Ok(redacted)
},
| Err(e) => {
Err!(Request(Forbidden(debug_error!("Signature verification failed: {e}"))))
},
}
}
/// Checks PDU check 4: Passes authorisation rules based on the event's auth
/// events ([spec]).
///
/// If the auth check fails, false is returned, otherwise true.
///
/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
pub(super) async fn auth_state_check_4(
&self,
pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
) -> Result<bool> {
let state_fetch = |ty: &StateEventType, sk: &str| {
let key = (ty.to_owned(), sk.into());
ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))
};
state_res::event_auth::auth_check(
room_version_rules,
pdu,
None, // TODO: third party invite
state_fetch,
create_event,
)
.await
.map_err(|e| err!("Event self-authentication failed: {e:?}"))
}
/// Checks that the event passes PDU check 5, which ensures that the event
/// is authorised based on the state before the event (which is the resolved
/// state across all prev events).
///
/// Returns a boolean indicating whether the event is authorised, and also
/// the resolved state before the event for later use. Returns an error if
/// state fetching or auth checking fails.
pub(super) async fn state_before_check_5(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
origin: &ServerName,
) -> Result<(bool, HashMap<u64, OwnedEventId>)> {
debug!(
event_id = %incoming_pdu.event_id,
"Resolving state at event"
);
let room_id = incoming_pdu.room_id_or_hash();
// If the incoming event only has one prev event, we can just use the state at
// that event, but otherwise we have to resolve across each fork. If we're
// missing even one of the prev events, we have to ask a remote server for help.
//
// TODO: this can be optimised by only loading auth chain events into memory,
// rather than the entire state.
let state_before = self
.state_before_incoming(&incoming_pdu, room_version_rules)
.await?;
let state_before = match state_before {
| Some(s) => s,
| None => {
trace!("Could not calculate incoming state, asking remote {origin} for it");
self.fetch_state(origin, create_event, &room_id, incoming_pdu.event_id())
.await
.inspect_err(|e| {
debug_error!("Could not fetch state from {origin}: {e}");
})?
},
};
if state_before.is_empty()
&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
{
// This can happen if the remote sends an event but cannot be reached to fetch
// the state at it, and all other servers in the room (which might just be the
// unreachable server) are unable to provide required info.
// returning an error here allows the upgrade to be attempted at another time.
return Err!(Request(Forbidden("Could not resolve incoming state before event")));
}
trace!(state_events = state_before.len(), "Calculated incoming state");
let state_fetch_state = &state_before;
let state_fetch = |k: StateEventType, s: StateKey| async move {
let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
let event_id = state_fetch_state.get(&shortstatekey)?;
self.services.timeline.get_pdu(event_id).await.ok()
};
debug!(
event_id = %incoming_pdu.event_id,
"Running state-before auth check"
);
// PDU check: 5
let auth_check = state_res::event_auth::auth_check(
room_version_rules,
incoming_pdu,
None, // TODO: third party invite
|ty, sk| state_fetch(ty.clone(), sk.into()),
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
Ok((auth_check, state_before))
}
/// Checks that the event passes PDU check 6, which ensures that the event
/// is authorised based on the room's current state (which is the resolved
/// state across all current forward extremities).
///
/// Returns a boolean indicating whether the event is authorised, or an
/// error if the auth check fails.
pub(super) async fn current_state_check_6(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
) -> Result<bool> {
debug!(
event_id = %incoming_pdu.event_id,
"Gathering auth events"
);
let auth_events = self
.services
.state
.get_auth_events(
&incoming_pdu.room_id_or_hash(),
incoming_pdu.kind(),
incoming_pdu.sender(),
incoming_pdu.state_key(),
incoming_pdu.content(),
room_version_rules,
)
.await?;
let state_fetch = |k: &StateEventType, s: &str| {
let key = k.with_state_key(s);
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
debug!(
event_id = %incoming_pdu.event_id,
"Running current state auth check"
);
state_res::event_auth::auth_check(
room_version_rules,
incoming_pdu,
None, // third-party invite
state_fetch,
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))
}
/// Performs PDU check 7 - does the policy server allow this event.
///
/// If the policy server forbids the event, false is returned. If there is a
/// problem contacting the policy server, or it returns an unrecognised
/// response, an appropriate error is returned.
pub(super) async fn policy_server_check_7(
&self,
incoming_pdu: &PduEvent,
pdu_json: &mut CanonicalJsonObject,
room_version_rules: &RoomVersionRules,
) -> Result<bool> {
let event_id = pdu_json
.remove("event_id")
.expect("event_id should be present in pdu_json at this stage");
if let Err(e) = self
.policy_server_allows_event(
incoming_pdu,
pdu_json,
&incoming_pdu.room_id_or_hash(),
room_version_rules,
true,
)
.await
.debug_inspect(|()| {
debug!(
event_id = %incoming_pdu.event_id,
"Event has passed policy server check."
);
}) {
return if matches!(e.kind(), ErrorKind::Forbidden) {
info!(
event_id = %incoming_pdu.event_id,
error = %e,
"Event has been marked as spam by policy server: {}",
e.message(),
);
Ok(false)
} else {
Err(e)
};
}
pdu_json.insert("event_id".to_owned(), event_id);
Ok(true)
}
}
@@ -16,10 +16,31 @@
use crate::rooms::short::ShortStateHash;
impl super::Service {
// TODO: if we know the prev_events of the incoming event we can avoid the
// request and build the state from a known point and resolve if > 1 prev_event
/// Resolves the state before the incoming event.
///
/// If we do not know enough information to resolve the state, `Ok(None)` is
/// returned, and the caller will have to figure it out some other way (e.g.
/// by fetching the state from a remote server).
pub(super) async fn state_before_incoming<Pdu>(
&self,
incoming_pdu: &Pdu,
room_version_rules: &RoomVersionRules,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
where
Pdu: Event + Send + Sync,
{
if incoming_pdu.prev_events().count() == 1 {
self.state_before_incoming_degree_one(incoming_pdu).await
} else {
self.state_before_incoming_resolved(incoming_pdu, room_version_rules)
.await
}
}
/// Determines the state before the incoming pdu, when it has only one prev
/// event. This is a special case that does not require state resolution.
#[tracing::instrument(name = "state", level = "debug", skip_all)]
pub(super) async fn state_at_incoming_degree_one<Pdu>(
async fn state_before_incoming_degree_one<Pdu>(
&self,
incoming_pdu: &Pdu,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
@@ -66,7 +87,7 @@ pub(super) async fn state_at_incoming_degree_one<Pdu>(
.await;
state.insert(shortstatekey, prev_event.to_owned());
// Now it's the state after the pdu
// Now it's the state at the pdu
}
debug_assert!(!state.is_empty(), "should be returning None for empty HashMap result");
@@ -74,11 +95,14 @@ pub(super) async fn state_at_incoming_degree_one<Pdu>(
Ok(Some(state))
}
/// Resolves the state before the incoming pdu across all of its prev
/// events. If we do not know enough information to resolve the state,
/// `Ok(None)` is returned, and the caller will have to figure it out some
/// other way (e.g. by fetching the state from a remote server).
#[tracing::instrument(name = "state", level = "debug", skip_all)]
pub(super) async fn state_at_incoming_resolved<Pdu>(
async fn state_before_incoming_resolved<Pdu>(
&self,
incoming_pdu: &Pdu,
room_id: &RoomId,
room_version_rules: &RoomVersionRules,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
where
@@ -108,6 +132,7 @@ pub(super) async fn state_at_incoming_resolved<Pdu>(
};
trace!("Calculating fork states...");
let room_id = &incoming_pdu.room_id_or_hash();
let (fork_states, auth_chain_sets): (Vec<StateMap<_>>, Vec<HashSet<_>>) =
extremity_sstatehashes
.into_iter()
@@ -145,6 +170,8 @@ pub(super) async fn state_at_incoming_resolved<Pdu>(
.await
}
/// Determines the state at an incoming fork (aka the state at a prev
/// event), returning the resolved state and its associated auth chain.
async fn state_at_incoming_fork<Pdu>(
&self,
room_id: &RoomId,
@@ -1,27 +1,15 @@
use std::{borrow::Borrow, sync::Arc, time::Instant};
use std::time::Instant;
use conduwuit::{
Err, Result, debug, debug_error, debug_info, err, info, is_equal_to,
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
result::DebugInspect,
Err, Result, debug, debug_info, debug_warn, is_true,
matrix::{Event, PduEvent},
trace,
utils::{
IterStream,
stream::{BroadbandExt, ReadyExt},
},
warn,
};
use futures::{FutureExt, StreamExt, future::ready};
use ruma::{
CanonicalJsonObject, RoomId, ServerName, api::error::ErrorKind, events::StateEventType,
};
use ruma::{CanonicalJsonObject, RoomId, ServerName, events::StateEventType};
use tokio::join;
use super::get_room_version_rules;
use crate::rooms::{
state_compressor::{CompressedState, HashSetCompressStateEvent},
timeline::RawPduId,
};
use crate::rooms::timeline::RawPduId;
impl super::Service {
#[tracing::instrument(name="upgrade_outlier", skip_all, fields(event_id=%incoming_pdu.event_id()))]
@@ -46,16 +34,24 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
trace!(event_id=%incoming_pdu.event_id(), "Skipping upgrade of already upgraded PDU");
return Ok(Some(id));
} else if rejected {
return Err!(Request(Forbidden("Event has been rejected")));
return Err!(Request(Forbidden(debug_info!("Event has been rejected"))));
} else if soft_failed {
return Err!(Request(Forbidden("Event has been soft-failed")));
// Soft-failed events cannot be promoted.
return Err!(Request(Forbidden(debug_info!("Event has been soft-failed"))));
}
// These should never happen, but they're good last-minute sanity checks to
// ensure we never promote totally illegal events.
assert_eq!(
*create_event.kind(),
StateEventType::RoomCreate.into(),
"tried to upgrade a PDU with a create_event that is not a room create event"
);
assert_eq!(
incoming_pdu.room_id_or_hash(),
*room_id,
"room ID mismatch: PDU room ID differs from parameter"
);
debug!(
event_id = %incoming_pdu.event_id,
@@ -65,326 +61,118 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
let min_depth = self.services.metadata.get_mindepth(room_id).await;
let room_version_rules = get_room_version_rules(create_event)?;
// 10. Fetch missing state and auth chain events by calling /state_ids at
// backwards extremities doing all the checks in this list starting at 1.
// These are not timeline events.
debug!(
event_id = %incoming_pdu.event_id,
"Resolving state at event"
);
let state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
self.state_at_incoming_degree_one(&incoming_pdu).await?
} else {
self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_rules)
.await?
};
let state_at_incoming_event = match state_at_incoming_event {
| Some(s) => s,
| None => {
trace!("Could not calculate incoming state, asking remote {origin} for it");
self.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
.await
.debug_inspect_err(|e| {
debug_error!("Could not fetch state from {origin}: {e}");
})?
},
};
if state_at_incoming_event.is_empty()
&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
{
// This can happen if the remote sends an event but cannot be reached to fetch
// the state at it, and all other servers in the room (which might just be the
// unreachable server) are unable to provide required info.
// returning an error here allows the upgrade to be attempted at another time.
return Err!(Request(Forbidden("Could not resolve incoming state at event")));
}
trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state");
debug!(
event_id = %incoming_pdu.event_id,
"Performing auth check to upgrade"
);
// 11. Check the auth of the event passes based on the state of the event
let state_fetch_state = &state_at_incoming_event;
let state_fetch = |k: StateEventType, s: StateKey| async move {
let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
let event_id = state_fetch_state.get(&shortstatekey)?;
self.services.timeline.get_pdu(event_id).await.ok()
};
debug!(
event_id = %incoming_pdu.event_id,
"Running initial auth check"
);
// PDU check: 5
let auth_check = state_res::event_auth::auth_check(
&room_version_rules,
&incoming_pdu,
None, // TODO: third party invite
|ty, sk| state_fetch(ty.clone(), sk.into()),
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
if !auth_check {
self.services
.pdu_metadata
.mark_event_rejected(incoming_pdu.event_id());
return Err!(Request(Forbidden(
"Event authorisation fails based on the state before the event"
)));
}
debug!(
event_id = %incoming_pdu.event_id,
"Gathering auth events"
);
let auth_events = self
.services
.state
.get_auth_events(
room_id,
incoming_pdu.kind(),
incoming_pdu.sender(),
incoming_pdu.state_key(),
incoming_pdu.content(),
&room_version_rules,
)
// We now need to resolve the state before the event so that we can perform PDU
// check 5 (event auth passes based on state before the event). To do this, we
// either need to have all the prev events locally, or ask a remote server
// for the state at the event.
let (passes_state_before, state_before) = self
.state_before_check_5(&incoming_pdu, &room_version_rules, create_event, origin)
.await?;
let state_fetch = |k: &StateEventType, s: &str| {
let key = k.with_state_key(s);
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
debug!(
event_id = %incoming_pdu.event_id,
"Running auth check with claimed state auth"
);
// PDU check: 6
let auth_check = state_res::event_auth::auth_check(
&room_version_rules,
&incoming_pdu,
None, // third-party invite
state_fetch,
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
if !auth_check {
warn!(
event_id = %incoming_pdu.event_id,
"Event authorization fails based on the current state of the room"
);
if !passes_state_before {
self.reject_and_persist(incoming_pdu.event_id(), &val);
return Err!(Request(Forbidden(debug_warn!(
"Event authorisation fails based on the state before the event"
))));
}
// Soft fail check before doing state res
debug!(
event_id = %incoming_pdu.event_id,
"Performing soft-fail check"
);
let mut soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_rules)) {
| (false, _) => true,
| (true, None) => false,
| (true, Some(redact_id)) => {
if !self
.services
.state_accessor
.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
.await?
{
warn!(redacts = %redact_id, "User is not allowed to redact event");
true
} else {
false
}
},
};
// Now that we know the event passes both self-authentication, and
// authentication based on the state before the event, we need to check that it
// passes based on the *current* room state (state across all forward
// extremities). If it doesn't, we accept it, but soft-fail it, and this
// prevents it being promoted.
// 13. Use state resolution to find new room state
// We start looking at current room state now, so lets lock the room
// We lock the room here to prevent the current state from changing beneath us
// mid-check.
trace!(
room_id = %room_id,
"Locking the room"
);
let state_lock = self.services.state.mutex.lock(room_id).await;
let state_ids_compressed: Arc<CompressedState> = self
.services
.state_compressor
.compress_state_events(
state_at_incoming_event
.iter()
.map(|(ssk, eid)| (ssk, eid.borrow())),
)
.collect()
.map(Arc::new)
.await;
if incoming_pdu.state_key().is_some() {
debug!("Event is a state-event. Deriving new room state");
// We also add state after incoming event to the fork states
let mut state_after = state_at_incoming_event.clone();
if let Some(state_key) = incoming_pdu.state_key() {
let shortstatekey = self
.services
.short
.get_or_create_shortstatekey(
&incoming_pdu.kind().to_string().into(),
state_key,
)
.await;
let event_id = incoming_pdu.event_id();
state_after.insert(shortstatekey, event_id.to_owned());
}
let new_room_state = self
.resolve_state(room_id, &room_version_rules, state_after)
.await?;
// Set the new room state to the resolved state
debug!("Forcing new room state");
let HashSetCompressStateEvent { shortstatehash, added, removed } = self
.services
.state_compressor
.save_state(room_id, new_room_state)
.await?;
self.services
.state
.force_state(room_id, shortstatehash, added, removed, &state_lock)
.await?;
}
if !soft_fail {
// Don't call the below checks on events that have already soft-failed, there's
// no reason to re-calculate that.
// 14-pre. ask the policy server to sign the event, if possible
debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
let tmp_evt_id = val.remove("event_id");
if let Err(e) = self
.policy_server_allows_event(
&incoming_pdu,
&mut val,
room_id,
&room_version_rules,
true,
)
.await
{
if matches!(e.kind(), ErrorKind::Forbidden) {
info!(
event_id = %incoming_pdu.event_id,
error = %e,
"Event has been marked as spam by policy server: {}",
e.message(),
let passes_current_state = self
.current_state_check_6(&incoming_pdu, &room_version_rules, create_event)
.await
.inspect(|passes| {
if !*passes {
debug_warn!(
"Event authorisation fails based on the current room state - will be \
soft-failed"
);
soft_fail = true;
} else {
return Err(e);
}
} else {
debug!(
event_id = %incoming_pdu.event_id,
"Event has passed policy server check."
);
}
if let Some(id) = tmp_evt_id {
val.insert("event_id".to_owned(), id);
}
})?;
// Additionally, if this is a redaction for a soft-failed event, we soft-fail it
// also.
// Determine whether this PDU should be soft-failed.
// If the auth check failed, invariably yes. Otherwise, only if the user isn't
// allowed to redact the target event (if any).
let mut should_soft_fail =
match (passes_current_state, incoming_pdu.redacts_id(&room_version_rules)) {
| (false, _) => true,
| (true, None) => false,
| (true, Some(redact_id)) => self
.services
.state_accessor
.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
.await
.is_ok_and(is_true!()),
};
// TODO: this is supposed to hide redactions from policy servers, however, for
// full efficacy it also needs to hide redactions for unknown events. This
// needs to be investigated at a later time.
if !should_soft_fail {
// Now we can perform check 7, which is ensuring the event passes policy server
// checks.
// We explicitly only do this if we aren't already going to soft-fail the event,
// since the policy server refusing this event also soft-fails it.
debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
should_soft_fail = !self
.policy_server_check_7(&incoming_pdu, &mut val, &room_version_rules)
.await
.inspect(|passes| {
if !*passes {
debug_warn!(
"Event did not pass the policy server check and will be soft-failed"
);
}
})?;
// TODO: this is supposed to hide redactions from policy servers and janitorial
// bots, however, for full efficacy it also needs to hide redactions for
// unknown events. This needs to be investigated at a later time.
if let Some(redact_id) = incoming_pdu.redacts_id(&room_version_rules) {
debug!(
redact_id = %redact_id,
"Checking if redaction is for a soft-failed event"
"Checking if redaction is for a soft-failed/rejected event"
);
if self
if !self
.services
.pdu_metadata
.is_event_soft_failed(&redact_id)
.is_event_accepted(&redact_id)
.await
{
info!(
redact_id = %redact_id,
"Redaction is for a soft-failed event"
debug_info!(
"Soft-failing valid redaction because it targets a non-accepted event"
);
soft_fail = true;
should_soft_fail = true;
}
}
}
// The PDU has now passed all checks! We can now promote it (or soft-fail it if
// the verdict is such).
trace!("Appending pdu to timeline");
let mut extremities: Vec<_> = self
.services
.state
.get_forward_extremities(room_id)
.collect()
.await;
if !soft_fail {
// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
// are not added as forward extremities.
// Now we calculate the set of extremities this room has after the incoming
// event has been applied. We start with the previous extremities (aka leaves)
trace!("Calculating extremities");
extremities = extremities
.into_iter()
.stream()
.ready_filter(|event_id| {
// Remove any that are referenced by this incoming event's prev_events
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
})
.broad_filter_map(|event_id| async move {
// Only keep those extremities were not referenced yet
self.services
.pdu_metadata
.is_event_referenced(room_id, &event_id)
.await
.eq(&false)
.then_some(event_id)
})
.collect::<Vec<_>>()
.await;
extremities.push(incoming_pdu.event_id().to_owned());
debug!(
"Retained {} extremities checked against {} prev_events",
extremities.len(),
incoming_pdu.prev_events().count()
);
assert!(!extremities.is_empty(), "extremities must not empty");
}
let pdu_id = self
.services
.timeline
.append_incoming_pdu(
&incoming_pdu,
val,
extremities.iter().map(Borrow::borrow),
state_ids_compressed,
soft_fail,
&room_version_rules,
state_before,
should_soft_fail,
&state_lock,
room_id,
)
.await?;
if soft_fail {
self.services
.pdu_metadata
.mark_event_soft_failed(incoming_pdu.event_id());
info!(
if should_soft_fail {
debug_info!(
elapsed = ?timer.elapsed(),
event_id = %incoming_pdu.event_id,
"Event was soft failed"
+1 -1
View File
@@ -520,7 +520,7 @@ pub async fn join_remote_room(
return state;
},
};
if !pdu_fits(&mut value.clone()) {
if !pdu_fits(&value.clone()) {
warn!(
"dropping incoming PDU {event_id} in room {room_id} from room join \
because it exceeds 65535 bytes or is otherwise too large."
+169 -26
View File
@@ -1,69 +1,113 @@
use std::{
collections::{BTreeMap, HashSet},
borrow::Borrow,
collections::{BTreeMap, HashMap, HashSet},
sync::Arc,
};
use conduwuit::{
debug_warn,
Result, debug, debug_warn, is_equal_to,
matrix::{Event, PduEvent},
pdu::{Count, ShortRoomId},
trace,
utils::{IterStream, TryFutureExtExt, stream::BroadbandExt},
utils::{
IterStream, TryFutureExtExt,
mutex_map::Guard,
stream::{BroadbandExt, ReadyExt},
},
warn,
};
use conduwuit_core::{
Result, err, error,
matrix::{
event::Event,
pdu::{PduCount, PduEvent, PduId, RawPduId},
},
utils::{self, ReadyExt},
err, error,
matrix::pdu::{PduCount, PduId, RawPduId},
utils::{self},
};
use futures::{StreamExt, TryFutureExt};
use futures::{FutureExt, StreamExt, TryFutureExt};
use ruma::{
CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, UserId,
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId,
RoomVersionId, UserId,
events::{
GlobalAccountDataEventType, TimelineEventType,
push_rules::PushRulesEvent,
room::{encrypted::Relation, redaction::RoomRedactionEventContent},
},
push::{Action, Ruleset, Tweak},
room_version_rules::RoomVersionRules,
};
use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard};
use crate::{appservice::RegistrationInfo, rooms::state_compressor::CompressedState};
use crate::{
appservice::RegistrationInfo,
rooms::state_compressor::{CompressedState, HashSetCompressStateEvent},
};
impl super::Service {
/// Append the incoming event setting the state snapshot to the state from
/// the server that sent the event.
#[allow(clippy::too_many_arguments)]
pub async fn append_incoming_pdu<'a, Leaves>(
pub async fn append_incoming_pdu<'a>(
&'a self,
pdu: &'a PduEvent,
pdu_json: CanonicalJsonObject,
new_room_leaves: Leaves,
state_ids_compressed: Arc<CompressedState>,
soft_fail: bool,
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
should_soft_fail: bool,
state_lock: &'a RoomMutexGuard,
room_id: &'a RoomId,
) -> Result<Option<RawPduId>>
where
Leaves: Iterator<Item = &'a EventId> + Send + 'a,
{
) -> Result<Option<RawPduId>> {
let room_id = pdu.room_id_or_hash();
let sb = state_before.iter().map(|(ssk, eid)| (ssk, eid.as_ref()));
let state_ids_compressed: Arc<CompressedState> = self
.services
.state_compressor
.compress_state_events(sb)
.collect()
.map(Arc::new)
.await;
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
self.services
.state
.set_event_state(&pdu.event_id, room_id, state_ids_compressed)
.set_event_state(&pdu.event_id, &room_id, state_ids_compressed)
.await?;
if soft_fail {
if should_soft_fail {
// Nothing else to do with a soft-failed event.
self.services
.pdu_metadata
.mark_event_soft_failed(pdu.event_id());
return Ok(None);
}
// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
// are not added as forward extremities.
// This also means we set the state here.
// We do this BEFORE setting the extremities so that there's never a point in
// time where we have fresh extremities referencing stale state.
let mut extremities: Vec<_> = self
.services
.state
.get_forward_extremities(&room_id)
.collect()
.await;
extremities = self
.progress_state_and_extremities(
pdu,
room_version_rules,
state_before.clone(),
extremities,
state_lock,
)
.await?;
let pdu_id = self
.append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id)
.append_pdu(
pdu,
pdu_json,
extremities.iter().map(Borrow::borrow),
state_lock,
&room_id,
)
.await?;
// Process admin commands for federation events
@@ -86,11 +130,110 @@ pub async fn append_incoming_pdu<'a, Leaves>(
}
}
self.services.sync.wake_all_joined(room_id).await;
self.services.sync.wake_all_joined(&room_id).await;
Ok(Some(pdu_id))
}
/// Derives new room state from the incoming event and filters forward
/// extremities accordingly. Does not set forward extremities.
///
/// Only call this function if the incoming PDU is not soft-failed or
/// rejected.
async fn progress_state_and_extremities(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
forward_extremities: Vec<OwnedEventId>,
state_lock: &Guard<OwnedRoomId, ()>,
) -> Result<Vec<OwnedEventId>> {
if incoming_pdu.state_key().is_some() {
debug!("Event is a state-event. Deriving new room state");
self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock)
.await?;
}
// Now we calculate the set of extremities this room has after the incoming
// event has been applied. We start with the previous extremities
trace!("Calculating extremities");
let mut forward_extremities = forward_extremities
.into_iter()
.stream()
.ready_filter(|event_id| {
// Remove any that are referenced by this incoming event's prev_events
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
})
.broad_filter_map(|event_id| async move {
// Only keep those extremities were not referenced yet
self.services
.pdu_metadata
.is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id)
.await
.eq(&false)
.then_some(event_id)
})
.collect::<Vec<_>>()
.await;
forward_extremities.push(incoming_pdu.event_id().to_owned());
debug!(
"Retained {} extremities checked against {} prev_events",
forward_extremities.len(),
incoming_pdu.prev_events().count()
);
assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty");
Ok(forward_extremities)
}
/// Derives a new room state by adding the incoming PDU to the state before
/// it to create the state at, which then becomes the current room state.
///
/// The caller MUST ensure forward extremities are set appropriately,
/// including this incoming pdu, either before or after calling this
/// function. Failing to do so will result in an inconsistent current state
/// cache, which may affect event authentication.
async fn derive_new_state(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
state_lock: &Guard<OwnedRoomId, ()>,
) -> Result {
let room_id = incoming_pdu.room_id_or_hash();
// We also add state after incoming event to the fork states
let mut state_at_incoming_event = state_before;
let shortstatekey = self
.services
.short
.get_or_create_shortstatekey(
&incoming_pdu.kind().to_string().into(),
incoming_pdu.state_key().unwrap(),
)
.await;
let event_id = incoming_pdu.event_id();
state_at_incoming_event.insert(shortstatekey, event_id.to_owned());
debug!("Resolving new room state");
let new_room_state = self
.services
.event_handler
.resolve_state(&room_id, room_version_rules, state_at_incoming_event)
.await?;
debug!("Forcing new room state");
let HashSetCompressStateEvent { shortstatehash, added, removed } = self
.services
.state_compressor
.save_state(&room_id, new_room_state)
.await?;
self.services
.state
.force_state(&room_id, shortstatehash, added, removed, state_lock)
.await
}
/// Populates the unsigned data of a PDU
async fn populate_unsigned(&self, pdu: &PduEvent, pdu_json: &mut CanonicalJsonObject) {
let Some(state_key) = pdu.state_key() else {
+2 -2
View File
@@ -143,7 +143,7 @@ pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Resu
| Ok(value) => {
self.services
.event_handler
.handle_incoming_pdu(&backfill_server, room_id, event_id, value, false)
.handle_incoming_pdu(&backfill_server, room_id, event_id, value, true)
.boxed()
.await?;
debug!("Successfully backfilled {event_id} from {backfill_server}");
@@ -185,7 +185,7 @@ pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) ->
self.services
.event_handler
.handle_incoming_pdu(origin, &room_id, &event_id, value, false)
.handle_incoming_pdu(origin, &room_id, &event_id, value, true)
.boxed()
.await?;
+4 -2
View File
@@ -22,7 +22,9 @@
use super::RoomMutexGuard;
pub fn pdu_fits(owned_obj: &mut CanonicalJsonObject) -> bool {
/// Ensures the given PDU fits inside the size limits for a PDU.
#[must_use]
pub fn pdu_fits(owned_obj: &CanonicalJsonObject) -> bool {
// room IDs, event IDs, senders, types, and state keys must all be <= 255 bytes
if let Some(CanonicalJsonValue::String(room_id)) = owned_obj.get("room_id") {
if room_id.len() > 255 {
@@ -311,7 +313,7 @@ pub async fn create_hash_and_sign_event(
pdu_json
.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
// Verify that the *full* PDU isn't over 64KiB.
if !pdu_fits(&mut pdu_json.clone()) {
if !pdu_fits(&pdu_json.clone()) {
// feckin huge PDU mate
return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")));
}
+3
View File
@@ -77,6 +77,7 @@ struct Services {
short: Dep<rooms::short::Service>,
state: Dep<rooms::state::Service>,
state_accessor: Dep<rooms::state_accessor::Service>,
state_compressor: Dep<rooms::state_compressor::Service>,
state_cache: Dep<rooms::state_cache::Service>,
sync: Dep<sync::Service>,
threads: Dep<rooms::threads::Service>,
@@ -110,6 +111,8 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
state: args.depend::<rooms::state::Service>("rooms::state"),
state_accessor: args
.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
state_compressor: args
.depend::<rooms::state_compressor::Service>("rooms::state_compressor"),
state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
sync: args.depend::<sync::Service>("sync"),
threads: args.depend::<rooms::threads::Service>("rooms::threads"),