Compare commits

...

51 Commits

Author SHA1 Message Date
timedout 5f1b68c8de fix: list indentation 2026-07-14 04:52:09 +01:00
timedout 72b605710e meta: Release schedule doc 2026-07-14 04:46:22 +01:00
Ginger 0ac0bd93ae fix: Default deserialize for compatibility_mode 2026-07-12 16:28:27 +01:00
Ginger fe9f718c6e fix: Stop using the federation DNS resolver for non-federation requests 2026-07-12 16:11:00 +01:00
nex e333f5a8e5 chore: announce 2026-07-12 04:14:23 +00:00
Ginger b8c64a68d4 chore: Release 2026-07-12 00:09:26 -04:00
Ginger 163f049757 chore: Update changelog 2026-07-12 00:07:20 -04:00
Ginger f48125c37a fix: Remove dependency on aws-lc-rs 2026-07-11 22:48:30 -04:00
timedout 4001b99261 style: Make send_join's docstring more useful 2026-07-12 02:11:50 +00:00
timedout b6e5a106ff fix: Don't unhelpfully panic when encountering an unrecognised room version 2026-07-12 02:11:50 +00:00
timedout e1eea45fd6 style: Fix EOL 2026-07-12 02:11:50 +00:00
timedout 7cb53f41d6 chore: Add newsfrag 2026-07-12 02:11:50 +00:00
timedout fbf81fcdeb fix: Opportunistically re-use room format rules when parsing incoming PDUs 2026-07-12 02:11:50 +00:00
timedout 87030a3a22 fix: Ensure PDU returned by create_hash_and_sign_event is itself signed 2026-07-12 02:11:50 +00:00
timedout 4f509fa113 fix: Convert room summary to federation format when sending invites 2026-07-12 02:11:50 +00:00
timedout aa7ed885c0 fix: Check correct field name when determining event type 2026-07-12 02:11:50 +00:00
timedout d5e6e617d1 fix: Remove redundant banned room server checks
These don't make sense here anyway
2026-07-12 02:11:50 +00:00
timedout 1f7680ad5f feat: Give send_join the invite treatment 2026-07-12 02:11:50 +00:00
timedout 0c37a542d4 feat: Give send_leave the invite treatment 2026-07-12 02:11:50 +00:00
timedout efd07da0d7 feat: Give send_knock the invite treatment 2026-07-12 02:11:50 +00:00
timedout f0590e882a fix: Only run ACL checks when we know we have live state
Prevents a potential bug where we might inadvertently reject valid invites because we have a stale state cache that blocks the sender or even ourselves

Also fixes the banned remote server room check by using the create event instead of room ID
2026-07-12 02:11:50 +00:00
timedout 7225bf25ae feat: Persist create events received during the invite process 2026-07-12 02:11:50 +00:00
timedout e52bb5db69 style: Move invite validation into own functions 2026-07-12 02:11:50 +00:00
timedout 995726923b feat: Enforce new federation invite checks
These were introduced in spec v1.16 however we didn't implement them until now for compatibility.
2026-07-12 02:11:50 +00:00
Ginger 556141b404 fix: Trim whitespace from OIDC client secret file 2026-07-12 02:06:48 +00:00
Henry-Hiles 6858e1b893 fix: fallback rev to empty string
Fixes flake-compat support
2026-07-11 20:13:58 +00:00
Ginger 07e241f616 chore: News fragment 2026-07-11 19:40:25 +00:00
Ginger cfbe590f1c fix: Do not require an access token on 3pid token request route 2026-07-11 19:40:25 +00:00
Ginger d9c2f04d73 fix: Correctly handle preferred_username claim 2026-07-11 19:21:33 +00:00
Ginger ada0b3184b fix: Properly sync newly created rooms 2026-07-11 19:21:19 +00:00
Renovate Bot 1d6ad485f6 chore(deps): update ruma digest to 9b6a2e7 2026-07-11 16:29:54 +00:00
timedout 9d2ce72f73 style: Remove redundant length check for dag build 2026-07-11 17:27:42 +01:00
timedout 13bcb5af13 style: Accept event id reference in pdu_format_check_1 2026-07-11 17:27:42 +01:00
timedout 5868510df1 fix: Remove redundant "correct room ID" filter 2026-07-11 17:27:42 +01:00
timedout c34d1fe76e feat: Move PDU validation out of parsing and into check 1 2026-07-11 17:27:42 +01:00
timedout 838764b0ac style: Remove needless async def for parse_incoming_pdu_with_known_room 2026-07-11 17:27:42 +01:00
timedout b51a34c1c1 fix: Address review feedback 2026-07-11 17:27:42 +01:00
timedout 2fbb779d12 fix: Invert boolean returned by policy server check 2026-07-11 17:27:42 +01:00
timedout e16a62e1ed feat: Add more debug logging to PDU handle funcs
Tracking down weird soft-fails
2026-07-11 17:27:42 +01:00
timedout 64a08262ca perf: Change signature of build_local_dag to remove refmap hacks 2026-07-11 17:27:42 +01:00
timedout 70a3905761 style: Remove duplicate pdu check 1 call in handle_incoming_pdu 2026-07-11 17:27:42 +01:00
timedout cf9179e8f6 fix: Perform checks 1-3 on auth chain PDUs 2026-07-11 17:27:42 +01:00
timedout 6b899b7f80 refactor: Move PDU checks 1, 2, and 3 into pdu_checks.rs 2026-07-11 17:27:42 +01:00
timedout ad57fa06fc style: Unify and document state_before_incoming 2026-07-11 17:27:42 +01:00
timedout ec9efbada5 fix: Don't reject auth events when they're innocent in auth chain response 2026-07-11 17:27:42 +01:00
timedout 318d2b205a style: Move PDU check functions into their own module 2026-07-11 17:27:42 +01:00
timedout 99a5f2a9fb 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-11 17:27:42 +01:00
Renovate Bot ace0d52074 chore(deps): Update rust crate termimad to 0.35.0 2026-07-11 15:35:53 +00:00
Ginger 0b1e24a9ad fix: Add admin command to ensure all appservice puppets are active 2026-07-11 14:55:43 +00:00
Ginger f55207d5c7 fix: Stop appservice users from being wrongly deactivated during migration 2026-07-11 14:55:43 +00:00
Renovate Bot 3f2fae4ca7 chore(deps): update dependency typescript to v7 2026-07-11 05:04:52 +00:00
58 changed files with 2123 additions and 1324 deletions
+15
View File
@@ -1,3 +1,18 @@
# Continuwuity 26.6.1 (2026-07-12)
## Features
- Added enforcement for new federated invite checks and corrected a bunch of related spec compliance issues along the way. Contributed by @nex. (#1952)
## Bugfixes
- Fixed existing accounts failing to link when logging in with OIDC if `prompt_for_localpart` was `false`. (#1942)
- Authentication is no longer required on the `/_matrix/client/v3/account/3pid/email/requestToken` endpoint. (#1953)
- Fixed newly created rooms failing to sync properly in clients using legacy sync.
- Stopped appservice users from being erroneously marked as deactivated during a 26.6 database migration.
- Whitespace will now automatically be trimmed from the start and end of the `global.oauth.oidc.client_secret_file`.
# Continuwuity 26.6.0 (2026-07-10)
## Features
Generated
+29 -29
View File
@@ -826,7 +826,7 @@ dependencies = [
[[package]]
name = "conduwuit"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"aws-lc-rs",
"clap",
@@ -864,7 +864,7 @@ dependencies = [
[[package]]
name = "conduwuit_admin"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"assign",
"clap",
@@ -890,7 +890,7 @@ dependencies = [
[[package]]
name = "conduwuit_api"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"assign",
"async-trait",
@@ -928,7 +928,7 @@ dependencies = [
[[package]]
name = "conduwuit_build_metadata"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"built",
"cargo_metadata",
@@ -936,7 +936,7 @@ dependencies = [
[[package]]
name = "conduwuit_core"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"argon2",
"arrayvec",
@@ -1004,7 +1004,7 @@ dependencies = [
[[package]]
name = "conduwuit_database"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"async-channel",
"conduwuit_core",
@@ -1025,7 +1025,7 @@ dependencies = [
[[package]]
name = "conduwuit_macros"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"cargo_toml",
"itertools 0.15.0",
@@ -1036,7 +1036,7 @@ dependencies = [
[[package]]
name = "conduwuit_router"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"assign",
"axum",
@@ -1074,7 +1074,7 @@ dependencies = [
[[package]]
name = "conduwuit_service"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"askama",
"assign",
@@ -1126,7 +1126,7 @@ dependencies = [
[[package]]
name = "conduwuit_web"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"askama",
"assign",
@@ -3253,9 +3253,9 @@ dependencies = [
[[package]]
name = "minimad"
version = "0.14.0"
version = "0.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df8b688969b16915f3ecadc7829d5b7779dee4977e503f767f34136803d5c06f"
checksum = "de632ee829aec3a874d18a4192eae64a0460b3a45c54ed556b334f6fe5a1d62f"
dependencies = [
"once_cell",
]
@@ -4539,9 +4539,9 @@ checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7"
[[package]]
name = "resolvematrix"
version = "1.1.0"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a574e37f06e219dcfdb14a4f0abf56e806c4551e030aae53802d2c42b29aea33"
checksum = "a68b9735d8c1096d8152d4dfb5a705c0135dd14d181542fd87301f8d3cb4ed03"
dependencies = [
"futures",
"hickory-resolver",
@@ -4601,7 +4601,7 @@ dependencies = [
[[package]]
name = "ruma"
version = "0.16.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"assign",
"js_int",
@@ -4620,7 +4620,7 @@ dependencies = [
[[package]]
name = "ruma-appservice-api"
version = "0.16.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"js_int",
"ruma-common",
@@ -4632,7 +4632,7 @@ dependencies = [
[[package]]
name = "ruma-client-api"
version = "0.24.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"as_variant",
"assign",
@@ -4654,7 +4654,7 @@ dependencies = [
[[package]]
name = "ruma-common"
version = "0.19.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"as_variant",
"base64 0.22.1",
@@ -4687,7 +4687,7 @@ dependencies = [
[[package]]
name = "ruma-events"
version = "0.34.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"as_variant",
"indexmap 2.14.0",
@@ -4708,7 +4708,7 @@ dependencies = [
[[package]]
name = "ruma-federation-api"
version = "0.15.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"bytes",
"headers",
@@ -4731,7 +4731,7 @@ dependencies = [
[[package]]
name = "ruma-identifiers-validation"
version = "0.12.1"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"js_int",
"thiserror 2.0.18",
@@ -4740,7 +4740,7 @@ dependencies = [
[[package]]
name = "ruma-macros"
version = "0.19.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"as_variant",
"cfg-if",
@@ -4756,7 +4756,7 @@ dependencies = [
[[package]]
name = "ruma-push-gateway-api"
version = "0.15.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"js_int",
"ruma-common",
@@ -4768,7 +4768,7 @@ dependencies = [
[[package]]
name = "ruma-signatures"
version = "0.21.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"base64 0.22.1",
"ed25519-dalek",
@@ -4785,7 +4785,7 @@ dependencies = [
[[package]]
name = "ruma-state-res"
version = "0.17.0"
source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
source = "git+https://github.com/ruma/ruma.git?rev=9b6a2e7323649af926e4b5363d87239ee4247f4a#9b6a2e7323649af926e4b5363d87239ee4247f4a"
dependencies = [
"js_int",
"ruma-common",
@@ -4799,7 +4799,7 @@ dependencies = [
[[package]]
name = "ruminuwuity"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"assign",
"ruma",
@@ -5722,9 +5722,9 @@ dependencies = [
[[package]]
name = "termimad"
version = "0.34.1"
version = "0.35.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "889a9370996b74cf46016ce35b96c248a9ac36d69aab1d112b3e09bc33affa49"
checksum = "903906afabf58273a7a90b47e0e60701cbd01731ca72ba88508b38b5539647f4"
dependencies = [
"coolor",
"crokey",
@@ -6972,7 +6972,7 @@ dependencies = [
[[package]]
name = "xtask"
version = "26.6.0"
version = "26.6.1"
dependencies = [
"askama",
"cargo_metadata",
+4 -10
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"
version = "26.6.1"
[workspace.metadata.crane]
name = "conduwuit"
@@ -141,12 +141,6 @@ features = [
version = "0.23.25"
default-features = false
[workspace.dependencies.aws-lc-sys]
version = "0.41.0"
[workspace.dependencies.aws-lc-rs]
version = "1.17.0"
[workspace.dependencies.reqwest]
version = "0.13.2"
default-features = false
@@ -350,7 +344,7 @@ version = "1.1.1"
[workspace.dependencies.ruma]
# version = "0.14.1"
git = "https://github.com/ruma/ruma.git"
rev = "fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
rev = "9b6a2e7323649af926e4b5363d87239ee4247f4a"
features = [
"appservice-api-c",
"client-api",
@@ -502,7 +496,7 @@ version = "0.4.9"
default-features = false
[workspace.dependencies.termimad]
version = "0.34.0"
version = "0.35.0"
default-features = false
[workspace.dependencies.checked_ops]
@@ -570,7 +564,7 @@ features = ["std"]
version = "0.3.0"
[workspace.dependencies.resolvematrix]
version = "1.1.0"
version = "1.2.0"
[workspace.dependencies.serde_urlencoded]
version = "0.7.1"
+1
View File
@@ -0,0 +1 @@
Fixed `global.oauth.compatibility_mode` being required, despite being ignored, when the `[global.oauth.oidc]` config section is provided.
+30 -25
View File
@@ -1,27 +1,32 @@
[
{
"type": "file",
"name": "index",
"label": "Development Guide"
},
{
"type": "file",
"name": "contributing",
"label": "Contributing"
},
{
"type": "file",
"name": "code_style",
"label": "Code Style Guide"
},
{
"type": "file",
"name": "testing",
"label": "Testing"
},
{
"type": "file",
"name": "hot_reload",
"label": "Hot Reloading"
}
{
"type": "file",
"name": "index",
"label": "Development Guide"
},
{
"type": "file",
"name": "contributing",
"label": "Contributing"
},
{
"type": "file",
"name": "code_style",
"label": "Code Style Guide"
},
{
"type": "file",
"name": "testing",
"label": "Testing"
},
{
"type": "file",
"name": "hot_reload",
"label": "Hot Reloading"
},
{
"type": "file",
"name": "releasing",
"label": "Releasing"
}
]
+77
View File
@@ -0,0 +1,77 @@
# Releasing
Continuwuity is rapidly evolving software that needs to keep up with an ecosystem that demands equally faced paced
development. While we generally encourage people follow the `main` branch, not everyone has the time or will to keep up
with even nightly updates. For this reason, we do regular releases.
In order to ensure that releases are stable and predictable, the process defined in this document should be followed.
## Schedule
The release `YY.MM` should be forked from the `main` branch on the first day of the final week of the month.
Immediately after, `YY.MM.0-rc.1` should be tagged and released as a pre-release.
Then, for each day of the final week of the month, if there have been bug fixes since the last release candidate, a new
release candidate should be tagged and released as a pre-release. This means there can be between 1 and 6
release candidates for a given release.
On the last day of the final week of the month, if there are no outstanding serious bugs, the full release should be
tagged and released.
If there ARE serious bugs found at the time the final release is due, the release should be delayed until the
bugs are fixed. Additionally, another release candidate should be tagged for at least 24 hours after the patch is
released, to allow for testing and verification of the fix. Only then can the final release be made.
This means a release timeline MIGHT look like:
- Week 1: Bugfixes from the previous release that missed the cycle
- Week 2: New major features & other changes
- Week 3: New minor features & bugfixing
- Week 4: Release candidates and final release
- Day 1: Fork release branch from `main` and tag `YY.MM.0-rc.1`
- Day 2: If there are bugfixes, tag `YY.MM.0-rc.2`
- Day 3: If there are bugfixes, tag `YY.MM.0-rc.3`
- Day 4: If there are bugfixes, tag `YY.MM.0-rc.4`
- Day 5: If there are bugfixes, tag `YY.MM.0-rc.5`
- Day 6: If there are bugfixes, tag `YY.MM.0-rc.6`
- Day 7: If there are no serious bugs, tag `YY.MM.0` and release.
New changes of any sort may be made to the `main` branch at any point during the month, but they will not be included
in the release until the next cycle.
### Off-schedule releases (patches)
Sometimes, critical bugs may be discovered that require an immediate release.
In this case, a new branch should be forked **from the previous release tag**, incrementing the patch value (e.g. `YY.MM.0` ->` YY.MM.1`)
This will ensure that the critical bug is fixed without introducing any new changes that may potentially be incompatible.
### Security releases
Following the [security policy](../security.mdx), security releases might be made off-cycle, and may even require
creating new releases for older versions (backports). It is imperative that security patches are tested BEFORE being
pushed to the public repository, as once the cat is out of the bag, it doesn't go back in. Consequently,
there is no "pre-release" system for security releases - security patches always go straight to release.
Not all security fixes require a new release. If there are several low-severity security problems that are awaiting
an upstream merge, they should instead wait for the next scheduled release, unless they are critical enough to warrant
an immediate release.
Security patches should NEVER be pushed to the public repository when a release is not planned to be made very soon
after. The shorter the window between the diff going live and the release, the smaller the window for exploit.
### Release candidates
Release candidates are the primary form of pre-release - they imply a feature freeze, which allows maintainers and
contributors to focus on fixing bugs before the ultimate release, which ensures all new features are stable and working
as intended. Simply relying on deployments to run main has proven to be insufficient for filling this role,
so release candidates are a necessary step in the release process.
Release candidates should be announced in the same way any other release is (ping in the announcements room,
announcement sent out via the checker, etc), but should be clearly marked as a pre-release. Extra attention to detail
should be spent ensuring breaking changes are clearly communicated.
### Alphas and Betas
Alphas and betas may be created during any time of the month, and it is recommended to do so after merging a major
feature, especially if the release candidate cycle is not close to starting. This is not expected to be commonplace, so
announcing these releases is up to the discretion of the release manager.
@@ -9,7 +9,7 @@
"id": 15,
"mention_room": true,
"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"
"message": "[Continuwuity 26.6.1](https://forgejo.ellis.link/continuwuation/continuwuity/releases/tag/v26.6.1) has just been released! This release contains a bunch of bugfixes for bugs found initially after 26.6.0's release - please read the changelog before updating!"
}
]
}
+4
View File
@@ -27,3 +27,7 @@ ## `!admin appservices show-appservice-config`
## `!admin appservices list-registered`
List all the currently registered appservices
## `!admin appservices ensure-puppets-active`
Ensure no appservice puppets are marked as deactivated. This is a debug command to fix issues caused by a faulty database migration in Continuwuity 26.6.0
+4
View File
@@ -146,6 +146,10 @@ ## `!admin debug send-test-email`
Send a test email to the invoking admin's email address
## `!admin debug rooms-by-extremity-count`
Lists room IDs by forward extremity count in descending order
## `!admin debug tester`
Developer test stubs
+1 -3
View File
@@ -106,9 +106,7 @@ ## `!admin query resolver`
### `!admin query resolver cache`
Query the destinations or overrides cache
Shows entire cache by default, but can be narrowed to only show overrides with `-o/--overrides true`, or to exclude them with `false`
Query the destinations or overrides cache, depending on the value of the `overrides` flag (default false)
### `!admin query resolver flush-cache`
+2 -2
View File
@@ -43,8 +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;
GIT_COMMIT_HASH = self.rev or self.dirtyRev or "";
GIT_COMMIT_HASH_SHORT = self.shortRev or self.dirtyShortRev or "";
}
// (lib.optionalAttrs (rocksdb != null) {
ROCKSDB_INCLUDE_DIR = "${rocksdb}/include";
+368 -7
View File
@@ -12,7 +12,7 @@
"@rspress/core": "^2.0.0",
"@rspress/plugin-client-redirects": "^2.0.0",
"@rspress/plugin-sitemap": "^2.0.0",
"typescript": "^6.0.0"
"typescript": "^7.0.0"
}
},
"node_modules/@emnapi/core": {
@@ -735,6 +735,346 @@
"dev": true,
"license": "MIT"
},
"node_modules/@typescript/typescript-aix-ppc64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-aix-ppc64/-/typescript-aix-ppc64-7.0.2.tgz",
"integrity": "sha512-MTKKkWB7p/0E9xi1d1tHtZ5PiLkGEMIq88pK2CubZjOsLtYTLqhgIgi6zepFa+9GHZ6h05NMCkQxGKiPXMxXtQ==",
"cpu": [
"ppc64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"aix"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-darwin-arm64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-darwin-arm64/-/typescript-darwin-arm64-7.0.2.tgz",
"integrity": "sha512-gowzar9MwS/aRWp6f3a4KUqzRjAZjOsmGNCM6LcTgXum+dBfgsBVMN+AgvOCCbguXyick6LJhpBszxMebJ8syA==",
"cpu": [
"arm64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"darwin"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-darwin-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-darwin-x64/-/typescript-darwin-x64-7.0.2.tgz",
"integrity": "sha512-SZ9xZInqApNlNGc9s0W1VSsktYSOe9cFqNOIqmN1Gs8SmkjKZYFt017G4VwPxASInODuAdbTW7sXiFUf893RgA==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"darwin"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-freebsd-arm64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-freebsd-arm64/-/typescript-freebsd-arm64-7.0.2.tgz",
"integrity": "sha512-W5NH4y/J0plIIS5b2xvTEkU7JFxyqdMAOgf+Ilhl0vHQXKO5dZoxd+C/jEtq56c4F3wk71RB4BMRQ2XdI+bwYQ==",
"cpu": [
"arm64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"freebsd"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-freebsd-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-freebsd-x64/-/typescript-freebsd-x64-7.0.2.tgz",
"integrity": "sha512-UMGDx5sTpzNw3WiPebH7l90IWfJggEd+egHt/q6p7/Cm3zqoV7VxkGXt+3DxPIw8CcmvAB0j3sVVfbhX+M4Tpw==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"freebsd"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-arm": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-arm/-/typescript-linux-arm-7.0.2.tgz",
"integrity": "sha512-gffT3xPz9sR7j/YJExkyPntrI0P2EP9XbOyWzth2/Gs0RstK+90RBcO0ncXoXy/beYll1SXw846Nf2zdnEz0QQ==",
"cpu": [
"arm"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-arm64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-arm64/-/typescript-linux-arm64-7.0.2.tgz",
"integrity": "sha512-Qh4eU4/y3yDjnfjjyPYihMj5/ODIlmt+Bzu17OI+fiSRDW57QmU5SiN63exPRNJPKUzcc1INa1NXdrJ+MqHjUQ==",
"cpu": [
"arm64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-loong64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-loong64/-/typescript-linux-loong64-7.0.2.tgz",
"integrity": "sha512-uEHck9i8hoAzXPiYRib1O7miOnz23SxIeVl6F4LXox+qov1K35jHcEW6VHKvZI+pyvl7fZEP4MCU5LYvIq1GuQ==",
"cpu": [
"loong64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-mips64el": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-mips64el/-/typescript-linux-mips64el-7.0.2.tgz",
"integrity": "sha512-R4KvAMnE43W5Qeqb0Ly56O3mWMWIAgsMyz36DCaycd5nbg/9kzm0liw3JocfRqyJY0KPmzFjbswozXyW0DnIYA==",
"cpu": [
"mips64el"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-ppc64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-ppc64/-/typescript-linux-ppc64-7.0.2.tgz",
"integrity": "sha512-DORx5b3sd/4S7eayxm4FQv+A7CrkUIGRaHiwI8oiHTAI1fAPWhF4J0vAlkC8biAlHSVVwxMQ3tjZ2/DVbnQiiA==",
"cpu": [
"ppc64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-riscv64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-riscv64/-/typescript-linux-riscv64-7.0.2.tgz",
"integrity": "sha512-wf0jqEDOjrPRnKwYRyyJDRo11KMbvMFrU+q4zqKyChODBzvlkbhNQfKvLxQCcwTpdDaXSHZTVuh0JoCrKCUMHQ==",
"cpu": [
"riscv64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-s390x": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-s390x/-/typescript-linux-s390x-7.0.2.tgz",
"integrity": "sha512-IkwJc3L7yhytWd/ewjyxNDfOmswCm9GWMJT/ue/dU4aZNbwZeYAetq42VyLmsmSjvoX7z74X6ZaYCtzAr0EuGw==",
"cpu": [
"s390x"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-linux-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-linux-x64/-/typescript-linux-x64-7.0.2.tgz",
"integrity": "sha512-EYdf2cNg7rgCWJnxCdJ+F3V39O8ihb37eHAu1LK8oAFizgTQbPOK7zHHXbPt8rX24COqODXeI3sIf0fCXG7H/A==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"linux"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-netbsd-arm64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-netbsd-arm64/-/typescript-netbsd-arm64-7.0.2.tgz",
"integrity": "sha512-+polYF4MF04aPpO5FTkHran9yUQDSXqy5GiSDKpsll5jy3l3+g9QLhpf39T+ePtefhXLOGrLl0QIjkQP6VnelA==",
"cpu": [
"arm64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"netbsd"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-netbsd-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-netbsd-x64/-/typescript-netbsd-x64-7.0.2.tgz",
"integrity": "sha512-8YIT0EHM/3dq10ZOVF/A7pc/YSMtbcecct4rWtexrnSCHOPcpC2KTLXfTCR6vDpnSiY12heNb1GiN/wu+T/FyA==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"netbsd"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-openbsd-arm64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-openbsd-arm64/-/typescript-openbsd-arm64-7.0.2.tgz",
"integrity": "sha512-APT8+ClYnuYm1u9+kgGXoMj2VzWzcymwh2gNSQVySHfkRDGOTVkoWLjCmOQSaO+PoqQ57B0flRp9SA+7GnnkzQ==",
"cpu": [
"arm64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"openbsd"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-openbsd-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-openbsd-x64/-/typescript-openbsd-x64-7.0.2.tgz",
"integrity": "sha512-yX7s+Q0Dln0Dt9tEzZsAjXXR/+ytBM7AlglaqyeMPxQszJ1JhlJdZ6jLA+IzldHtflX81em7lDao1xXu+aRRkg==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"openbsd"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-sunos-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-sunos-x64/-/typescript-sunos-x64-7.0.2.tgz",
"integrity": "sha512-dLJDGaLZ1D4HPQn62u1n8mBDkJREwMsAkCdkwd4Ieqw+x3TUyTsqY0YiBCtE6H6OzzgGk3iuZ3vFWRS+E8/d1g==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"sunos"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-win32-arm64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-win32-arm64/-/typescript-win32-arm64-7.0.2.tgz",
"integrity": "sha512-Gyl1Vy6OsWesLzmq+EP0Fb7b4Nid5232AvcA2SFcdYreldpNtYFFofPjnt62y9hQy7VTaZp65ICJjuAQRaVcIQ==",
"cpu": [
"arm64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"win32"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@typescript/typescript-win32-x64": {
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/@typescript/typescript-win32-x64/-/typescript-win32-x64-7.0.2.tgz",
"integrity": "sha512-0BQ3HkAHHlKLSp1qRvf3SUhGpGsDuhB/jgFw75guyqbxJqEaS0Cw/VFO8i2nHglJUzQCRtMMR/IBAKE3ETMC4g==",
"cpu": [
"x64"
],
"dev": true,
"license": "Apache-2.0",
"optional": true,
"os": [
"win32"
],
"engines": {
"node": ">=16.20.0"
}
},
"node_modules/@ungap/structured-clone": {
"version": "1.3.1",
"resolved": "https://registry.npmjs.org/@ungap/structured-clone/-/structured-clone-1.3.1.tgz",
@@ -3361,17 +3701,38 @@
"license": "0BSD"
},
"node_modules/typescript": {
"version": "6.0.3",
"resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz",
"integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==",
"version": "7.0.2",
"resolved": "https://registry.npmjs.org/typescript/-/typescript-7.0.2.tgz",
"integrity": "sha512-8FYau96o3NKOhbjKi/qNvG/W5jhzxkbdm5sj9AbZ/5T5sWqn3hJgLfGx27sRKZWTvyzCP8dLRBTf5tBTSRVUNA==",
"dev": true,
"license": "Apache-2.0",
"bin": {
"tsc": "bin/tsc",
"tsserver": "bin/tsserver"
"tsc": "bin/tsc"
},
"engines": {
"node": ">=14.17"
"node": ">=16.20.0"
},
"optionalDependencies": {
"@typescript/typescript-aix-ppc64": "7.0.2",
"@typescript/typescript-darwin-arm64": "7.0.2",
"@typescript/typescript-darwin-x64": "7.0.2",
"@typescript/typescript-freebsd-arm64": "7.0.2",
"@typescript/typescript-freebsd-x64": "7.0.2",
"@typescript/typescript-linux-arm": "7.0.2",
"@typescript/typescript-linux-arm64": "7.0.2",
"@typescript/typescript-linux-loong64": "7.0.2",
"@typescript/typescript-linux-mips64el": "7.0.2",
"@typescript/typescript-linux-ppc64": "7.0.2",
"@typescript/typescript-linux-riscv64": "7.0.2",
"@typescript/typescript-linux-s390x": "7.0.2",
"@typescript/typescript-linux-x64": "7.0.2",
"@typescript/typescript-netbsd-arm64": "7.0.2",
"@typescript/typescript-netbsd-x64": "7.0.2",
"@typescript/typescript-openbsd-arm64": "7.0.2",
"@typescript/typescript-openbsd-x64": "7.0.2",
"@typescript/typescript-sunos-x64": "7.0.2",
"@typescript/typescript-win32-arm64": "7.0.2",
"@typescript/typescript-win32-x64": "7.0.2"
}
},
"node_modules/unhead": {
+1 -1
View File
@@ -25,6 +25,6 @@
"@rspress/core": "^2.0.0",
"@rspress/plugin-client-redirects": "^2.0.0",
"@rspress/plugin-sitemap": "^2.0.0",
"typescript": "^6.0.0"
"typescript": "^7.0.0"
}
}
+25
View File
@@ -73,4 +73,29 @@ pub(super) async fn list_registered(&self) -> Result {
})
.await
}
pub(super) async fn ensure_puppets_active(&self) -> Result {
self.services
.users
.stream_local_users()
.filter_map(async |user_id| {
if self.services.appservice.is_user_id(&user_id).await {
Some(user_id)
} else {
None
}
})
.for_each(async |user_id| {
self.services
.users
.convert_to_shadow_account(&user_id)
.await
.expect("should be able to convert to shadow");
})
.await;
write!(self, "All appservice puppets have been marked as active.").await?;
Ok(())
}
}
+6
View File
@@ -37,4 +37,10 @@ pub enum AppserviceCommand {
/// List all the currently registered appservices
#[clap(alias("list"))]
ListRegistered,
/// Ensure no appservice puppets are marked as deactivated.
/// This is a debug command to fix issues caused by a faulty database
/// migration in Continuwuity 26.6.0.
#[clap(hide = true)]
EnsurePuppetsActive,
}
+3 -3
View File
@@ -486,7 +486,7 @@ pub(super) async fn get_remote_pdu(
.services
.rooms
.event_handler
.parse_incoming_pdu(&response.pdu)
.parse_incoming_pdu(&response.pdu, None)
.boxed()
.await;
@@ -832,7 +832,7 @@ pub(super) async fn force_set_room_state_from_server(
.services
.rooms
.event_handler
.parse_incoming_pdu(&pdu)
.parse_incoming_pdu(&pdu, Some(&room_version_rules))
.await
{
| Ok(t) => t,
@@ -1014,7 +1014,7 @@ pub(super) async fn resolve_true_destination(
let resolver: &MatrixResolver = if no_cache {
&MatrixResolverBuilder::new()
.dangerous_tls_accept_invalid_certs(self.services.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(self.services.client.default.clone())
.http_client(self.services.client.dns.clone())
.build()?
} else {
&self.services.client.matrix_resolver
+1 -1
View File
@@ -54,7 +54,7 @@ pub(super) async fn fetch_support_well_known(&self, server_name: OwnedServerName
let response = self
.services
.client
.default
.external_resource
.get(format!("https://{server_name}/.well-known/matrix/support"))
.send()
.await?;
+2 -4
View File
@@ -57,13 +57,11 @@ pub(crate) async fn request_3pid_management_token_via_email_route(
State(services): State<crate::State>,
body: Ruma<request_3pid_management_token_via_email::v3::Request>,
) -> Result<request_3pid_management_token_via_email::v3::Response> {
// Authentication for this endpoint is technically optional,
// but we require the user to be logged in
let sender_user = body
.identity
.as_ref()
.map(ClientIdentity::expect_sender_user)
.ok_or_else(|| err!(Request(MissingToken("Missing access token."))))??;
.transpose()?;
if !services.threepid.email_requirement().may_change() {
return Err!(Request(Forbidden("You may not change your email address.")));
@@ -88,7 +86,7 @@ pub(crate) async fn request_3pid_management_token_via_email_route(
Mailbox::new(None, email),
|verification_link| messages::ChangeEmail {
server_name: services.config.server_name.as_str(),
user_id: Some(sender_user),
user_id: sender_user,
verification_link,
},
&body.client_secret,
+4 -3
View File
@@ -3,7 +3,7 @@
use conduwuit::{
Err, Result, debug_error, err, info,
matrix::{event::gen_event_id_canonical_json, pdu::PartialPdu},
warn,
trace, warn,
};
use futures::FutureExt;
use ruma::{
@@ -166,7 +166,7 @@ pub(crate) async fn invite_helper(
let invite_room_state = services
.rooms
.state
.summary_stripped(&pdu, room_id, recipient_user)
.summary_stripped(&pdu, room_id, recipient_user, true)
.await;
drop(state_lock);
@@ -193,6 +193,7 @@ pub(crate) async fn invite_helper(
.await
.ok();
trace!(?request, "Sending invite");
let response = services
.sending
.send_federation_request(recipient_user.server_name(), request)
@@ -221,7 +222,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(|| {
+15 -10
View File
@@ -381,16 +381,21 @@ async fn fetch_shortstatehashes(
| Some((_, pdu_after_last_sync_end)) => {
trace!(?pdu_after_last_sync_end.event_id, "pdu at last sync end");
Some(
services
.rooms
.state_accessor
.pdu_shortstatehash(&pdu_after_last_sync_end.event_id)
.await
.map_err(|err| {
err!("Last sync end PDU has no shortstatehash: {err}")
}),
)
if pdu_after_last_sync_end.kind == RoomCreate {
// Room create events have no previous state
None
} else {
Some(
services
.rooms
.state_accessor
.pdu_shortstatehash(&pdu_after_last_sync_end.event_id)
.await
.map_err(|err| {
err!("Last sync end PDU has no shortstatehash: {err}")
}),
)
}
},
| None => {
// No events have been sent since the last sync, or we just joined this room
+294 -110
View File
@@ -1,23 +1,29 @@
use std::collections::{HashMap, hash_map::Entry};
use axum::extract::State;
use axum_client_ip::ClientIp;
use base64::{Engine as _, engine::general_purpose};
use conduwuit::{
Err, Error, PduEvent, Result, err, error,
matrix::{Event, event::gen_event_id},
utils::{self, hash::sha256},
Err, Error, EventTypeExt, PduEvent, Result, debug, err, error,
matrix::{Event, StateKey},
result::FlatOk,
state_res, trace,
utils::hash::sha256,
warn,
};
use ruma::{
CanonicalJsonValue, UserId,
CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId, OwnedUserId, ServerName,
UserId,
api::{
error::{ErrorKind, IncompatibleRoomVersionErrorData},
federation::membership::{RawStrippedState, create_invite},
},
events::room::member::{MembershipState, RoomMemberEventContent},
serde::JsonObject,
events::{StateEventType, room::member::MembershipState},
room_version_rules::RoomVersionRules,
};
use serde::Deserialize;
use crate::Ruma;
use crate::{Ruma, server::utils::validate_any_membership_event};
/// # `PUT /_matrix/federation/v2/invite/{roomId}/{eventId}`
///
@@ -28,30 +34,16 @@ pub(crate) async fn create_invite_route(
ClientIp(client): ClientIp,
body: Ruma<create_invite::v2::Request>,
) -> Result<create_invite::v2::Response> {
// ACL check origin
services
.rooms
.event_handler
.acl_check(&body.identity, &body.room_id)
.await?;
if !services.server.supported_room_version(&body.room_version) {
return Err(Error::BadRequest(
ErrorKind::IncompatibleRoomVersion(IncompatibleRoomVersionErrorData::new(
body.room_version.clone(),
)),
"Server does not support this room version.",
"This server does not support that room version",
));
}
let room_version_rules = body.room_version.rules().unwrap();
if let Some(server) = body.room_id.server_name() {
if services.moderation.is_remote_server_forbidden(server) {
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
}
if services
.moderation
.is_remote_server_forbidden(&body.identity)
@@ -61,90 +53,61 @@ pub(crate) async fn create_invite_route(
body.identity, body.room_id
);
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
return Err!(Request(Forbidden("Federation denied with {}", body.identity)));
}
let mut signed_event = utils::to_canonical_object(&body.event)
.map_err(|_| err!(Request(InvalidParam("Invite event is invalid."))))?;
// Ensure this is a membership event
if signed_event
.get("type")
.expect("event must have a type")
.as_str()
.expect("type must be a string")
!= "m.room.member"
{
return Err!(Request(BadJson(
"Not allowed to send non-membership event to invite endpoint."
)));
}
let content: RoomMemberEventContent = serde_json::from_value(
signed_event
.get("content")
.ok_or_else(|| err!(Request(BadJson("Event missing content property"))))?
.clone()
.into(),
// First, validate the invite room state, so we can compare with the create
// event.
debug!(
event_id=%body.event_id,
room_id=%body.room_id,
room_version=?body.room_version,
via=?body.via,
"Validating invite room state for invite request"
);
let (create_event_id, state) = validate_invite_state(
&services,
&body.invite_room_state,
&room_version_rules,
body.room_id.clone(),
)
.map_err(|e| err!(Request(BadJson(warn!("Event content is empty or invalid: {e}")))))?;
.await?;
let create_event_json = state
.get(&StateEventType::RoomCreate.with_state_key(""))
.expect("must have create event in invite state by this point");
// Ensure this is an invite membership event
if content.membership != MembershipState::Invite {
return Err!(Request(BadJson(
"Not allowed to send a non-invite membership event to invite endpoint."
)));
}
// Ensure the sending user isn't a lying bozo
let sender_user = signed_event
// We can now perform the banned remote server check with the create event.
// N.B. this checks the sender field, which is technically incorrect for rooms
// v10 and below. This usually isn't the case though so sue me
let creator = create_event_json
.get("sender")
.and_then(|v| v.as_str())
.map(UserId::parse)
.and_then(Result::ok)
.ok_or_else(|| err!(Request(InvalidParam("Invalid sender property"))))?;
if sender_user.server_name() != body.identity {
return Err!(Request(Forbidden("Sender's server does not match the origin server.",)));
}
// Ensure the target user belongs to this server
let recipient_user = signed_event
.get("state_key")
.and_then(|v| v.as_str())
.map(UserId::parse)
.and_then(Result::ok)
.ok_or_else(|| err!(Request(InvalidParam("Invalid state_key property"))))?;
if !services
.globals
.server_is_ours(recipient_user.server_name())
.flat_ok()
.expect("must have valid sender in create event");
if services
.moderation
.is_remote_server_forbidden(creator.server_name())
{
return Err!(Request(InvalidParam("User does not belong to this homeserver.")));
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
// Make sure we're not ACL'ed from their room.
services
.rooms
.event_handler
.acl_check(recipient_user.server_name(), &body.room_id)
.await?;
services
.server_keys
.hash_and_sign_event(&mut signed_event, &room_version_rules)
.map_err(|e| err!(Request(InvalidParam("Failed to sign event: {e}"))))?;
// Generate event id
let event_id = gen_event_id(&signed_event, &room_version_rules)?;
// Add event_id back
signed_event.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.to_string()));
// And then we can validate the member event itself
let (mut signed_event, sender_user, recipient_user) = validate_invite_membership_event(
&services,
&body.event,
&room_version_rules,
&body.identity,
create_event_id.clone(),
body.room_id.clone(),
body.event_id.clone(),
)
.await?;
if services.rooms.metadata.is_banned(&body.room_id).await
&& !services.users.is_admin(&recipient_user).await
{
return Err!(Request(Forbidden("This room is banned on this homeserver.")));
return Err!(Request(Forbidden("That room is banned on this homeserver.")));
}
if services.config.block_non_admin_invites && !services.users.is_admin(&recipient_user).await
@@ -161,30 +124,57 @@ pub(crate) async fn create_invite_route(
return Err!(Request(Forbidden("Invite rejected by antispam service.")));
}
// If we're already in the room, ensure that neither the origin nor ourselves
// are ACL'd.
let resident = services
.rooms
.state_cache
.server_in_room(services.globals.server_name(), &body.room_id)
.await;
if resident {
services
.rooms
.event_handler
.acl_check(&body.identity, &body.room_id)
.await?;
services
.rooms
.event_handler
.acl_check(recipient_user.server_name(), &body.room_id)
.await
.map_err(|_| err!(Request(Forbidden("This server is ACL'd from that room"))))?;
}
services
.server_keys
.hash_and_sign_event(&mut signed_event, &room_version_rules)
.map_err(|e| err!(Request(InvalidParam("Failed to sign event: {e}"))))?;
// Add event_id back
signed_event
.insert("event_id".to_owned(), CanonicalJsonValue::String(body.event_id.to_string()));
let mut invite_state = body.invite_room_state.clone();
let mut event: JsonObject = serde_json::from_str(body.event.get())
.map_err(|e| err!(Request(BadJson("Invalid invite event PDU: {e}"))))?;
event.insert("event_id".to_owned(), "$placeholder".into());
let pdu: PduEvent = serde_json::from_value(event.into())
.map_err(|e| err!(Request(BadJson("Invalid invite event PDU: {e}"))))?;
invite_state.push(RawStrippedState::Pdu(
serde_json::value::to_raw_value(&pdu).expect("PDU was just created, it must be valid"),
));
let pdu = PduEvent::from_id_val(&body.event_id, signed_event.clone())
.expect("must be able to create PDU object");
invite_state.push(RawStrippedState::Pdu(serde_json::value::to_raw_value(&signed_event)?));
// If we are active in the room, the remote server will notify us about the
// join/invite through /send. If we are not in the room, we need to manually
// record the invited state for client /sync through update_membership(), and
// send the invite PDU to the relevant appservices.
if !services
.rooms
.state_cache
.server_in_room(services.globals.server_name(), &body.room_id)
.await
{
if !resident {
// We will start by recording the room's create event as an outlier.
// This will allow us to recognise it later in case the sender revokes the
// invite over federation later. We could store more state from the invite
// request, but we will get that during send_join anyway.
// This is safe to just add directly as an outlier as we already auth checked it
// during validation.
services
.rooms
.outlier
.add_pdu_outlier(&create_event_id, create_event_json);
services
.rooms
.state_cache
@@ -240,3 +230,197 @@ pub(crate) async fn create_invite_route(
.await,
))
}
/// Validates the *membership event* in the invite request, per the steps listed
/// under the invite endpoint's [spec].
///
/// Returns the validated JSON body, sender user ID, and recipient user ID.
///
/// Since this function performs a PDU format check, the create event must be
/// known ahead of time. This implies validating the invite state before the
/// invite event itself.
///
/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#put_matrixfederationv2inviteroomideventid
async fn validate_invite_membership_event(
services: &crate::State,
body: &serde_json::value::RawValue,
room_version_rules: &RoomVersionRules,
origin: &ServerName,
create_event_id: OwnedEventId,
room_id: OwnedRoomId,
event_id: OwnedEventId,
) -> Result<(CanonicalJsonObject, OwnedUserId, OwnedUserId)> {
trace!(?body, "Invite membership event");
let (pdu, target_membership, sender_user, recipient_user) = validate_any_membership_event(
services,
body,
room_version_rules,
create_event_id,
room_id,
event_id,
)
.await?;
// Ensure the sender belongs to the remote that is sending the invite
if sender_user.server_name() != origin {
return Err!(Request(Forbidden("Sender belongs to a different server")));
}
// Ensure the target user belongs to this server
if !services
.globals
.server_is_ours(recipient_user.server_name())
{
return Err!(Request(InvalidParam("Recipient does not belong to this homeserver")));
}
if target_membership != MembershipState::Invite {
return Err!(Request(BadJson("Invalid membership (expected `invite`)")));
}
Ok((pdu, sender_user, recipient_user))
}
/// Validates the *invite state* of an invite request, per the steps listed
/// under the endpoint's [spec].
///
/// Returns the create event's event ID, and the partial state map.
///
/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#put_matrixfederationv2inviteroomideventid
async fn validate_invite_state(
services: &crate::State,
invite_state: &[RawStrippedState],
room_version_rules: &RoomVersionRules,
room_id: OwnedRoomId,
) -> Result<(OwnedEventId, HashMap<(StateEventType, StateKey), CanonicalJsonObject>)> {
trace!(?invite_state, "Raw invite state");
let mut invite_state_map: HashMap<(StateEventType, StateKey), _> =
HashMap::with_capacity(invite_state.len());
let mut create_event_id: Option<OwnedEventId> = None;
for (idx, invite_state_event) in invite_state.iter().cloned().enumerate() {
trace!(%idx, ?invite_state_event, "Invite state event");
// Stripped state hasn't been sent over federation since v1.16.
let RawStrippedState::Pdu(raw_pdu) = invite_state_event else {
debug!(%idx, "Invite state event is not a PDU");
return Err!(Request(InvalidParam(
"PDU in invite state (index {idx}) violates the room event format"
)));
};
let (state_event_room_id, state_event_id, state_event_json) = services
.rooms
.event_handler
.parse_incoming_pdu(&raw_pdu, Some(room_version_rules))
.await
.map_err(|e| {
err!(Request(InvalidParam(debug_warn!("Invalid PDU in invite state: {e}"))))
})?;
if state_event_room_id != room_id {
return Err!(Request(InvalidParam(debug_warn!(
%state_event_room_id,
%room_id,
"PDU in invite state ({state_event_id}) belongs to the wrong room"
))));
}
services
.server_keys
.verify_event(&state_event_json, room_version_rules)
.await
.map_err(|e| {
err!(Request(InvalidParam("Signature verification failed on invite event: {e}")))
})?;
let Some(state_key) = state_event_json.get("state_key").and_then(|k| k.as_str()) else {
return Err!(Request(InvalidParam(debug_info!(
"PDU in invite state ({state_event_id}) is not a state event"
))));
};
let Some(event_type) = state_event_json.get("type").and_then(|k| k.as_str()) else {
return Err!(Request(InvalidParam(debug_warn!(
"PDU in invite state ({state_event_id}) is not an event?"
))));
};
let key = StateEventType::from(event_type).with_state_key(state_key);
match invite_state_map.entry(key) {
| Entry::Occupied(entry) =>
return Err!(Request(InvalidParam(
"Duplicate state events in invite state for state key: {:?}",
entry.key(),
))),
| Entry::Vacant(entry) => {
if entry.key().0 == StateEventType::RoomCreate {
// Ensure this is a legal create event.
let pdu_event =
PduEvent::from_id_val(&state_event_id, state_event_json.clone())
.expect("must be able to create pdu event from event json");
debug!("Validating discovered create event in invite room state");
validate_invite_create_event(&pdu_event, room_version_rules).await?;
create_event_id = Some(state_event_id);
}
entry.insert(state_event_json);
},
}
}
let Some(create_event_id) = create_event_id else {
return Err!(Request(InvalidParam(debug_warn!(
parsed_state=?invite_state_map,
"Invite state does not contain the m.room.create event"
))));
};
invite_state_map.iter().try_for_each(|(key, event_json)| {
service::rooms::event_handler::Service::pdu_format_check_1(
event_json,
room_version_rules,
&create_event_id,
)
.map_err(|e| {
err!(Request(InvalidParam(
"PDU in invite state for {key:?} violates the room event format: {e}"
)))
})
})?;
Ok((create_event_id, invite_state_map))
}
#[derive(Deserialize)]
struct MFederate {
#[serde(rename = "m.federate")]
mfederate: Option<bool>,
}
/// Validates that a create event is suitable for the invite, namely:
///
/// 1. It passes auth checks (aka is valid)
/// 2. The room is federated (there's no point persisting unfederated rooms)
async fn validate_invite_create_event(
pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
) -> Result {
if !state_res::auth_check(
room_version_rules,
pdu,
None,
|_, _| async {
unreachable!("No state should be fetched when processing a lone create event");
},
pdu,
)
.await
.unwrap_or_default()
{
return Err!(Request(InvalidParam("m.room.create event fails auth check")));
}
let can_federate = pdu.get_content::<MFederate>()?.mfederate;
if !can_federate.unwrap_or(true) {
return Err!(Request(InvalidParam(
"Cannot receive invites to a room with m.federate=false"
)));
}
Ok(())
}
-8
View File
@@ -75,14 +75,6 @@ pub(crate) async fn create_join_event_template_route(
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
if let Some(server) = body.room_id.server_name() {
if services.moderation.is_remote_server_forbidden(server) {
return Err!(Request(Forbidden(warn!(
"Room ID server name {server} is banned on this homeserver."
))));
}
}
let room_version = services.rooms.state.get_room_version(&body.room_id).await?;
let room_version_rules = room_version.rules().unwrap();
if !body.ver.contains(&room_version) {
-6
View File
@@ -58,12 +58,6 @@ pub(crate) async fn create_knock_event_template_route(
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
if let Some(server) = body.room_id.server_name() {
if services.moderation.is_remote_server_forbidden(server) {
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
}
let room_version = services.rooms.state.get_room_version(&body.room_id).await?;
let room_version_rules = room_version.rules().unwrap();
+1
View File
@@ -27,6 +27,7 @@ pub(crate) async fn create_leave_event_template_route(
{
info!(
origin = body.identity.as_str(),
room_id = %body.room_id,
"Refusing to serve make_leave for room we aren't participating in"
);
return Err!(Request(NotFound("This server is not participating in that room.")));
+11 -19
View File
@@ -150,7 +150,7 @@ async fn process_inbound_transaction(
.pdus
.iter()
.stream()
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu))
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu, None))
.inspect_err(|e| warn!("Could not parse incoming PDU: {e}"))
.ready_filter_map(Result::ok);
@@ -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(|_| ());
+67 -115
View File
@@ -2,34 +2,36 @@
use axum::extract::State;
use conduwuit::{
Err, Event, Result, at, debug, err, info,
matrix::event::gen_event_id_canonical_json,
trace,
Err, Event, Result, at, debug, err, info, trace,
utils::stream::{BroadbandExt, IterStream, TryBroadbandExt},
warn,
};
use conduwuit_service::Services;
use futures::{FutureExt, StreamExt, TryStreamExt};
use ruma::{
CanonicalJsonValue, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, ServerName,
CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName, UserId,
api::federation::membership::create_join_event,
assign,
events::{
StateEventType, TimelineEventType,
TimelineEventType,
room::member::{MembershipState, RoomMemberEventContent},
},
room_version_rules::RoomVersionRules,
};
use serde_json::value::{RawValue as RawJsonValue, to_raw_value};
use serde_json::value::to_raw_value;
use crate::Ruma;
use crate::{Ruma, server::utils::validate_any_membership_event};
/// helper method for /send_join v1 and v2
/// Creates a join membership event for the target, returning a computed
/// response with room state, auth chains, etc.
#[tracing::instrument(skip(services, pdu, omit_members), fields(room_id = room_id.as_str(), origin = origin.as_str()), level = "info")]
async fn create_join_event(
services: &Services,
origin: &ServerName,
room_id: &RoomId,
pdu: &RawJsonValue,
event_id: &EventId,
room_version_rules: &RoomVersionRules,
mut pdu: CanonicalJsonObject,
omit_members: bool,
) -> Result<create_join_event::v2::RoomState> {
if !services.rooms.metadata.exists(room_id).await {
@@ -43,6 +45,7 @@ async fn create_join_event(
{
info!(
origin = origin.as_str(),
room_id = %room_id,
"Refusing to serve send_join for room we aren't participating in"
);
return Err!(Request(NotFound("This server is not participating in that room.")));
@@ -64,94 +67,25 @@ async fn create_join_event(
.await
.map_err(|e| err!(Request(NotFound(error!("Room has no state: {e}")))))?;
// We do not add the event_id field to the pdu here because of signature and
// hashes checks
trace!("Getting room version");
let room_version = services.rooms.state.get_room_version(room_id).await?;
let room_version_rules = room_version.rules().unwrap();
trace!("Generating event ID and converting to canonical json");
let Ok((event_id, mut value)) = gen_event_id_canonical_json(pdu, &room_version_rules) else {
// Event could not be converted to canonical json
return Err!(Request(BadJson("Could not convert event to canonical json.")));
};
let event_room_id: OwnedRoomId = serde_json::from_value(
value
.get("room_id")
.ok_or_else(|| err!(Request(BadJson("Event missing room_id property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("room_id field is not a valid room ID: {e}")))))?;
if event_room_id != room_id {
return Err!(Request(BadJson("Event room_id does not match request path room ID.")));
}
let event_type: StateEventType = serde_json::from_value(
value
.get("type")
.ok_or_else(|| err!(Request(BadJson("Event missing type property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("Event has invalid state event type: {e}")))))?;
if event_type != StateEventType::RoomMember {
return Err!(Request(BadJson(
"Not allowed to send non-membership state event to join endpoint."
)));
}
let sender = pdu
.get("sender")
.and_then(|v| v.as_str())
.map(UserId::parse)
.and_then(Result::ok)
.expect("sender was already validated");
let content: RoomMemberEventContent = serde_json::from_value(
value
.get("content")
pdu.get("content")
.ok_or_else(|| err!(Request(BadJson("Event missing content property"))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("Event content is empty or invalid: {e}")))))?;
if content.membership != MembershipState::Join {
return Err!(Request(BadJson(
"Not allowed to send a non-join membership event to join endpoint."
)));
}
let sender: OwnedUserId = serde_json::from_value(
value
.get("sender")
.ok_or_else(|| err!(Request(BadJson("Event missing sender property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("sender property is not a valid user ID: {e}")))))?;
// check if origin server is trying to send for another server
if sender.server_name() != origin {
return Err!(Request(Forbidden("Not allowed to join on behalf of another server.")));
}
let state_key: OwnedUserId = serde_json::from_value(
value
.get("state_key")
.ok_or_else(|| err!(Request(BadJson("Event missing state_key property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("State key is not a valid user ID: {e}")))))?;
if state_key != sender {
return Err!(Request(BadJson("State key does not match sender user.")));
}
if let Some(authorising_user) = content.join_authorized_via_users_server {
use ruma::RoomVersionId::*;
if matches!(room_version, V1 | V2 | V3 | V4 | V5 | V6 | V7) {
if !room_version_rules.authorization.restricted_join_rule {
return Err!(Request(InvalidParam(
"Room version {room_version} does not support restricted rooms but \
"Room version does not support restricted rooms but \
join_authorised_via_users_server ({authorising_user}) was found in the event."
)));
}
@@ -174,8 +108,7 @@ async fn create_join_event(
they cannot authorise your join."
)));
}
if !super::user_can_perform_restricted_join(services, &state_key, room_id).await? {
if !super::user_can_perform_restricted_join(services, &sender, room_id).await? {
return Err!(Request(UnableToAuthorizeJoin(
"Joining user did not pass restricted room's rules."
)));
@@ -183,7 +116,7 @@ async fn create_join_event(
services
.server_keys
.hash_and_sign_event(&mut value, &room_version_rules)
.hash_and_sign_event(&mut pdu, room_version_rules)
.map_err(|e| {
err!(Request(InvalidParam(warn!("Failed to sign send_join event: {e}"))))
})?;
@@ -200,7 +133,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, pdu.clone(), false)
.boxed()
.await?
.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;
@@ -220,14 +153,12 @@ async fn create_join_event(
.iter()
.try_stream()
.broad_filter_map(|event_id| async move {
if omit_members {
if let Ok(e) = event_id.as_ref() {
let pdu = services.rooms.timeline.get_pdu(e).await;
if pdu.is_ok_and(|p| *p.kind() == TimelineEventType::RoomMember) {
trace!("omitting member event {e:?} from returned state");
// skip members
return None;
}
if omit_members && let Ok(e) = event_id.as_ref() {
let pdu = services.rooms.timeline.get_pdu(e).await;
if pdu.is_ok_and(|p| *p.kind() == TimelineEventType::RoomMember) {
trace!("omitting member event {e:?} from returned state");
// skip members
return None;
}
}
Some(event_id)
@@ -289,7 +220,7 @@ async fn create_join_event(
Ok(assign!(create_join_event::v2::RoomState::new(), {
auth_chain,
state,
event: to_raw_value(&CanonicalJsonValue::Object(value)).ok(),
event: to_raw_value(&CanonicalJsonValue::Object(pdu)).ok(),
members_omitted: omit_members,
servers_in_room,
}))
@@ -309,24 +240,45 @@ pub(crate) async fn create_join_event_v2_route(
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
if let Some(server) = body.room_id.server_name() {
if services.moderation.is_remote_server_forbidden(server) {
warn!(
"Server {} tried joining room ID {} through us which has a server name that is \
globally forbidden. Rejecting.",
body.identity, &body.room_id,
);
return Err!(Request(Forbidden(warn!(
"Room ID server name {server} is banned on this homeserver."
))));
}
let room_version = services.rooms.state.get_room_version(&body.room_id).await?;
let create_event = services
.rooms
.state_accessor
.get_room_create_event(&body.room_id)
.await;
let room_version_rules = room_version.rules().unwrap();
let (value, membership, sender, target) = validate_any_membership_event(
&services,
&body.pdu,
&room_version_rules,
create_event.event_id.clone(),
body.room_id.clone(),
body.event_id.clone(),
)
.await?;
if membership != MembershipState::Join {
return Err!(Request(InvalidParam("Invalid membership (expected `join`)")));
}
if sender.server_name() != body.identity {
return Err!(Request(InvalidParam("Sender belongs to a different server")));
}
if sender != target {
return Err!(Request(InvalidParam("Sender does not match state key")));
}
let now = Instant::now();
let room_state =
create_join_event(&services, &body.identity, &body.room_id, &body.pdu, body.omit_members)
.boxed()
.await?;
let room_state = create_join_event(
&services,
&body.identity,
&body.room_id,
&body.event_id,
&room_version_rules,
value,
body.omit_members,
)
.boxed()
.await?;
info!(
"Finished sending a join for {} in {} in {:?}",
body.identity,
+29 -97
View File
@@ -1,21 +1,11 @@
use axum::extract::State;
use conduwuit::{
Err, Result, err, info,
matrix::{event::gen_event_id_canonical_json, pdu::PduEvent},
warn,
};
use conduwuit::{Err, Event, Result, debug_info, err, matrix::pdu::PduEvent, warn};
use futures::FutureExt;
use ruma::{
OwnedUserId,
api::federation::membership::create_knock_event,
events::{
StateEventType,
room::member::{MembershipState, RoomMemberEventContent},
},
serde::JsonObject,
api::federation::membership::create_knock_event, events::room::member::MembershipState,
};
use crate::Ruma;
use crate::{Ruma, server::utils::validate_any_membership_event};
/// # `PUT /_matrix/federation/v1/send_knock/{roomId}/{eventId}`
///
@@ -33,18 +23,7 @@ pub(crate) async fn create_knock_event_v1_route(
forbidden. Rejecting.",
body.identity, &body.room_id,
);
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
if let Some(server) = body.room_id.server_name() {
if services.moderation.is_remote_server_forbidden(server) {
warn!(
"Server {} tried knocking room ID {} which has a server name that is globally \
forbidden. Rejecting.",
body.identity, &body.room_id,
);
return Err!(Request(Forbidden("Server is banned on this homeserver.")));
}
return Err!(Request(Forbidden("Federation denied with {}", body.identity)));
}
if !services.rooms.metadata.exists(&body.room_id).await {
@@ -57,8 +36,9 @@ pub(crate) async fn create_knock_event_v1_route(
.server_in_room(services.globals.server_name(), &body.room_id)
.await
{
info!(
debug_info!(
origin = body.identity.as_str(),
room_id = %body.room_id,
"Refusing to serve send_knock for room we aren't participating in"
);
return Err!(Request(NotFound("This server is not participating in that room.")));
@@ -72,88 +52,40 @@ pub(crate) async fn create_knock_event_v1_route(
.await?;
let room_version = services.rooms.state.get_room_version(&body.room_id).await?;
let create_event = services
.rooms
.state_accessor
.get_room_create_event(&body.room_id)
.await;
let room_version_rules = room_version.rules().unwrap();
if !room_version_rules.authorization.knocking {
return Err!(Request(Forbidden("Room version does not support knocking.")));
}
let Ok((event_id, value)) = gen_event_id_canonical_json(&body.pdu, &room_version_rules)
else {
// Event could not be converted to canonical json
return Err!(Request(InvalidParam("Could not convert event to canonical json.")));
};
let event_type: StateEventType = serde_json::from_value(
value
.get("type")
.ok_or_else(|| err!(Request(InvalidParam("Event has no event type."))))?
.clone()
.into(),
let (mut event, target_membership, sender, target) = validate_any_membership_event(
&services,
&body.pdu,
&room_version_rules,
create_event.event_id().to_owned(),
body.room_id.clone(),
body.event_id.clone(),
)
.map_err(|e| err!(Request(InvalidParam("Event has invalid event type: {e}"))))?;
.await?;
if event_type != StateEventType::RoomMember {
return Err!(Request(InvalidParam(
"Not allowed to send non-membership state event to knock endpoint.",
)));
if target_membership != MembershipState::Knock {
return Err!(Request(InvalidParam("Invalid membership (expected `knock`)")));
}
let content: RoomMemberEventContent = serde_json::from_value(
value
.get("content")
.ok_or_else(|| err!(Request(InvalidParam("Membership event has no content"))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(InvalidParam("Event has invalid membership content: {e}"))))?;
if content.membership != MembershipState::Knock {
return Err!(Request(InvalidParam(
"Not allowed to send a non-knock membership event to knock endpoint."
)));
}
// ACL check sender server name
let sender: OwnedUserId = serde_json::from_value(
value
.get("sender")
.ok_or_else(|| err!(Request(InvalidParam("Event has no sender user ID."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson("Event sender is not a valid user ID: {e}"))))?;
services
.rooms
.event_handler
.acl_check(sender.server_name(), &body.room_id)
.await?;
// check if origin server is trying to send for another server
if sender.server_name() != body.identity {
return Err!(Request(BadJson("Not allowed to knock on behalf of another server/user.")));
return Err!(Request(InvalidParam("Sender belongs to a different server")));
}
if sender != target {
return Err!(Request(InvalidParam("Sender does not match state key")));
}
let state_key: OwnedUserId = serde_json::from_value(
value
.get("state_key")
.ok_or_else(|| err!(Request(InvalidParam("Event does not have a state_key"))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson("Event does not have a valid state_key: {e}"))))?;
event.insert("event_id".to_owned(), body.event_id.as_str().into());
if state_key != sender {
return Err!(Request(InvalidParam("state_key does not match sender user of event.")));
}
let mut event: JsonObject = serde_json::from_str(body.pdu.get())
.map_err(|e| err!(Request(InvalidParam("Invalid knock event PDU: {e}"))))?;
event.insert("event_id".to_owned(), "$placeholder".into());
let pdu: PduEvent = serde_json::from_value(event.into())
let pdu = PduEvent::from_id_val(&body.event_id, event.clone())
.map_err(|e| err!(Request(InvalidParam("Invalid knock event PDU: {e}"))))?;
let mutex_lock = services
@@ -166,7 +98,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, &body.event_id, event, false)
.boxed()
.await?
.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;
@@ -181,7 +113,7 @@ pub(crate) async fn create_knock_event_v1_route(
let knock_room_state = services
.rooms
.state
.summary_stripped(&pdu, &body.room_id, &sender)
.summary_stripped(&pdu, &body.room_id, &sender, true)
.await;
Ok(create_knock_event::v1::Response::new(knock_room_state))
+31 -104
View File
@@ -1,18 +1,11 @@
use axum::extract::State;
use conduwuit::{Err, Result, err, info, matrix::event::gen_event_id_canonical_json};
use conduwuit_service::Services;
use conduwuit::{Err, Result, debug_info, err};
use futures::FutureExt;
use ruma::{
OwnedRoomId, OwnedUserId, RoomId, ServerName,
api::federation::membership::create_leave_event,
events::{
StateEventType,
room::member::{MembershipState, RoomMemberEventContent},
},
api::federation::membership::create_leave_event, events::room::member::MembershipState,
};
use serde_json::value::RawValue as RawJsonValue;
use crate::Ruma;
use crate::{Ruma, server::utils::validate_any_membership_event};
/// # `PUT /_matrix/federation/v2/send_leave/{roomId}/{eventId}`
///
@@ -21,17 +14,8 @@ pub(crate) async fn create_leave_event_v2_route(
State(services): State<crate::State>,
body: Ruma<create_leave_event::v2::Request>,
) -> Result<create_leave_event::v2::Response> {
create_leave_event(&services, &body.identity, &body.room_id, &body.pdu).await?;
Ok(create_leave_event::v2::Response::new())
}
async fn create_leave_event(
services: &Services,
origin: &ServerName,
room_id: &RoomId,
pdu: &RawJsonValue,
) -> Result {
let room_id = body.room_id.as_ref();
let origin = &body.identity;
if !services.rooms.metadata.exists(room_id).await {
return Err!(Request(NotFound("Room is unknown to this server.")));
}
@@ -42,9 +26,10 @@ async fn create_leave_event(
.server_in_room(services.globals.server_name(), room_id)
.await
{
info!(
debug_info!(
origin = origin.as_str(),
"Refusing to serve backfill for room we aren't participating in"
room_id = %room_id,
"Refusing to send_leave for room we aren't participating in"
);
return Err!(Request(NotFound("This server is not participating in that room.")));
}
@@ -56,91 +41,31 @@ async fn create_leave_event(
.acl_check(origin, room_id)
.await?;
// We do not add the event_id field to the pdu here because of signature and
// hashes checks
let room_version = services.rooms.state.get_room_version(room_id).await?;
let create_event = services
.rooms
.state_accessor
.get_room_create_event(room_id)
.await;
let room_version_rules = room_version.rules().unwrap();
let Ok((event_id, value)) = gen_event_id_canonical_json(pdu, &room_version_rules) else {
// Event could not be converted to canonical json
return Err!(Request(BadJson("Could not convert event to canonical json.")));
};
let event_room_id: OwnedRoomId = serde_json::from_value(
serde_json::to_value(
value
.get("room_id")
.ok_or_else(|| err!(Request(BadJson("Event missing room_id property."))))?,
)
.expect("CanonicalJson is valid json value"),
let (value, membership, sender, target) = validate_any_membership_event(
&services,
&body.pdu,
&room_version_rules,
create_event.event_id.clone(),
body.room_id.clone(),
body.event_id.clone(),
)
.map_err(|e| err!(Request(BadJson(warn!("room_id field is not a valid room ID: {e}")))))?;
if event_room_id != room_id {
return Err!(Request(BadJson("Event room_id does not match request path room ID.")));
.await?;
if membership != MembershipState::Leave {
return Err!(Request(InvalidParam("Invalid membership (expected `leave`)")));
}
let content: RoomMemberEventContent = serde_json::from_value(
value
.get("content")
.ok_or_else(|| err!(Request(BadJson("Event missing content property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("Event content is empty or invalid: {e}")))))?;
if content.membership != MembershipState::Leave {
return Err!(Request(BadJson(
"Not allowed to send a non-leave membership event to leave endpoint."
)));
if sender.server_name() != body.identity {
return Err!(Request(InvalidParam("Sender belongs to a different server")));
}
let event_type: StateEventType = serde_json::from_value(
value
.get("type")
.ok_or_else(|| err!(Request(BadJson("Event missing type property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("Event has invalid state event type: {e}")))))?;
if event_type != StateEventType::RoomMember {
return Err!(Request(BadJson(
"Not allowed to send non-membership state event to leave endpoint."
)));
}
// ACL check sender server name
let sender: OwnedUserId = serde_json::from_value(
value
.get("sender")
.ok_or_else(|| err!(Request(BadJson("Event missing sender property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("sender property is not a valid user ID: {e}")))))?;
services
.rooms
.event_handler
.acl_check(sender.server_name(), room_id)
.await?;
if sender.server_name() != origin {
return Err!(Request(BadJson("Not allowed to leave on behalf of another server/user.")));
}
let state_key: OwnedUserId = serde_json::from_value(
value
.get("state_key")
.ok_or_else(|| err!(Request(BadJson("Event missing state_key property."))))?
.clone()
.into(),
)
.map_err(|e| err!(Request(BadJson(warn!("State key is not a valid user ID: {e}")))))?;
if state_key != sender {
return Err!(Request(BadJson("State key does not match sender user.")));
if sender != target {
return Err!(Request(InvalidParam("Sender does not match state key")));
}
let mutex_lock = services
@@ -153,7 +78,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, &body.event_id, value, false)
.boxed()
.await?
.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;
@@ -164,5 +89,7 @@ async fn create_leave_event(
.sending
.send_pdu_room(room_id, &pdu_id)
.boxed()
.await
.await?;
Ok(create_leave_event::v2::Response::new())
}
+105 -2
View File
@@ -1,7 +1,10 @@
use conduwuit::{Err, Result, is_false};
use conduwuit::{Err, Result, err, is_false};
use conduwuit_service::Services;
use futures::{FutureExt, future::OptionFuture, join};
use ruma::{EventId, RoomId, ServerName};
use ruma::{
CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, ServerName,
UserId, events::room::member::MembershipState, room_version_rules::RoomVersionRules,
};
pub(super) struct AccessCheck<'a> {
pub(super) services: &'a Services,
@@ -63,3 +66,103 @@ pub(super) async fn assert(&self) -> Result {
Ok(())
}
}
/// Performs validation on a membership event that should be run on any event a
/// remote is trying to send via us.
///
/// ## Checks performed
///
/// 1. PDU room ID matches request path room ID
/// 2. PDU event ID matches request path event ID
/// 3. Signature check
/// 4. Event type check
/// 5. `sender` field presence (and parsing)
/// 6. `state_key` field presence (and parsing)
/// 7. PDU room format check (PDU check 1)
///
/// ## Returns
///
/// A resulting tuple of (PDU JSON, target membership state, sender, recipient).
pub(crate) async fn validate_any_membership_event(
services: &crate::State,
body: &serde_json::value::RawValue,
room_version_rules: &RoomVersionRules,
create_event_id: OwnedEventId,
expected_room_id: OwnedRoomId,
expected_event_id: OwnedEventId,
) -> Result<(CanonicalJsonObject, MembershipState, OwnedUserId, OwnedUserId)> {
let (template_room_id, template_event_id, pdu) = services
.rooms
.event_handler
.parse_incoming_pdu(body, Some(room_version_rules))
.await
.map_err(|e| err!(Request(BadJson("Invalid membership PDU: {e}"))))?;
if template_room_id != expected_room_id {
return Err!(Request(InvalidParam("Membership event does not belong to requested room")));
}
if template_event_id != expected_event_id {
return Err!(Request(InvalidParam(debug_warn!(
%template_event_id,
%expected_event_id,
"Membership event ID does not match provided event ID"
))));
}
services
.server_keys
.verify_event(&pdu, room_version_rules)
.await
.map_err(|e| {
err!(Request(InvalidParam("Signature verification failed on membership event: {e}")))
})?;
// Ensure this is a membership event
if pdu
.get("type")
.expect("event must have a type")
.as_str()
.expect("type must be a string")
!= "m.room.member"
{
return Err!(Request(BadJson(
"Not allowed to send non-membership event to this endpoint"
)));
}
let membership = pdu
.get("content")
.ok_or_else(|| err!(Request(BadJson("Event missing content property"))))?
.as_object()
.ok_or_else(|| err!(Request(BadJson("Event content is not an object"))))?
.get("membership")
.ok_or_else(|| err!(Request(BadJson("Event missing membership property"))))?
.as_str()
.ok_or_else(|| err!(Request(BadJson("Event is not a string"))))?
.to_owned();
let sender_user = pdu
.get("sender")
.and_then(|v| v.as_str())
.map(UserId::parse)
.and_then(Result::ok)
.ok_or_else(|| err!(Request(InvalidParam("Invalid sender property"))))?;
let recipient_user = pdu
.get("state_key")
.and_then(|v| v.as_str())
.map(UserId::parse)
.and_then(Result::ok)
.ok_or_else(|| err!(Request(InvalidParam("Invalid state_key property"))))?;
// Do a quick format check. The spec doesn't suggest this, but it's probably
// a good idea nonetheless.
service::rooms::event_handler::Service::pdu_format_check_1(
&pdu,
room_version_rules,
&create_event_id,
)
.map_err(|e| {
err!(Request(InvalidParam("Membership event violates the room event format: {e}")))
})?;
Ok((pdu, membership.into(), sender_user, recipient_user))
}
+1
View File
@@ -2442,6 +2442,7 @@ pub struct OauthConfig {
/// legacy authentication will be unable to log in.
///
/// default: "hybrid"
#[serde(default)]
compatibility_mode: OAuthMode,
/// display: hidden
+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
+2 -2
View File
@@ -219,7 +219,7 @@ fn configure_output_err(mut output: MadSkin) -> MadSkin {
use termimad::{Alignment, CompoundStyle, LineStyle, crossterm::style::Color};
let code_style = CompoundStyle::with_fgbg(Color::AnsiValue(196), Color::AnsiValue(234));
output.inline_code = code_style.clone();
output.inline_code = code_style;
output.code_block = LineStyle {
left_margin: 0,
right_margin: 0,
@@ -234,7 +234,7 @@ fn configure_output(mut output: MadSkin) -> MadSkin {
use termimad::{Alignment, CompoundStyle, LineStyle, crossterm::style::Color};
let code_style = CompoundStyle::with_fgbg(Color::AnsiValue(40), Color::AnsiValue(234));
output.inline_code = code_style.clone();
output.inline_code = code_style;
output.code_block = LineStyle {
left_margin: 0,
right_margin: 0,
+1 -1
View File
@@ -134,7 +134,7 @@ async fn check(&self) -> Result<()> {
let response = self
.services
.client
.default
.external_resource
.get(CHECK_FOR_ANNOUNCEMENTS_URL)
.send()
.await?
+9
View File
@@ -224,6 +224,15 @@ pub async fn is_exclusive_user_id(&self, user_id: &UserId) -> bool {
.any(|info| info.is_exclusive_user_match(user_id))
}
/// Checks if a given user id matches any appservice regex, exclusive or
/// otherwise
pub async fn is_user_id(&self, user_id: &UserId) -> bool {
self.read()
.await
.values()
.any(|info| info.is_user_match(user_id))
}
/// Checks if a given room alias matches any exclusive appservice regex
pub async fn is_exclusive_alias(&self, alias: &RoomAliasId) -> bool {
self.read()
+14 -19
View File
@@ -16,9 +16,9 @@ pub struct Service {
pub matrix_resolver: Arc<MatrixResolver>,
pub dns_resolver: Arc<TokioResolver>,
pub default: reqwest::Client,
pub dns: reqwest::Client,
pub url_preview: reqwest::Client,
pub extern_media: reqwest::Client,
pub external_resource: reqwest::Client,
pub federation: reqwest::Client,
pub federation_slow: reqwest::Client,
pub sender: reqwest::Client,
@@ -35,17 +35,17 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
let dns_resolver = get_dns_resolver(args.server)?;
let dns_client = base(&args.server.config)?
.connect_timeout(Duration::from_secs(args.server.config.well_known_conn_timeout))
.read_timeout(Duration::from_secs(args.server.config.well_known_timeout))
.timeout(Duration::from_secs(args.server.config.well_known_timeout))
.pool_max_idle_per_host(0)
.redirect(redirect::Policy::limited(4))
.build()?;
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)?
.connect_timeout(Duration::from_secs(args.server.config.well_known_conn_timeout))
.read_timeout(Duration::from_secs(args.server.config.well_known_timeout))
.timeout(Duration::from_secs(args.server.config.well_known_timeout))
.pool_max_idle_per_host(0)
.redirect(redirect::Policy::limited(4))
.build()?
)
.http_client(dns_client.clone())
.dns_resolver(dns_resolver.clone())
.build()?);
let matrix_dns_resolver = matrix_resolver.create_dns_resolver();
@@ -69,23 +69,19 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
matrix_resolver,
dns_resolver,
default: base(config)?
.dns_resolver(matrix_dns_resolver.clone())
.build()?,
dns: dns_client,
url_preview: base(config)
.and_then(|builder| {
builder_interface(builder, url_preview_bind_iface.as_deref())
})?
.local_address(url_preview_bind_addr)
.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(matrix_dns_resolver.clone())
external_resource: base(config)?
.redirect(redirect::Policy::limited(3))
.build()?,
@@ -128,7 +124,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
appservice: base(config)?
.dns_resolver(matrix_dns_resolver.clone())
.dns_resolver(matrix_dns_resolver)
.connect_timeout(Duration::from_secs(5))
.read_timeout(Duration::from_secs(config.appservice_timeout))
.timeout(Duration::from_secs(config.appservice_timeout))
@@ -138,7 +134,6 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
pusher: base(config)?
.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)
+1 -1
View File
@@ -260,7 +260,7 @@ async fn location_request(&self, location: &str) -> Result<FileMeta> {
let response = self
.services
.client
.extern_media
.external_resource
.get(location)
.send()
.await?;
+2 -2
View File
@@ -829,7 +829,7 @@ async fn fix_local_invite_state(services: &Services) -> Result {
&& services.globals.user_is_local(&membership_event.sender) {
// build and save stripped state for their invite in the database
let stripped_state = services.rooms.state.summary_stripped(&membership_event, &room_id, &user_id).await;
let stripped_state = services.rooms.state.summary_stripped(&membership_event, &room_id, &user_id, false).await;
userroomid_invitestate.put((&user_id, &room_id), Json(stripped_state));
fixed = fixed.saturating_add(1);
}
@@ -867,7 +867,7 @@ async fn split_userid_password(services: &Services) -> Result {
userid_password.remove(&user_id);
remote_users = remote_users.saturating_add(1);
} else if hash.is_empty() {
if !(services.appservice.is_exclusive_user_id(&user_id).await
if !(services.appservice.is_user_id(&user_id).await
|| user_id == services.globals.server_user)
{
info!("Marking {user_id} as deactivated");
+39 -6
View File
@@ -115,6 +115,21 @@ pub enum SessionCompletionStatus {
Complete(OwnedUserId),
}
pub enum ClaimedLocalUser {
/// The claim refers to an existing user.
Existing(OwnedUserId),
/// The claim refers to a new user ID which should be registered.
New(OwnedUserId),
}
impl ClaimedLocalUser {
fn into_user_id(self) -> OwnedUserId {
match self {
| Self::Existing(user_id) | Self::New(user_id) => user_id,
}
}
}
#[async_trait]
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@@ -136,6 +151,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
config: config.clone(),
client_secret: if let Some(client_secret_file) = &config.client_secret_file {
std::fs::read_to_string(client_secret_file)
.map(|client_secret| client_secret.trim().to_owned())
.map(ClientSecret::new)
.map_err(|err| err!("Failed to read OIDC client secret file: {err}"))?
} else if let Some(client_secret) = &config.client_secret {
@@ -303,7 +319,7 @@ pub async fn complete_session(
.expect("claims should be an object")
.to_owned();
debug!(?all_claims);
debug!(?all_claims, "Got claims from the identity provider");
let subject = claims.subject().as_str();
@@ -326,14 +342,13 @@ pub async fn complete_session(
.get(&config.preferred_username_claim)
.and_then(|claim| claim.as_str())
{
self.services
.users
.determine_registration_user_id(Some(preferred_username.to_owned()), None, None)
self.identify_claimed_local_user(preferred_username)
.await
.map(ClaimedLocalUser::into_user_id)
.map_err(|err| {
error!("Preferred username claim is not a valid localpart: {err}");
"Your preferred username is not a valid Matrix user ID localpart. Contact \
your homeserver's administrator."
"Your preferred username could not be converted to a valid Matrix user ID. \
Contact your homeserver's administrator."
})?
} else {
error!("Preferred username claim was not present or was not a string");
@@ -482,4 +497,22 @@ pub fn link_user(&self, user_id: &UserId, subject: &str) {
}
pub fn unlink_user(&self, subject: &str) { self.db.openidsubject_localpart.remove(subject); }
/// Determine what user ID a localpart claim refers to.
pub async fn identify_claimed_local_user(&self, claim: &str) -> Result<ClaimedLocalUser> {
if let Ok(user_id) =
UserId::parse(format!("@{}:{}", claim, self.services.globals.server_name()))
&& self.services.users.status(&user_id).await.is_active()
{
Ok(ClaimedLocalUser::Existing(user_id))
} else {
let user_id = self
.services
.users
.determine_registration_user_id(Some(claim.to_owned()), None, None)
.await?;
Ok(ClaimedLocalUser::New(user_id))
}
}
}
@@ -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()
@@ -227,7 +241,7 @@ pub async fn get_missing_events(
latest_events.clear();
for raw_event in response.events {
let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event).await?;
let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event, None).await?;
let pdu = PduEvent::from_id_val(&event_id, pdu_json).map_err(|e| {
err!(Request(BadJson("Failed to parse gapfilled event {event_id}: {e}")))
})?;
@@ -370,8 +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)
let (_, calculated_event_id, value) = self
.parse_incoming_pdu(&res.pdu, Some(room_version_rules))
.await?;
if calculated_event_id != event_id {
@@ -538,11 +552,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());
+52 -21
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;
@@ -49,17 +48,52 @@ pub(super) async fn fetch_and_persist_event_auth<Pdu>(
let mut auth_chain_map = HashMap::with_capacity(event_auth.auth_chain.len());
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?;
let (auth_event_room_id, auth_event_id, auth_pdu_json) = match self
.parse_incoming_pdu(&auth_pdu_json, Some(room_version_rules))
.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 +131,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 +155,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 +184,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());
@@ -314,7 +314,7 @@ pub(super) async fn fetch_full_state(
.iter()
.stream()
.broad_filter_map(|raw_event_json| async {
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok()
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json, None).await.ok()
&& parsed.0 == room_id
{
Some(parsed)
@@ -351,7 +351,7 @@ pub(super) async fn fetch_full_state(
.iter()
.stream()
.broad_filter_map(|raw_event_json| async {
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok()
if let Some(parsed) = self.parse_incoming_pdu(raw_event_json, None).await.ok()
&& parsed.0 == room_id
{
Some(parsed)
@@ -4,46 +4,37 @@
};
use conduwuit::{
Err, Event, Result, debug, debug_error, debug_warn, defer, err, error, info,
matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
Err, Event, Result, debug, debug_error, debug_warn, defer, matrix::PartialPdu, trace,
utils::time::jitter,
};
use futures::{
FutureExt, StreamExt,
future::{OptionFuture, try_join4},
};
use ruma::{CanonicalJsonValue, EventId, OwnedUserId, RoomId, ServerName, UserId};
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};
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,
@@ -55,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(
@@ -112,21 +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
{
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 {
if !room_exists {
if is_interesting_member_event {
// TODO: handle interesting membership events where we aren't in
// the room
}
return Err!(Request(NotFound("Room is unknown to this server")));
}
@@ -152,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);
}
@@ -166,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(
@@ -181,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,24 +1,25 @@
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.
///
/// **External callers likely want `handle_incoming_pdu` instead.**
///
/// [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))]
pub(super) async fn handle_outlier_pdu<'a, Pdu>(
#[tracing::instrument(name = "handle_outlier", skip_all)]
pub async fn handle_outlier_pdu<'a, Pdu>(
&self,
origin: &'a ServerName,
create_event: &'a Pdu,
@@ -29,61 +30,38 @@ 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
// 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)?;
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"
);
})?;
value.remove("unsigned");
// 2. Check signatures, otherwise drop
// 3. check content hash, redact if doesn't match
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));
}
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}"
))));
},
};
// 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 +69,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 +91,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 +128,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 +140,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 +170,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 +181,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);
}
}
+2 -17
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,7 +20,7 @@
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};
@@ -114,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,45 +82,14 @@ 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(
/// 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,
room_version_rules: &RoomVersionRules,
) -> Result<(OwnedEventId, CanonicalJsonObject)> {
let (event_id, value) =
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)?;
Ok((event_id, value))
}
pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
room_version_rules: Option<&RoomVersionRules>,
) -> Result<Parsed> {
let value = serde_json::from_str::<CanonicalJsonObject>(pdu.get()).map_err(|e| {
err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}")))
})?;
@@ -129,20 +100,23 @@ pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
let room_id = extract_room_id(event_type, &value)?;
let room_version_rules = self
.services
.state
.get_room_version(&room_id)
.await
.unwrap_or(RoomVersionId::V1)
.rules()
.unwrap();
let room_version_rules = match room_version_rules {
| Some(r) => r,
| None => &self
.services
.state
.get_room_version(&room_id)
.await
.unwrap_or(RoomVersionId::V1)
.rules()
.expect("room version must be supported"),
};
let (event_id, value) =
gen_event_id_canonical_json(pdu, &room_version_rules).map_err(|e| {
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,291 @@
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 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")));
} else if event_format.allow_room_create_in_auth_events && !create_event_in_auth_events {
return Err!(Request(BadJson("PDU does not reference the room 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 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 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."
+18 -3
View File
@@ -1,7 +1,7 @@
use std::{collections::HashMap, fmt::Write, sync::Arc};
use async_trait::async_trait;
use conduwuit::debug;
use conduwuit::{debug, utils::stream::WidebandExt};
use conduwuit_core::{
Event, PduEvent, Result, err,
result::FlatOk,
@@ -30,7 +30,7 @@
short::{ShortEventId, ShortStateHash},
state_compressor::{CompressedState, parse_compressed_state_event},
},
sync,
sending, sync,
};
pub struct Service {
@@ -45,6 +45,7 @@ struct Services {
state_cache: Dep<rooms::state_cache::Service>,
state_accessor: Dep<rooms::state_accessor::Service>,
state_compressor: Dep<rooms::state_compressor::Service>,
sending: Dep<sending::Service>,
sync: Dep<sync::Service>,
timeline: Dep<rooms::timeline::Service>,
}
@@ -71,6 +72,7 @@ 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"),
sending: args.depend::<sending::Service>("sending"),
sync: args.depend::<sync::Service>("sync"),
timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
},
@@ -307,6 +309,7 @@ pub async fn summary_stripped(
event: &PduEvent,
room_id: &RoomId,
target_user: &UserId,
federation: bool,
) -> Vec<RawStrippedState> {
let mut state_events = [
(&StateEventType::RoomCreate, ""),
@@ -343,8 +346,20 @@ pub async fn summary_stripped(
.await
.into_iter()
.filter_map(Result::ok)
.map(|pdu| RawStrippedState::Pdu(serde_json::value::to_raw_value(&pdu).unwrap()))
.stream()
.wide_then(async |pdu| {
let formatted = if federation {
self.services
.sending
.convert_to_outgoing_federation_event(pdu.to_canonical_object())
.await
} else {
serde_json::value::to_raw_value(&pdu).unwrap()
};
RawStrippedState::Pdu(formatted)
})
.collect()
.await
}
/// Set the state hash to a new version, but does not update state_cache.
+1 -1
View File
@@ -114,7 +114,7 @@ pub async fn update_membership(
let invite_state = if is_local {
self.services
.state
.summary_stripped(pdu, room_id, user_id)
.summary_stripped(pdu, room_id, user_id, false)
.await
} else {
vec![]
+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 {
+7 -4
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}");
@@ -166,8 +166,11 @@ pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Resu
/// Backfills a single PDU.
#[tracing::instrument(skip(self, pdu), level = "debug")]
pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) -> Result<()> {
let (room_id, event_id, value) =
self.services.event_handler.parse_incoming_pdu(&pdu).await?;
let (room_id, event_id, value) = self
.services
.event_handler
.parse_incoming_pdu(&pdu, None)
.await?;
// Lock so we cannot backfill the same pdu twice at the same time
let mutex_lock = self
@@ -185,7 +188,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?;
+19 -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 {
@@ -306,12 +308,27 @@ pub async fn create_hash_and_sign_event(
| _ => Err!(Request(Unknown(warn!("Signing event failed: {e}")))),
};
}
// Evil hack because only the PDU JSON gets signed
pdu.hashes = serde_json::from_value(serde_json::to_value(
pdu_json
.get("hashes")
.expect("must have hashes after signing")
.clone(),
)?)?;
pdu.signatures = serde_json::from_value(serde_json::to_value(
pdu_json
.get("signatures")
.expect("must have signatures after signing")
.clone(),
)?)?;
// Generate event id
pdu.event_id = gen_event_id(&pdu_json, &room_version_rules)?;
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"),
+39 -41
View File
@@ -6,9 +6,12 @@
response::Redirect,
routing::on,
};
use conduwuit_service::{oauth::grant::AuthorizationCodeResponse, oidc::SessionCompletionStatus};
use conduwuit_service::{
oauth::grant::AuthorizationCodeResponse,
oidc::{ClaimedLocalUser, SessionCompletionStatus},
};
use futures::FutureExt;
use ruma::{OwnedServerName, UserId};
use ruma::OwnedServerName;
use serde::{Deserialize, de::IgnoredAny};
use tower_sessions::Session;
@@ -99,65 +102,60 @@ async fn route_complete(
state: OidcSessionState::Authorized { claims: Box::new(claims.clone()) },
})
.await
.expect("Should be able to serialize oidc session");
.expect("should be able to serialize oidc session");
services.oidc.complete_session(&claims, None).await
},
| OidcSessionState::Authorized { claims } => {
let supplied_user_id = if let Some(form) = form {
if let Ok(user_id) = UserId::parse(format!(
"@{}:{}",
&form.username,
services.globals.server_name()
)) && services.users.status(&user_id).await.is_active()
match services
.oidc
.identify_claimed_local_user(&form.username)
.await
{
let user_card = UserCard::for_local_user(&services, user_id.clone()).await;
| Ok(ClaimedLocalUser::Existing(user_id)) => {
let user_card =
UserCard::for_local_user(&services, user_id.clone()).await;
if let Some(password) = form.password {
if services
.users
.check_password(&user_id, &password)
.await
.is_ok()
{
Some(user_id)
if let Some(password) = form.password {
if services
.users
.check_password(&user_id, &password)
.await
.is_ok()
{
Some(user_id)
} else {
return response!(OidcComplete::new(
context,
OidcCompleteBody::PasswordPrompt {
username: form.username,
user_card,
password_error: true
}
));
}
} else {
return response!(OidcComplete::new(
context,
OidcCompleteBody::PasswordPrompt {
username: form.username,
user_card,
password_error: true
password_error: false,
}
));
}
} else {
},
| Ok(ClaimedLocalUser::New(user_id)) => Some(user_id),
| Err(err) => {
return response!(OidcComplete::new(
context,
OidcCompleteBody::PasswordPrompt {
username: form.username,
user_card,
password_error: false,
OidcCompleteBody::UsernamePrompt {
server_name: services.globals.server_name().to_owned(),
username_error: Some(err.message()),
}
));
}
} else {
match services
.users
.determine_registration_user_id(Some(form.username), None, None)
.await
{
| Ok(user_id) => Some(user_id),
| Err(err) => {
return response!(OidcComplete::new(
context,
OidcCompleteBody::UsernamePrompt {
server_name: services.globals.server_name().to_owned(),
username_error: Some(err.message()),
}
));
},
}
},
}
} else {
None