Compare commits

...

49 Commits

Author SHA1 Message Date
strawberry 6662aa21ef bump complement to 6e4426a9e63233f9821a4d2382bfed145244183f
• Updated input 'complement':
    'github:matrix-org/complement/0d14432e010482ea9e13a6f7c47c1533c0c9d62f' (2024-07-10)
  → 'github:matrix-org/complement/6e4426a9e63233f9821a4d2382bfed145244183f' (2024-07-30)

Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-08 17:18:26 -04:00
Jason Volk 9b9a91f6ef add our_modules() for log capture filter convenience
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk 94b805de0b generalize log capture to all admin commands; simplify handler
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk eded585f79 rename some command types for consistency
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk a552321287 add Clap to Error
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk f98bfbbdcd add string stream convenience util; add ?Sized bounds on log fmt functors
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk e9ab548445 tweak some tracing log levels
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk 15184d1a79 Add span scope names array to capture filter data; optimize values visitor vec.
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk 0c6bbde25f fix markdown table formatter for log capture
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk 13aa353765 add arrayvec as a dependency
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk ac941a801a add interface to query rocksdb properties w/ admin cmd
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk f261d44edb add method to iterate opened database columns
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk 849cfdcdfa add rocksdb stats level option with conf item
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
Jason Volk 1470331f7e add repair_mode documentation to example conf
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-08 18:09:21 +00:00
strawberry dcaa96a79f sync: always include_redundant_members when using element_hacks feature
see:
- https://github.com/matrix-org/matrix-js-sdk/pull/4329
- https://github.com/element-hq/synapse/blob/568051c0f07393b786b9d813a1db53dd332c9fc2/synapse/handlers/pagination.py#L639
- https://matrix.to/#/!n8DKU1BeeJilOJXDPr:seirdy.one/$gvGF7GuPrLPR8nHg4UiHvP-ZVwuiMkYS8O6VlsDCW9w?via=awawawawawawawawawawawawawawawawawawawawawawawawawawawawawawaw.gay&via=transfem.dev&via=matrix.org

Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:07:01 -04:00
strawberry 101a966922 docs: mention potential DNS issues when using docker
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:06:28 -04:00
strawberry 57940f38ea set last_seen_ip on new/initial device creation
this is not automatically updating, but at least have something
useful there instead of nothing

Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:06:19 -04:00
strawberry 22f8c339fc debian: remove --verbose from postinst
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:06:09 -04:00
strawberry c896acb729 dont consider presence status changes apart of presence spam
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:06:01 -04:00
strawberry 7f1bc490d6 docs: update configuration docs for the new -O option
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:05:27 -04:00
strawberry 1091c30845 docs: mention FreeBSD specific build commandline for deployment
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:05:16 -04:00
strawberry 12d9e454db display success/failure count in get-remote-pdu-list
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:04:51 -04:00
strawberry 82b973a8b1 nix: use cortex-a73 / ARMv8-a, strip and remove unnecessary bloat
for some reason gcc is included in OCI images

Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:04:41 -04:00
strawberry 71ee31a014 add missing newline in banned room list output
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-07 12:04:12 -04:00
skedastically 69b4ec81b6 update container registry links (#549 / #548)
* gitlab URL

* Update docker.md
2024-08-05 15:23:40 -04:00
Jason Volk fe0969e7ca fix release-max-perf profile [ci skip]
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-01 13:37:27 +00:00
Jason Volk 2e3e14b045 refactor some additional errors
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-01 10:58:59 +00:00
Jason Volk f51d4237c7 bump cargo
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-01 10:58:59 +00:00
strawberry d8a87258d6 rust: bump console-subscriber to 0.4
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-01 10:58:58 +00:00
strawberry 3da2242730 bump tokio, use default-features = false for more crates
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-08-01 10:58:58 +00:00
Jason Volk 5add9a8c34 support field values in err! macro
Signed-off-by: Jason Volk <jason@zemos.net>
2024-08-01 10:58:58 +00:00
Jason Volk 5b1642f641 split server_keys from event_handler into service
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-31 21:10:22 +00:00
Jason Volk eb94aa7079 remove dangling sending/resolve unit
this was probably un-deleted during a rebase conflict resolution,
but remained outside of the mod tree.

Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-31 21:10:22 +00:00
Jason Volk 24884c3b9b move media helpers from api to services
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-31 03:04:17 +00:00
Jason Volk a0a002b17f disable legacy media compat by default; cleanse directory when disabled
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-31 03:04:17 +00:00
Jason Volk f632b06e6d use Services instead of Server for middleware function state
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-31 03:04:17 +00:00
Jason Volk 1f88866612 optimize api state extractor
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-31 03:04:17 +00:00
Jason Volk ccef1a4c8b add formal wrapping for api state
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 21:32:43 +00:00
Jason Volk 5c258f41c8 fixes for modules
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk 15126ee1b2 additional weak references where applicable
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk e37ac56dba use weak references to break services dependency cycle
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk a74613be96 improve database repair/shutdown log messages
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk a95ded167f fix type complexity
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk 95121b6e05 use single-line codeblock for specific features lists
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk e6884ef13a fix various nightly incidental lints
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk 49033db2b8 fix trailing body passed to commands
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk ddc8c3b46c split presence data object into file; improve service encapsulations
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
Jason Volk ea95627dce hoist rustc flag string array to global const
Signed-off-by: Jason Volk <jason@zemos.net>
2024-07-28 05:59:12 -07:00
strawberry c3e751542d ci: default to empty string to "unset" variables
Signed-off-by: strawberry <strawberry@puppygock.gay>
2024-07-28 05:59:12 -07:00
85 changed files with 1586 additions and 1394 deletions
+2 -2
View File
@@ -29,8 +29,8 @@ concurrency:
env:
# sccache only on main repo
SCCACHE_GHA_ENABLED: "${{ (github.event.pull_request.draft != true) && (vars.DOCKER_USERNAME != '') && (vars.GITLAB_USERNAME != '') && (vars.SCCACHE_ENDPOINT != '') && (github.event.pull_request.user.login != 'renovate[bot]') && 'true' || 'false' }}"
RUSTC_WRAPPER: "${{ (github.event.pull_request.draft != true) && (vars.DOCKER_USERNAME != '') && (vars.GITLAB_USERNAME != '') && (vars.SCCACHE_ENDPOINT != '') && (github.event.pull_request.user.login != 'renovate[bot]') && 'sccache' }}"
SCCACHE_BUCKET: "${{ (github.event.pull_request.draft != true) && (vars.DOCKER_USERNAME != '') && (vars.GITLAB_USERNAME != '') && (vars.SCCACHE_ENDPOINT != '') && (github.event.pull_request.user.login != 'renovate[bot]') && 'sccache' }}"
RUSTC_WRAPPER: "${{ (github.event.pull_request.draft != true) && (vars.DOCKER_USERNAME != '') && (vars.GITLAB_USERNAME != '') && (vars.SCCACHE_ENDPOINT != '') && (github.event.pull_request.user.login != 'renovate[bot]') && 'sccache' || '' }}"
SCCACHE_BUCKET: "${{ (github.event.pull_request.draft != true) && (vars.DOCKER_USERNAME != '') && (vars.GITLAB_USERNAME != '') && (vars.SCCACHE_ENDPOINT != '') && (github.event.pull_request.user.login != 'renovate[bot]') && 'sccache' || '' }}"
SCCACHE_S3_USE_SSL: ${{ vars.SCCACHE_S3_USE_SSL }}
SCCACHE_REGION: ${{ vars.SCCACHE_REGION }}
SCCACHE_ENDPOINT: ${{ vars.SCCACHE_ENDPOINT }}
Generated
+155 -238
View File
@@ -71,6 +71,12 @@ dependencies = [
"password-hash",
]
[[package]]
name = "arrayvec"
version = "0.7.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96d30a06541fbafbc7f82ed10c06164cfbd2c401138f6addd8404629c4b16711"
[[package]]
name = "as_variant"
version = "1.2.0"
@@ -153,34 +159,6 @@ version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c4b4d0bd25bd0b74681c0ad21497610ce1b7c91b1022cd21c80c6fbdd9476b0"
[[package]]
name = "axum"
version = "0.6.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf"
dependencies = [
"async-trait",
"axum-core 0.3.4",
"bitflags 1.3.2",
"bytes",
"futures-util",
"http 0.2.12",
"http-body 0.4.6",
"hyper 0.14.30",
"itoa",
"matchit",
"memchr",
"mime",
"percent-encoding",
"pin-project-lite",
"rustversion",
"serde",
"sync_wrapper 0.1.2",
"tower",
"tower-layer",
"tower-service",
]
[[package]]
name = "axum"
version = "0.7.5"
@@ -188,13 +166,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a6c9af12842a67734c9a2e355436e5d03b22383ed60cf13cd0c18fbfe3dcbcf"
dependencies = [
"async-trait",
"axum-core 0.4.3",
"axum-core",
"bytes",
"futures-util",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"http-body-util",
"hyper 1.4.1",
"hyper",
"hyper-util",
"itoa",
"matchit",
@@ -221,28 +199,11 @@ version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72188bed20deb981f3a4a9fe674e5980fd9e9c2bd880baa94715ad5d60d64c67"
dependencies = [
"axum 0.7.5",
"axum",
"forwarded-header-value",
"serde",
]
[[package]]
name = "axum-core"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c"
dependencies = [
"async-trait",
"bytes",
"futures-util",
"http 0.2.12",
"http-body 0.4.6",
"mime",
"rustversion",
"tower-layer",
"tower-service",
]
[[package]]
name = "axum-core"
version = "0.4.3"
@@ -252,8 +213,8 @@ dependencies = [
"async-trait",
"bytes",
"futures-util",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"http-body-util",
"mime",
"pin-project-lite",
@@ -270,13 +231,13 @@ version = "0.9.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0be6ea09c9b96cb5076af0de2e383bd2bc0c18f827cf1967bdd353e0b910d733"
dependencies = [
"axum 0.7.5",
"axum-core 0.4.3",
"axum",
"axum-core",
"bytes",
"futures-util",
"headers",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"http-body-util",
"mime",
"pin-project-lite",
@@ -295,10 +256,10 @@ dependencies = [
"arc-swap",
"bytes",
"futures-util",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"http-body-util",
"hyper 1.4.1",
"hyper",
"hyper-util",
"pin-project-lite",
"rustls 0.21.12",
@@ -317,7 +278,7 @@ checksum = "7ea4cd08ae2a5f075d28fa31190163c8106a1d2d3189442494bae22b39040a0d"
dependencies = [
"axum-server",
"bytes",
"http 1.1.0",
"http",
"http-body-util",
"pin-project",
"tokio",
@@ -439,9 +400,9 @@ checksum = "79296716171880943b8470b5f8d03aa55eb2e645a4874bdbb28adb49162e012c"
[[package]]
name = "bytemuck"
version = "1.16.1"
version = "1.16.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b236fc92302c97ed75b38da1f4917b5cdda4984745740f153a5d3059e48d725e"
checksum = "102087e286b4677862ea56cf8fc58bb2cdfa8725c40ffb80fe3a008eb7f2fc83"
[[package]]
name = "byteorder"
@@ -457,9 +418,9 @@ checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495"
[[package]]
name = "bytes"
version = "1.6.1"
version = "1.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a12916984aab3fa6e39d655a33e09c0071eb36d6ab3aea5c2d78551f1df6d952"
checksum = "fca2be1d5c43812bae364ee3f30b3afcb7877cf59f4aeb94c66f313a41d2fac9"
[[package]]
name = "bzip2-sys"
@@ -484,9 +445,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.1.6"
version = "1.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2aba8f4e9906c7ce3c73463f62a7f0c65183ada1a2d47e397cc8810827f9694f"
checksum = "26a5c3fd7bfa1ce3897a3a3501d362b2d87b7f2583ebcb4a949ec25911025cbc"
dependencies = [
"jobserver",
"libc",
@@ -544,9 +505,9 @@ dependencies = [
[[package]]
name = "clap"
version = "4.5.11"
version = "4.5.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "35723e6a11662c2afb578bcf0b88bf6ea8e21282a953428f240574fcc3a2b5b3"
checksum = "0fbb260a053428790f3de475e304ff84cdbc4face759ea7a3e64c1edd938a7fc"
dependencies = [
"clap_builder",
"clap_derive",
@@ -554,9 +515,9 @@ dependencies = [
[[package]]
name = "clap_builder"
version = "4.5.11"
version = "4.5.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49eb96cbfa7cfa35017b7cd548c75b14c3118c98b423041d70562665e07fb0fa"
checksum = "64b17d7ea74e9f833c7dbf2cbe4fb12ff26783eda4782a8975b72f895c9b4d99"
dependencies = [
"anstyle",
"clap_lex",
@@ -564,9 +525,9 @@ dependencies = [
[[package]]
name = "clap_derive"
version = "4.5.11"
version = "4.5.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d029b67f89d30bbb547c89fd5161293c0aec155fc691d7924b64550662db93e"
checksum = "501d359d5f3dcaf6ecdeee48833ae73ec6e42723a1e52419c79abf9507eec0a0"
dependencies = [
"heck 0.5.0",
"proc-macro2",
@@ -639,7 +600,7 @@ dependencies = [
name = "conduit_api"
version = "0.4.6"
dependencies = [
"axum 0.7.5",
"axum",
"axum-client-ip",
"axum-extra",
"base64 0.22.1",
@@ -650,10 +611,9 @@ dependencies = [
"const-str",
"futures-util",
"hmac",
"http 1.1.0",
"http",
"http-body-util",
"hyper 1.4.1",
"image",
"hyper",
"ipaddress",
"jsonwebtoken",
"log",
@@ -666,7 +626,6 @@ dependencies = [
"sha-1",
"tokio",
"tracing",
"webpage",
]
[[package]]
@@ -674,18 +633,20 @@ name = "conduit_core"
version = "0.4.6"
dependencies = [
"argon2",
"axum 0.7.5",
"arrayvec",
"axum",
"bytes",
"cargo_toml",
"checked_ops",
"chrono",
"clap",
"conduit_macros",
"const-str",
"ctor",
"either",
"figment",
"hardened_malloc-rs",
"http 1.1.0",
"http",
"http-body-util",
"image",
"ipaddress",
@@ -741,7 +702,7 @@ dependencies = [
name = "conduit_router"
version = "0.4.6"
dependencies = [
"axum 0.7.5",
"axum",
"axum-client-ip",
"axum-server",
"axum-server-dual-protocol",
@@ -751,9 +712,9 @@ dependencies = [
"conduit_core",
"conduit_service",
"const-str",
"http 1.1.0",
"http",
"http-body-util",
"hyper 1.4.1",
"hyper",
"hyper-util",
"log",
"ruma",
@@ -781,7 +742,7 @@ dependencies = [
"cyborgtime",
"futures-util",
"hickory-resolver",
"http 1.1.0",
"http",
"image",
"ipaddress",
"itertools 0.13.0",
@@ -802,13 +763,14 @@ dependencies = [
"tokio",
"tracing",
"url",
"webpage",
]
[[package]]
name = "console-api"
version = "0.7.0"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a257c22cd7e487dd4a13d413beabc512c5052f0bc048db0da6a84c3d8a6142fd"
checksum = "86ed14aa9c9f927213c6e4f3ef75faaad3406134efe84ba2cb7983431d5f0931"
dependencies = [
"futures-core",
"prost",
@@ -819,9 +781,9 @@ dependencies = [
[[package]]
name = "console-subscriber"
version = "0.3.0"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "31c4cc54bae66f7d9188996404abdf7fdfa23034ef8e43478c8810828abad758"
checksum = "e2e3a111a37f3333946ebf9da370ba5c5577b18eb342ec683eb488dd21980302"
dependencies = [
"console-api",
"crossbeam-channel",
@@ -829,6 +791,7 @@ dependencies = [
"futures-task",
"hdrhistogram",
"humantime",
"hyper-util",
"prost",
"prost-types",
"serde",
@@ -1372,25 +1335,6 @@ version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b"
[[package]]
name = "h2"
version = "0.3.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "81fe527a889e1532da5c525686d96d4c2e74cdd345badf8dfef9f6b39dd5f5e8"
dependencies = [
"bytes",
"fnv",
"futures-core",
"futures-sink",
"futures-util",
"http 0.2.12",
"indexmap 2.2.6",
"slab",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "h2"
version = "0.4.5"
@@ -1402,8 +1346,8 @@ dependencies = [
"fnv",
"futures-core",
"futures-sink",
"http 1.1.0",
"indexmap 2.2.6",
"http",
"indexmap 2.3.0",
"slab",
"tokio",
"tokio-util",
@@ -1450,7 +1394,7 @@ dependencies = [
"base64 0.21.7",
"bytes",
"headers-core",
"http 1.1.0",
"http",
"httpdate",
"mime",
"sha1",
@@ -1462,7 +1406,7 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4"
dependencies = [
"http 1.1.0",
"http",
]
[[package]]
@@ -1579,17 +1523,6 @@ dependencies = [
"syn 2.0.72",
]
[[package]]
name = "http"
version = "0.2.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1"
dependencies = [
"bytes",
"fnv",
"itoa",
]
[[package]]
name = "http"
version = "1.1.0"
@@ -1610,17 +1543,6 @@ dependencies = [
"memchr",
]
[[package]]
name = "http-body"
version = "0.4.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2"
dependencies = [
"bytes",
"http 0.2.12",
"pin-project-lite",
]
[[package]]
name = "http-body"
version = "1.0.1"
@@ -1628,7 +1550,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
dependencies = [
"bytes",
"http 1.1.0",
"http",
]
[[package]]
@@ -1639,8 +1561,8 @@ checksum = "793429d76616a256bcb62c2a2ec2bed781c8307e797e2598c50010f2bee2544f"
dependencies = [
"bytes",
"futures-util",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"pin-project-lite",
]
@@ -1662,30 +1584,6 @@ version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
[[package]]
name = "hyper"
version = "0.14.30"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a152ddd61dfaec7273fe8419ab357f33aee0d914c5f4efbf0d96fa749eea5ec9"
dependencies = [
"bytes",
"futures-channel",
"futures-core",
"futures-util",
"h2 0.3.26",
"http 0.2.12",
"http-body 0.4.6",
"httparse",
"httpdate",
"itoa",
"pin-project-lite",
"socket2",
"tokio",
"tower-service",
"tracing",
"want",
]
[[package]]
name = "hyper"
version = "1.4.1"
@@ -1695,9 +1593,9 @@ dependencies = [
"bytes",
"futures-channel",
"futures-util",
"h2 0.4.5",
"http 1.1.0",
"http-body 1.0.1",
"h2",
"http",
"http-body",
"httparse",
"httpdate",
"itoa",
@@ -1714,8 +1612,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ee4be2c948921a1a5320b629c4193916ed787a7f7f293fd3f7f5a6c9de74155"
dependencies = [
"futures-util",
"http 1.1.0",
"hyper 1.4.1",
"http",
"hyper",
"hyper-util",
"rustls 0.23.12",
"rustls-native-certs",
@@ -1728,14 +1626,15 @@ dependencies = [
[[package]]
name = "hyper-timeout"
version = "0.4.1"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1"
checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793"
dependencies = [
"hyper 0.14.30",
"hyper",
"hyper-util",
"pin-project-lite",
"tokio",
"tokio-io-timeout",
"tower-service",
]
[[package]]
@@ -1747,9 +1646,9 @@ dependencies = [
"bytes",
"futures-channel",
"futures-util",
"http 1.1.0",
"http-body 1.0.1",
"hyper 1.4.1",
"http",
"http-body",
"hyper",
"pin-project-lite",
"socket2",
"tokio",
@@ -1817,9 +1716,9 @@ dependencies = [
[[package]]
name = "indexmap"
version = "2.2.6"
version = "2.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "168fb715dda47215e360912c096649d23d58bf392ac62f73919e831745e40f26"
checksum = "de3fc2e30ba82dd1b3911c8de1ffc143c74a914a14e99514d7637e3099df5ea0"
dependencies = [
"equivalent",
"hashbrown 0.14.5",
@@ -2356,7 +2255,7 @@ checksum = "1e32339a5dc40459130b3bd269e9892439f55b33e772d2a9d402a789baaf4e8a"
dependencies = [
"futures-core",
"futures-sink",
"indexmap 2.2.6",
"indexmap 2.3.0",
"js-sys",
"once_cell",
"pin-project-lite",
@@ -2403,7 +2302,7 @@ dependencies = [
"glob",
"once_cell",
"opentelemetry",
"ordered-float 4.2.1",
"ordered-float 4.2.2",
"percent-encoding",
"rand",
"thiserror",
@@ -2422,9 +2321,9 @@ dependencies = [
[[package]]
name = "ordered-float"
version = "4.2.1"
version = "4.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "19ff2cf528c6c03d9ed653d6c4ce1dc0582dc4af309790ad92f07c1cd551b0be"
checksum = "4a91171844676f8c7990ce64959210cd2eaef32c2612c50f9fae9f8aaa6065a6"
dependencies = [
"num-traits",
]
@@ -2651,9 +2550,12 @@ checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
[[package]]
name = "ppv-lite86"
version = "0.2.17"
version = "0.2.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
checksum = "dee4364d9f3b902ef14fab8a1ddffb783a1cb6b4bba3bfc1fa3922732c7de97f"
dependencies = [
"zerocopy",
]
[[package]]
name = "precomputed-hash"
@@ -2694,9 +2596,9 @@ dependencies = [
[[package]]
name = "prost"
version = "0.12.6"
version = "0.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "deb1435c188b76130da55f17a466d252ff7b1418b2ad3e037d127b94e3411f29"
checksum = "e13db3d3fde688c61e2446b4d843bc27a7e8af269a69440c0308021dc92333cc"
dependencies = [
"bytes",
"prost-derive",
@@ -2704,12 +2606,12 @@ dependencies = [
[[package]]
name = "prost-derive"
version = "0.12.6"
version = "0.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "81bddcdb20abf9501610992b6759a4c888aef7d1a7247ef75e2404275ac24af1"
checksum = "18bec9b0adc4eba778b33684b7ba3e7137789434769ee3ce3930463ef904cfca"
dependencies = [
"anyhow",
"itertools 0.12.1",
"itertools 0.13.0",
"proc-macro2",
"quote",
"syn 2.0.72",
@@ -2717,9 +2619,9 @@ dependencies = [
[[package]]
name = "prost-types"
version = "0.12.6"
version = "0.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9091c90b0a32608e984ff2fa4091273cbdd755d54935c51d520887f4a1dbd5b0"
checksum = "cee5168b05f49d4b0ca581206eb14a7b22fafd963efe729ac48eb03266e25cc2"
dependencies = [
"prost",
]
@@ -2904,12 +2806,12 @@ dependencies = [
"futures-channel",
"futures-core",
"futures-util",
"h2 0.4.5",
"h2",
"hickory-resolver",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"http-body-util",
"hyper 1.4.1",
"hyper",
"hyper-rustls",
"hyper-util",
"ipnet",
@@ -3009,7 +2911,7 @@ dependencies = [
"assign",
"bytes",
"date_header",
"http 1.1.0",
"http",
"js_int",
"js_option",
"maplit",
@@ -3032,8 +2934,8 @@ dependencies = [
"base64 0.22.1",
"bytes",
"form_urlencoded",
"http 1.1.0",
"indexmap 2.2.6",
"http",
"indexmap 2.3.0",
"js_int",
"konst",
"percent-encoding",
@@ -3059,7 +2961,7 @@ version = "0.28.1"
source = "git+https://github.com/girlbossceo/ruwuma?rev=c76e2873c1593a3308d4ba3e0e4a1db65acf8536#c76e2873c1593a3308d4ba3e0e4a1db65acf8536"
dependencies = [
"as_variant",
"indexmap 2.2.6",
"indexmap 2.3.0",
"js_int",
"js_option",
"percent-encoding",
@@ -3140,7 +3042,7 @@ version = "0.3.0"
source = "git+https://github.com/girlbossceo/ruwuma?rev=c76e2873c1593a3308d4ba3e0e4a1db65acf8536#c76e2873c1593a3308d4ba3e0e4a1db65acf8536"
dependencies = [
"headers",
"http 1.1.0",
"http",
"http-auth",
"ruma-common",
"thiserror",
@@ -3519,7 +3421,7 @@ version = "0.34.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c90802b38c899a2c9e557dff25ad186362eddf755d5f5244001b172dd03bead"
dependencies = [
"http 1.1.0",
"http",
"pin-project",
"sentry-core",
"tower-layer",
@@ -3583,7 +3485,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8de514ef58196f1fc96dcaef80fe6170a1ce6215df9687a93fe8300e773fefc5"
dependencies = [
"form_urlencoded",
"indexmap 2.2.6",
"indexmap 2.3.0",
"itoa",
"ryu",
"serde",
@@ -3591,11 +3493,12 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.120"
version = "1.0.121"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e0d21c9a8cae1235ad58a00c11cb40d4b1e5c784f1ef2c537876ed6ffd8b7c5"
checksum = "4ab380d7d9f22ef3f21ad3e6c1ebe8e4fc7a2000ccba2e4d71fc96f15b2cb609"
dependencies = [
"itoa",
"memchr",
"ryu",
"serde",
]
@@ -3647,7 +3550,7 @@ version = "0.9.34+deprecated"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
dependencies = [
"indexmap 2.2.6",
"indexmap 2.3.0",
"itoa",
"ryu",
"serde",
@@ -3714,9 +3617,9 @@ dependencies = [
[[package]]
name = "signal-hook-mio"
version = "0.2.3"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29ad2e15f37ec9a6cc544097b78a1ec90001e9f71b81338ca39f430adaca99af"
checksum = "34db1a06d485c9142248b7a054f034b349b212551f3dfd19c94d45a754a217cd"
dependencies = [
"libc",
"mio 0.8.11",
@@ -4055,9 +3958,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tokio"
version = "1.39.1"
version = "1.39.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d040ac2b29ab03b09d4129c2f5bbd012a3ac2f79d38ff506a4bf8dd34b0eac8a"
checksum = "daa4fb1bc778bd6f04cbfc4bb2d06a7396a8f299dc33ea1900cedaa316f467b1"
dependencies = [
"backtrace",
"bytes",
@@ -4071,16 +3974,6 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "tokio-io-timeout"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf"
dependencies = [
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-macros"
version = "2.4.0"
@@ -4127,9 +4020,9 @@ dependencies = [
[[package]]
name = "tokio-socks"
version = "0.5.1"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51165dfa029d2a65969413a6cc96f354b86b464498702f174a4efa13608fd8c0"
checksum = "0d4770b8024672c1101b3f6733eab95b18007dbe0847a8afe341fcf79e06043f"
dependencies = [
"either",
"futures-util",
@@ -4163,21 +4056,21 @@ dependencies = [
[[package]]
name = "toml"
version = "0.8.16"
version = "0.8.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "81967dd0dd2c1ab0bc3468bd7caecc32b8a4aa47d0c8c695d8c2b2108168d62c"
checksum = "a1ed1f98e3fdc28d6d910e6737ae6ab1a93bf1985935a1193e68f93eeb68d24e"
dependencies = [
"serde",
"serde_spanned",
"toml_datetime",
"toml_edit 0.22.17",
"toml_edit 0.22.20",
]
[[package]]
name = "toml_datetime"
version = "0.6.7"
version = "0.6.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8fb9f64314842840f1d940ac544da178732128f1c78c21772e876579e0da1db"
checksum = "0dd7358ecb8fc2f8d014bf86f6f638ce72ba252a2c3a2572f2a795f1d23efb41"
dependencies = [
"serde",
]
@@ -4188,43 +4081,46 @@ version = "0.21.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a8534fd7f78b5405e860340ad6575217ce99f38d4d5c8f2442cb5ecb50090e1"
dependencies = [
"indexmap 2.2.6",
"indexmap 2.3.0",
"toml_datetime",
"winnow 0.5.40",
]
[[package]]
name = "toml_edit"
version = "0.22.17"
version = "0.22.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d9f8729f5aea9562aac1cc0441f5d6de3cff1ee0c5d67293eeca5eb36ee7c16"
checksum = "583c44c02ad26b0c3f3066fe629275e50627026c51ac2e595cca4c230ce1ce1d"
dependencies = [
"indexmap 2.2.6",
"indexmap 2.3.0",
"serde",
"serde_spanned",
"toml_datetime",
"winnow 0.6.16",
"winnow 0.6.18",
]
[[package]]
name = "tonic"
version = "0.11.0"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76c4eb7a4e9ef9d4763600161f12f5070b92a578e1b634db88a6887844c91a13"
checksum = "38659f4a91aba8598d27821589f5db7dddd94601e7a01b1e485a50e5484c7401"
dependencies = [
"async-stream",
"async-trait",
"axum 0.6.20",
"base64 0.21.7",
"axum",
"base64 0.22.1",
"bytes",
"h2 0.3.26",
"http 0.2.12",
"http-body 0.4.6",
"hyper 0.14.30",
"h2",
"http",
"http-body",
"http-body-util",
"hyper",
"hyper-timeout",
"hyper-util",
"percent-encoding",
"pin-project",
"prost",
"socket2",
"tokio",
"tokio-stream",
"tower",
@@ -4264,8 +4160,8 @@ dependencies = [
"bytes",
"futures-core",
"futures-util",
"http 1.1.0",
"http-body 1.0.1",
"http",
"http-body",
"http-body-util",
"pin-project-lite",
"tokio",
@@ -4881,9 +4777,9 @@ dependencies = [
[[package]]
name = "winnow"
version = "0.6.16"
version = "0.6.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b480ae9340fc261e6be3e95a1ba86d54ae3f9171132a73ce8d4bbaf68339507c"
checksum = "68a9bda4691f099d435ad181000724da8e5899daa10713c2d432552b9ccd3a6f"
dependencies = [
"memchr",
]
@@ -4925,6 +4821,27 @@ version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049"
[[package]]
name = "zerocopy"
version = "0.6.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "854e949ac82d619ee9a14c66a1b674ac730422372ccb759ce0c39cabcf2bf8e6"
dependencies = [
"byteorder",
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
version = "0.6.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "125139de3f6b9d625c39e2efdd73d41bdac468ccd556556440e322be0e1bbd91"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.72",
]
[[package]]
name = "zeroize"
version = "1.8.1"
+32 -4
View File
@@ -25,6 +25,9 @@ version = "0.4.6"
[workspace.metadata.crane]
name = "conduit"
[workspace.dependencies.arrayvec]
version = "0.7.4"
[workspace.dependencies.const-str]
version = "0.5.7"
@@ -33,6 +36,7 @@ version = "0.2.8"
[workspace.dependencies.cargo_toml]
version = "0.20"
default-features = false
features = ["features"]
[workspace.dependencies.toml]
@@ -96,6 +100,7 @@ features = ["typed-header", "tracing"]
[workspace.dependencies.axum-server]
version = "0.6.0"
default-features = false
features = ["tls-rustls"]
[workspace.dependencies.axum-client-ip]
@@ -103,10 +108,12 @@ version = "0.6.0"
[workspace.dependencies.tower]
version = "0.4.13"
default-features = false
features = ["util"]
[workspace.dependencies.tower-http]
version = "0.5.2"
default-features = false
features = [
"add-extension",
"cors",
@@ -129,10 +136,12 @@ features = [
[workspace.dependencies.serde]
version = "1.0.204"
default-features = false
features = ["rc"]
[workspace.dependencies.serde_json]
version = "1.0.120"
version = "1.0.121"
default-features = false
features = ["raw_value"]
# Used for appservice registration files
@@ -200,7 +209,8 @@ version = "0.3.30"
default-features = false
[workspace.dependencies.tokio]
version = "1.39.1"
version = "1.39.2"
default-features = false
features = [
"fs",
"net",
@@ -221,6 +231,7 @@ version = "0.8.5"
# Validating urls in config, was already a transitive dependency
[workspace.dependencies.url]
version = "2.5.0"
default-features = false
features = ["serde"]
# standard date and time tools
@@ -231,6 +242,7 @@ default-features = false
[workspace.dependencies.hyper]
version = "1.4.1"
default-features = false
features = [
"server",
"http1",
@@ -239,6 +251,7 @@ features = [
[workspace.dependencies.hyper-util]
version = "0.1.6"
default-features = false
features = [
"client",
"server-auto",
@@ -250,11 +263,13 @@ features = [
# to support multiple variations of setting a config option
[workspace.dependencies.either]
version = "1.11.0"
default-features = false
features = ["serde"]
# Used for reading the configuration from conduwuit.toml & environment variables
[workspace.dependencies.figment]
version = "0.10.18"
default-features = false
features = ["env", "toml"]
[workspace.dependencies.hickory-resolver]
@@ -396,10 +411,11 @@ default-features = false
features = ["use_std"]
[workspace.dependencies.console-subscriber]
version = "0.3"
version = "0.4"
[workspace.dependencies.nix]
version = "0.29.0"
default-features = false
features = ["resource"]
[workspace.dependencies.sd-notify]
@@ -427,6 +443,7 @@ version = "0.1"
[workspace.dependencies.syn]
version = "2.0.72"
default-features = false
features = ["full", "extra-traits"]
[workspace.dependencies.quote]
@@ -555,7 +572,17 @@ lto = "fat"
[profile.release-max-perf.build-override]
inherits = "release-max-perf"
opt-level = 0
codegen-units = 32
#rustflags = [
# '-Crelocation-model=pic',
# '-Ctarget-feature=-crt-static',
# '-Clink-arg=-Wl,--no-gc-sections',
#]
[profile.release-max-perf.package.conduit_macros]
inherits = "release-max-perf.build-override"
#rustflags = [
# '-Crelocation-model=pic',
# '-Ctarget-feature=-crt-static',
#]
@@ -801,7 +828,7 @@ perf = { level = "warn", priority = -1 }
###################
#restriction = "warn"
allow_attributes = "warn"
#allow_attributes = "warn" # UNSTABLE
arithmetic_side_effects = "warn"
as_conversions = "warn"
as_underscore = "warn"
@@ -851,6 +878,7 @@ style = { level = "warn", priority = -1 }
## some sadness
# trivial assertions are quite alright
assertions_on_constants = { level = "allow", priority = 1 }
module_inception = { level = "allow", priority = 1 }
###################
suspicious = { level = "warn", priority = -1 }
+25
View File
@@ -514,6 +514,31 @@ allow_profile_lookup_federation_requests = true
# Defaults to false as this uses more CPU when compressing.
#rocksdb_bottommost_compression = false
# Level of statistics collection. Some admin commands to display database statistics may require
# this option to be set. Database performance may be impacted by higher settings.
#
# Option is a number ranging from 0 to 6:
# 0 = No statistics.
# 1 = No statistics in release mode (default).
# 2 to 3 = Statistics with no performance impact.
# 3 to 5 = Statistics with possible performance impact.
# 6 = All statistics.
#
# Defaults to 1 (No statistics, except in debug-mode)
#rocksdb_stats_level = 1
# Database repair mode (for RocksDB SST corruption)
#
# Use this option when the server reports corruption while running or panics. If the server refuses
# to start use the recovery mode options first. Corruption errors containing the acronym 'SST' which
# occur after startup will likely require this option.
#
# - Backing up your database directory is recommended prior to running the repair.
# - Disabling repair mode and restarting the server is recommended after running the repair.
#
# Defaults to false
#rocksdb_repair = false
# Database recovery mode (for RocksDB WAL corruption)
#
# Use this option when the server reports corruption and refuses to start. Set mode 2 (PointInTime)
-1
View File
@@ -16,7 +16,6 @@ case "$1" in
--home "$CONDUWUIT_DATABASE_PATH" \
--disabled-login \
--shell "/usr/sbin/nologin" \
--verbose \
conduwuit
fi
+1
View File
@@ -10,6 +10,7 @@
- [Docker](deploying/docker.md)
- [Arch Linux](deploying/arch-linux.md)
- [Debian](deploying/debian.md)
- [FreeBSD](deploying/freebsd.md)
- [TURN](turn.md)
- [Appservices](appservices.md)
- [Maintenance](maintenance.md)
+24 -4
View File
@@ -4,15 +4,35 @@ This chapter describes various ways to configure conduwuit.
## Basics
Conduwuit uses a config file for the majority of the settings. Please refer to the
[example config file](./configuration/examples.md#example-configuration) for all of those settings.
The config file to use can either be specified on the command line when running conduwuit by specifying the
conduwuit uses a config file for the majority of the settings, but also supports setting individual config options via commandline.
Please refer to the [example config file](./configuration/examples.md#example-configuration) for all of those settings.
The config file to use can be specified on the commandline when running conduwuit by specifying the
`-c`, `--config` flag. Alternatively, you can use the environment variable `CONDUWUIT_CONFIG` to specify the config
file to used.
file to used. Conduit's environment variables are supported for backwards compatibility.
## Option commandline flag
conduwuit supports setting individual config options in TOML format from the `-O` / `--option` flag. For example, you can set your server name via `-O server_name=\"example.com\"`.
Note that the config is parsed as TOML, and shells like bash will remove quotes. So unfortunately it is required to escape quotes if the config option takes a string.
This does not apply to options that take booleans or numbers:
- `--option allow_registration=true` works ✅
- `-O max_request_size=99999999` works ✅
- `-O server_name=example.com` does not work ❌
- `--option log=\"debug\"` works ✅
- `--option server_name='"example.com'"` works ✅
## Environment variables
All of the settings that are found in the config file can be specified by using environment variables.
The environment variable names should be all caps and prefixed with `CONDUWUIT_`.
For example, if the setting you are changing is `max_request_size`, then the environment variable to set is
`CONDUWUIT_MAX_REQUEST_SIZE`.
To modify config options not in the `[global]` context such as `[global.well_known]`, use the `__` suffix split: `CONDUWUIT_WELL_KNOWN__SERVER`
Conduit's environment variables are supported for backwards compatibility (e.g. `CONDUIT_SERVER_NAME`).
+2 -2
View File
@@ -17,9 +17,9 @@ OCI images for conduwuit are available in the registries listed below.
| GitLab Registry | [registry.gitlab.com/conduwuit/conduwuit:main][gl] | ![Image Size][shield-main] | Stable main branch. |
| Docker Hub | [docker.io/girlbossceo/conduwuit:main][dh] | ![Image Size][shield-main] | Stable main branch. |
[dh]: https://hub.docker.com/repository/docker/girlbossceo/conduwuit
[dh]: https://hub.docker.com/r/girlbossceo/conduwuit
[gh]: https://github.com/girlbossceo/conduwuit/pkgs/container/conduwuit
[gl]: https://gitlab.com/conduwuit/conduwuit/container_registry/6351657
[gl]: https://gitlab.com/conduwuit/conduwuit/container_registry/6369729
[shield-latest]: https://img.shields.io/docker/image-size/girlbossceo/conduwuit/latest
[shield-main]: https://img.shields.io/docker/image-size/girlbossceo/conduwuit/main
+8
View File
@@ -0,0 +1,8 @@
# conduwuit for FreeBSD
conduwuit at the moment does not provide FreeBSD builds. Building conduwuit on FreeBSD requires a specific environment variable to use the
system prebuilt RocksDB library instead of rust-rocksdb / rust-librocksdb-sys which does *not* work and will cause a build error or coredump.
Use the following environment variable: `ROCKSDB_LIB_DIR=/usr/local/lib`
Such example commandline with it can be: `ROCKSDB_LIB_DIR=/usr/local/lib cargo build --release`
+2 -4
View File
@@ -9,14 +9,12 @@
You may simply download the binary that fits your machine. Run `uname -m` to see what you need.
Prebuilt binaries can be downloaded from the latest tagged release [here](https://github.com/girlbossceo/conduwuit/releases/latest).
The latest tagged release also includes the Debian packages.
Prebuilt fully static musl binaries can be downloaded from the latest tagged release [here](https://github.com/girlbossceo/conduwuit/releases/latest) or `main` CI branch workflow artifact output. These also include Debian packages. These binaries have jemalloc and io_uring statically linked and included with them.
Alternatively, you may compile the binary yourself. We recommend using [Lix](https://lix.systems) to build conduwuit as this has the most guaranteed
reproducibiltiy and easiest to get a build environment and output going.
Otherwise, follow standard Rust project build guides (installing git and cloning the repo, getting the Rust toolchain via rustup, installing LLVM toolchain + libclang, installing liburing for io_uring and RocksDB, etc).
Otherwise, follow standard Rust project build guides (installing git and cloning the repo, getting the Rust toolchain via rustup, installing LLVM toolchain + libclang for RocksDB, installing liburing for io_uring and RocksDB, etc).
## Adding a conduwuit user
+11 -2
View File
@@ -8,6 +8,17 @@
>
> If there are things like Compose file issues or Dockerhub image issues, those can still be mentioned as long as they're something we can fix.
## General potential issues
#### Potential DNS issues when using Docker
Docker has issues with its default DNS setup that may cause DNS to not be properly functional when running conduwuit, resulting in federation issues.
The symptoms of this have shown in excessively long room joins (30+ minutes) from very long DNS timeouts, log entries of "mismatching responding nameservers", and/or partial or non-functional inbound/outbound federation.
This is **not** a conduwuit issue, and is purely a Docker issue. It is not sustainable for heavy DNS activity which is normal for Matrix federation. The workarounds for this are:
- Use DNS over TCP via the config option `query_over_tcp_only = true`
- Don't use Docker's default DNS setup and instead allow the container to use and communicate with your host's DNS servers (host's `/etc/resolv.conf`)
## Rocksdb / database issues
#### Direct IO
@@ -38,8 +49,6 @@ With this in mind:
- Leave your conduwuit running in `PointInTime` for at least 30-60 minutes so as much possible corruption is restored
- If all goes will, you should be able to restore back to using `TolerateCorruptedTailRecords` and you have successfully recovered your database
## Media
## Debugging
Note that users should not really be debugging things. If you find yourself debugging and find the issue, please let us know and/or how we can fix it. Various debug commands can be found in `!admin debug`.
Generated
+3 -3
View File
@@ -81,11 +81,11 @@
"complement": {
"flake": false,
"locked": {
"lastModified": 1720637557,
"narHash": "sha256-oZz6nCmFmdJZpC+K1iOG2KkzTI6rlAmndxANPDVU7X0=",
"lastModified": 1722323564,
"narHash": "sha256-6w6/N8walz4Ayc9zu7iySqJRmGFukhkaICLn4dweAcA=",
"owner": "matrix-org",
"repo": "complement",
"rev": "0d14432e010482ea9e13a6f7c47c1533c0c9d62f",
"rev": "6e4426a9e63233f9821a4d2382bfed145244183f",
"type": "github"
},
"original": {
+4
View File
@@ -137,7 +137,11 @@
# Useful for editing the book locally
mdbook
# used for rust caching in CI to speed it up
sccache
# needed so we can get rid of gcc and other unused deps that bloat OCI images
removeReferencesTo
])
++ scope.main.buildInputs
++ scope.main.propagatedBuildInputs
+15 -3
View File
@@ -7,6 +7,7 @@
, liburing
, pkgsBuildHost
, rocksdb
, removeReferencesTo
, rust
, rust-jemalloc-sys
, stdenv
@@ -94,8 +95,8 @@ buildDepsOnlyEnv =
else if stdenv.targetPlatform.isAarch64
then lib.subtractLists [ "-DPORTABLE=1" ] old.cmakeFlags
++ lib.optionals stdenv.targetPlatform.isAarch64 [
# cortex-a55 == ARMv8.2-a
"-DPORTABLE=armv8.2-a"
# cortex-a73 == ARMv8-A
"-DPORTABLE=armv8-a"
]
else old.cmakeFlags;
});
@@ -128,7 +129,7 @@ buildPackageEnv = {
+ lib.optionalString stdenv.targetPlatform.isx86_64
" -Ctarget-cpu=x86-64-v2"
+ lib.optionalString stdenv.targetPlatform.isAarch64
" -Ctarget-cpu=cortex-a55"; # cortex-a55 == ARMv8.2-a
" -Ctarget-cpu=cortex-a73"; # cortex-a73 == ARMv8-A
};
@@ -154,6 +155,7 @@ commonAttrs = {
};
dontStrip = profile == "dev" || profile == "test";
dontPatchELF = profile == "dev" || profile == "test";
buildInputs = lib.optional (featureEnabled "jemalloc") rust-jemalloc-sys';
@@ -168,6 +170,9 @@ commonAttrs = {
# differing values for `NIX_CFLAGS_COMPILE`, which contributes to spurious
# rebuilds of bindgen and its depedents.
jq
# needed so we can get rid of gcc and other unused deps that bloat OCI images
removeReferencesTo
]
++ lib.optionals stdenv.isDarwin [
# https://github.com/NixOS/nixpkgs/issues/206242
@@ -177,6 +182,13 @@ commonAttrs = {
# https://discourse.nixos.org/t/compile-a-rust-binary-on-macos-dbcrossbar/8612
pkgsBuildHost.darwin.apple_sdk.frameworks.Security
];
# for some reason gcc and other weird deps are added to OCI images and bloats it up
#
# <https://github.com/input-output-hk/haskell.nix/issues/829>
postInstall = with pkgsBuildHost; ''
find "$out" -type f -exec remove-references-to -t ${stdenv.cc} -t ${gcc} -t ${libgcc} -t ${linuxHeaders} -t ${libidn2} -t ${libunistring} '{}' +
'';
};
in
+4 -1
View File
@@ -1,6 +1,9 @@
use service::Services;
use std::time::SystemTime;
use conduit_service::Services;
pub(crate) struct Command<'a> {
pub(crate) services: &'a Services,
pub(crate) body: &'a [&'a str],
pub(crate) timer: SystemTime,
}
+43 -37
View File
@@ -1,16 +1,12 @@
use std::{
collections::{BTreeMap, HashMap},
fmt::Write,
sync::{Arc, Mutex},
sync::Arc,
time::{Instant, SystemTime},
};
use api::client::validate_and_add_event_id;
use conduit::{
debug, info, log,
log::{capture, Capture},
utils, warn, Error, PduEvent, Result,
};
use conduit::{debug, debug_error, err, info, trace, utils, warn, Error, PduEvent, Result};
use ruma::{
api::{client::error::ErrorKind, federation::event::get_room_state},
events::room::message::RoomMessageEventContent,
@@ -149,23 +145,32 @@ pub(super) async fn get_remote_pdu_list(
.filter_map(|pdu| EventId::parse(pdu).ok())
.collect::<Vec<_>>();
let mut failed_count: usize = 0;
let mut success_count: usize = 0;
for pdu in list {
if force {
if let Err(e) = self.get_remote_pdu(Box::from(pdu), server.clone()).await {
failed_count = failed_count.saturating_add(1);
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!(
"Failed to get remote PDU, ignoring error: {e}"
)))
.await;
warn!(%e, "Failed to get remote PDU, ignoring error");
warn!("Failed to get remote PDU, ignoring error: {e}");
} else {
success_count = success_count.saturating_add(1);
}
} else {
self.get_remote_pdu(Box::from(pdu), server.clone()).await?;
success_count = success_count.saturating_add(1);
}
}
Ok(RoomMessageEventContent::text_plain("Fetched list of remote PDUs."))
Ok(RoomMessageEventContent::text_plain(format!(
"Fetched {success_count} remote PDUs successfully with {failed_count} failures"
)))
}
#[admin_command]
@@ -204,7 +209,7 @@ pub(super) async fn get_remote_pdu(
Error::BadRequest(ErrorKind::Unknown, "Received response from server but failed to parse PDU")
})?;
debug!("Attempting to parse PDU: {:?}", &response.pdu);
trace!("Attempting to parse PDU: {:?}", &response.pdu);
let parsed_pdu = {
let parsed_result = self
.services
@@ -229,8 +234,7 @@ pub(super) async fn get_remote_pdu(
debug!("Attempting to fetch homeserver signing keys for {server}");
self.services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys(parsed_pdu.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
.await
.unwrap_or_else(|e| {
@@ -443,8 +447,7 @@ pub(super) async fn verify_json(&self) -> Result<RoomMessageEventContent> {
let pub_key_map = RwLock::new(BTreeMap::new());
self.services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys([&value], &pub_key_map)
.await?;
@@ -562,8 +565,7 @@ pub(super) async fn force_set_room_state_from_server(
info!("Fetching required signing keys for all the state events we got");
self.services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys(events.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
.await?;
@@ -578,8 +580,8 @@ pub(super) async fn force_set_room_state_from_server(
};
let pdu = PduEvent::from_id_val(&event_id, value.clone()).map_err(|e| {
warn!("Invalid PDU in fetching remote room state PDUs response: {} {:?}", e, value);
Error::BadServerResponse("Invalid PDU in send_join response.")
debug_error!("Invalid PDU in fetching remote room state PDUs response: {value:#?}");
err!(BadServerResponse(debug_error!("Invalid PDU in send_join response: {e:?}")))
})?;
self.services
@@ -682,8 +684,7 @@ pub(super) async fn get_verify_keys(
let signature_ids: Vec<String> = Vec::new();
let keys = self
.services
.rooms
.event_handler
.server_keys
.fetch_signing_keys_for_server(&server_name, signature_ids)
.await?;
@@ -712,30 +713,14 @@ pub(super) async fn resolve_true_destination(
));
}
let filter: &capture::Filter = &|data| {
data.level() <= log::Level::DEBUG
&& data.mod_name().starts_with("conduit")
&& matches!(data.span_name(), "actual" | "well-known" | "srv")
};
let state = &self.services.server.log.capture;
let logs = Arc::new(Mutex::new(String::new()));
let capture = Capture::new(state, Some(filter), capture::fmt_markdown(logs.clone()));
let capture_scope = capture.start();
let actual = self
.services
.resolver
.resolve_actual_dest(&server_name, !no_cache)
.await?;
drop(capture_scope);
let msg = format!(
"{}\nDestination: {}\nHostname URI: {}",
logs.lock().expect("locked"),
actual.dest,
actual.host,
);
let msg = format!("Destination: {}\nHostname URI: {}", actual.dest, actual.host,);
Ok(RoomMessageEventContent::text_markdown(msg))
}
@@ -823,3 +808,24 @@ pub(super) async fn list_dependencies(&self, names: bool) -> Result<RoomMessageE
Ok(RoomMessageEventContent::notice_markdown(out))
}
#[admin_command]
pub(super) async fn database_stats(
&self, property: Option<String>, map: Option<String>,
) -> Result<RoomMessageEventContent> {
let property = property.unwrap_or_else(|| "rocksdb.stats".to_owned());
let map_name = map.as_ref().map_or(utils::string::EMPTY, String::as_str);
let mut out = String::new();
for (name, map) in self.services.db.iter_maps() {
if !map_name.is_empty() && *map_name != *name {
continue;
}
let res = map.property(&property)?;
let res = res.trim();
writeln!(out, "##### {name}:\n```\n{res}\n```")?;
}
Ok(RoomMessageEventContent::notice_markdown(out))
}
+8
View File
@@ -184,6 +184,14 @@ pub(super) enum DebugCommand {
names: bool,
},
/// - Get database statistics
DatabaseStats {
property: Option<String>,
#[arg(short, long, alias("column"))]
map: Option<String>,
},
/// - Developer test stubs
#[command(subcommand)]
#[allow(non_snake_case)]
+85 -38
View File
@@ -1,7 +1,21 @@
use std::{panic::AssertUnwindSafe, sync::Arc, time::Instant};
use std::{
panic::AssertUnwindSafe,
sync::{Arc, Mutex},
time::SystemTime,
};
use clap::{CommandFactory, Parser};
use conduit::{error, trace, utils::string::common_prefix, Error, Result};
use conduit::{
debug, error,
log::{
capture,
capture::Capture,
fmt::{markdown_table, markdown_table_head},
},
trace,
utils::string::{collect_stream, common_prefix},
Error, Result,
};
use futures_util::future::FutureExt;
use ruma::{
events::{
@@ -11,9 +25,10 @@ use ruma::{
OwnedEventId,
};
use service::{
admin::{CommandInput, CommandOutput, CommandResult, HandlerResult},
admin::{CommandInput, CommandOutput, HandlerFuture, HandlerResult},
Services,
};
use tracing::Level;
use crate::{admin, admin::AdminCommand, Command};
@@ -21,12 +36,12 @@ use crate::{admin, admin::AdminCommand, Command};
pub(super) fn complete(line: &str) -> String { complete_command(AdminCommand::command(), line) }
#[must_use]
pub(super) fn handle(services: Arc<Services>, command: CommandInput) -> HandlerResult {
pub(super) fn handle(services: Arc<Services>, command: CommandInput) -> HandlerFuture {
Box::pin(handle_command(services, command))
}
#[tracing::instrument(skip_all, name = "admin")]
async fn handle_command(services: Arc<Services>, command: CommandInput) -> CommandResult {
async fn handle_command(services: Arc<Services>, command: CommandInput) -> HandlerResult {
AssertUnwindSafe(Box::pin(process_command(services, &command)))
.catch_unwind()
.await
@@ -34,13 +49,24 @@ async fn handle_command(services: Arc<Services>, command: CommandInput) -> Comma
.or_else(|error| handle_panic(&error, command))
}
async fn process_command(services: Arc<Services>, command: &CommandInput) -> CommandOutput {
process(services, &command.command)
async fn process_command(services: Arc<Services>, input: &CommandInput) -> CommandOutput {
let (command, args, body) = match parse(&services, input) {
Err(error) => return error,
Ok(parsed) => parsed,
};
let context = Command {
services: &services,
body: &body,
timer: SystemTime::now(),
};
process(&context, command, &args)
.await
.and_then(|content| reply(content, command.reply_id.clone()))
.and_then(|content| reply(content, input.reply_id.clone()))
}
fn handle_panic(error: &Error, command: CommandInput) -> CommandResult {
fn handle_panic(error: &Error, command: CommandInput) -> HandlerResult {
let link = "Please submit a [bug report](https://github.com/girlbossceo/conduwuit/issues/new). 🥺";
let msg = format!("Panic occurred while processing command:\n```\n{error:#?}\n```\n{link}");
let content = RoomMessageEventContent::notice_markdown(msg);
@@ -59,40 +85,61 @@ fn reply(mut content: RoomMessageEventContent, reply_id: Option<OwnedEventId>) -
}
// Parse and process a message from the admin room
async fn process(services: Arc<Services>, msg: &str) -> CommandOutput {
let mut lines = msg.lines().filter(|l| !l.trim().is_empty());
let command = lines.next().expect("each string has at least one line");
let (parsed, body) = match parse_command(command) {
Ok(parsed) => parsed,
Err(error) => {
let server_name = services.globals.server_name();
let message = error.replace("server.name", server_name.as_str());
return Some(RoomMessageEventContent::notice_markdown(message));
},
async fn process(context: &Command<'_>, command: AdminCommand, args: &[String]) -> CommandOutput {
let filter: &capture::Filter =
&|data| data.level() <= Level::DEBUG && data.our_modules() && data.scope.contains(&"admin");
let logs = Arc::new(Mutex::new(
collect_stream(|s| markdown_table_head(s)).expect("markdown table header"),
));
let capture = Capture::new(
&context.services.server.log.capture,
Some(filter),
capture::fmt(markdown_table, logs.clone()),
);
let capture_scope = capture.start();
let result = Box::pin(admin::process(command, context)).await;
drop(capture_scope);
debug!(
ok = result.is_ok(),
elapsed = ?context.timer.elapsed(),
command = ?args,
"command processed"
);
let logs = logs.lock().expect("locked");
let output = match result {
Err(error) => format!("{logs}\nEncountered an error while handling the command:\n```\n{error:#?}\n```"),
Ok(reply) => format!("{logs}\n{}", reply.body()), //TODO: content is recreated to add logs
};
let timer = Instant::now();
let body: Vec<&str> = body.iter().map(String::as_str).collect();
let context = Command {
services: &services,
body: &body,
};
let result = Box::pin(admin::process(parsed, &context)).await;
let elapsed = timer.elapsed();
conduit::debug!(?command, ok = result.is_ok(), "command processed in {elapsed:?}");
match result {
Ok(reply) => Some(reply),
Err(error) => Some(RoomMessageEventContent::notice_markdown(format!(
"Encountered an error while handling the command:\n```\n{error:#?}\n```"
))),
}
Some(RoomMessageEventContent::notice_markdown(output))
}
// Parse chat messages from the admin room into an AdminCommand object
fn parse_command(command_line: &str) -> Result<(AdminCommand, Vec<String>), String> {
let argv = parse_line(command_line);
let com = AdminCommand::try_parse_from(&argv).map_err(|error| error.to_string())?;
Ok((com, argv))
fn parse<'a>(
services: &Arc<Services>, input: &'a CommandInput,
) -> Result<(AdminCommand, Vec<String>, Vec<&'a str>), CommandOutput> {
let lines = input.command.lines().filter(|line| !line.trim().is_empty());
let command_line = lines.clone().next().expect("command missing first line");
let body = lines.skip(1).collect();
match parse_command(command_line) {
Ok((command, args)) => Ok((command, args, body)),
Err(error) => {
let message = error
.to_string()
.replace("server.name", services.globals.server_name().as_str());
Err(Some(RoomMessageEventContent::notice_markdown(message)))
},
}
}
fn parse_command(line: &str) -> Result<(AdminCommand, Vec<String>)> {
let argv = parse_line(line);
let command = AdminCommand::try_parse_from(&argv)?;
Ok((command, argv))
}
fn complete_command(mut cmd: clap::Command, line: &str) -> String {
+1 -1
View File
@@ -547,7 +547,7 @@ async fn list_banned_rooms(&self) -> Result<RoomMessageEventContent> {
rooms.reverse();
let output_plain = format!(
"Rooms Banned ({}):\n```\n{}```",
"Rooms Banned ({}):\n```\n{}\n```",
rooms.len(),
rooms
.iter()
+2 -2
View File
@@ -35,13 +35,13 @@ pub(super) async fn list_features(
};
if enabled && !available {
let features = info::rustc::features().join(delim);
let out = format!("```\n{features}\n```");
let out = format!("`\n{features}\n`");
return Ok(RoomMessageEventContent::text_markdown(out));
}
if available && !enabled {
let features = info::cargo::features().join(delim);
let out = format!("```\n{features}\n```");
let out = format!("`\n{features}\n`");
return Ok(RoomMessageEventContent::text_markdown(out));
}
-2
View File
@@ -47,7 +47,6 @@ hmac.workspace = true
http.workspace = true
http-body-util.workspace = true
hyper.workspace = true
image.workspace = true
ipaddress.workspace = true
jsonwebtoken.workspace = true
log.workspace = true
@@ -60,7 +59,6 @@ serde.workspace = true
sha-1.workspace = true
tokio.workspace = true
tracing.workspace = true
webpage.workspace = true
[lints]
workspace = true
+7 -3
View File
@@ -286,9 +286,13 @@ pub(crate) async fn register_route(
let token = utils::random_string(TOKEN_LENGTH);
// Create device for this account
services
.users
.create_device(&user_id, &device_id, &token, body.initial_device_display_name.clone())?;
services.users.create_device(
&user_id,
&device_id,
&token,
body.initial_device_display_name.clone(),
Some(client.to_string()),
)?;
debug_info!(%user_id, %device_id, "User account was created");
+3 -1
View File
@@ -21,11 +21,13 @@ pub(crate) async fn get_context_route(
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let sender_device = body.sender_device.as_ref().expect("user is authenticated");
// some clients, at least element, seem to require knowledge of redundant
// members for "inline" profiles on the timeline to work properly
let (lazy_load_enabled, lazy_load_send_redundant) = match &body.filter.lazy_load_options {
LazyLoadOptions::Enabled {
include_redundant_members,
} => (true, *include_redundant_members),
LazyLoadOptions::Disabled => (false, false),
LazyLoadOptions::Disabled => (false, cfg!(feature = "element_hacks")),
};
let mut lazy_loaded = HashSet::new();
+2 -3
View File
@@ -4,7 +4,7 @@ use std::{
};
use axum::extract::State;
use conduit::{debug, utils, utils::math::continue_exponential_backoff_secs, Error, Result};
use conduit::{utils, utils::math::continue_exponential_backoff_secs, Err, Error, Result};
use futures_util::{stream::FuturesUnordered, StreamExt};
use ruma::{
api::{
@@ -362,8 +362,7 @@ pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool + Send>(
const MIN: u64 = 5 * 60;
const MAX: u64 = 60 * 60 * 24;
if continue_exponential_backoff_secs(MIN, MAX, time.elapsed(), *tries) {
debug!("Backing off query from {server:?}");
return (server, Err(Error::BadServerResponse("bad query, still backing off")));
return (server, Err!(BadServerResponse("bad query from {server:?}, still backing off")));
}
}
+19 -254
View File
@@ -1,21 +1,18 @@
#![allow(deprecated)]
use std::{io::Cursor, time::Duration};
use std::time::Duration;
use axum::extract::State;
use axum_client_ip::InsecureClientIp;
use conduit::{
debug, debug_warn, error,
debug_warn, err, error,
utils::{
self,
content_disposition::{content_disposition_type, make_content_disposition, sanitise_filename},
math::ruma_from_usize,
},
warn, Error, Result,
warn, Err, Error, Result,
};
use image::io::Reader as ImgReader;
use ipaddress::IPAddress;
use reqwest::Url;
use ruma::api::client::{
error::{ErrorKind, RetryAfter},
media::{
@@ -24,16 +21,12 @@ use ruma::api::client::{
},
};
use service::{
media::{FileMeta, UrlPreviewData},
media::{FileMeta, MXC_LENGTH},
Services,
};
use webpage::HTML;
use crate::{Ruma, RumaResponse};
/// generated MXC ID (`media-id`) length
const MXC_LENGTH: usize = 32;
/// Cache control for immutable objects
const CACHE_CONTROL_IMMUTABLE: &str = "public,max-age=31536000,immutable";
@@ -76,12 +69,13 @@ pub(crate) async fn get_media_preview_route(
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let url = &body.url;
if !url_preview_allowed(&services, url) {
warn!(%sender_user, "URL is not allowed to be previewed: {url}");
return Err(Error::BadRequest(ErrorKind::forbidden(), "URL is not allowed to be previewed"));
if !services.media.url_preview_allowed(url) {
return Err!(Request(Forbidden(
warn!(%sender_user, %url, "URL is not allowed to be previewed")
)));
}
match get_url_preview(&services, url).await {
match services.media.get_url_preview(url).await {
Ok(preview) => {
let res = serde_json::value::to_raw_value(&preview).map_err(|e| {
error!(%sender_user, "Failed to convert UrlPreviewData into a serde json value: {e}");
@@ -97,7 +91,6 @@ pub(crate) async fn get_media_preview_route(
},
Err(e) => {
warn!(%sender_user, "Failed to generate a URL preview: {e}");
// there doesn't seem to be an agreed-upon error code in the spec.
// the only response codes in the preview_url spec page are 200 and 429.
Err(Error::BadRequest(
@@ -229,10 +222,7 @@ pub(crate) async fn get_content_route(
body.timeout_ms,
)
.await
.map_err(|e| {
debug_warn!("Fetching media `{}` failed: {:?}", mxc, e);
Error::BadRequest(ErrorKind::NotFound, "Remote media error.")
})?;
.map_err(|e| err!(Request(NotFound(debug_warn!("Fetching media `{mxc}` failed: {e:?}")))))?;
let content_disposition = Some(make_content_disposition(
&response.content_type,
@@ -248,7 +238,7 @@ pub(crate) async fn get_content_route(
cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_owned()),
})
} else {
Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
Err!(Request(NotFound("Media not found.")))
}
}
@@ -335,13 +325,10 @@ pub(crate) async fn get_content_as_filename_route(
cache_control: Some(CACHE_CONTROL_IMMUTABLE.into()),
})
},
Err(e) => {
debug_warn!("Fetching media `{}` failed: {:?}", mxc, e);
Err(Error::BadRequest(ErrorKind::NotFound, "Remote media error."))
},
Err(e) => Err!(Request(NotFound(debug_warn!("Fetching media `{mxc}` failed: {e:?}")))),
}
} else {
Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
Err!(Request(NotFound("Media not found.")))
}
}
@@ -392,10 +379,10 @@ pub(crate) async fn get_content_thumbnail_route(
&mxc,
body.width
.try_into()
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Width is invalid."))?,
.map_err(|e| err!(Request(InvalidParam("Width is invalid: {e:?}"))))?,
body.height
.try_into()
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Height is invalid."))?,
.map_err(|e| err!(Request(InvalidParam("Height is invalid: {e:?}"))))?,
)
.await?
{
@@ -418,7 +405,7 @@ pub(crate) async fn get_content_thumbnail_route(
// we'll lie to the client and say the blocked server's media was not found and
// log. the client has no way of telling anyways so this is a security bonus.
debug_warn!("Received request for media `{}` on blocklisted server", mxc);
return Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."));
return Err!(Request(NotFound("Media not found.")));
}
match services
@@ -467,13 +454,10 @@ pub(crate) async fn get_content_thumbnail_route(
content_disposition,
})
},
Err(e) => {
debug_warn!("Fetching media `{}` failed: {:?}", mxc, e);
Err(Error::BadRequest(ErrorKind::NotFound, "Remote media error."))
},
Err(e) => Err!(Request(NotFound(debug_warn!("Fetching media `{mxc}` failed: {e:?}")))),
}
} else {
Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
Err!(Request(NotFound("Media not found.")))
}
}
@@ -511,7 +495,7 @@ async fn get_remote_content(
// we'll lie to the client and say the blocked server's media was not found and
// log. the client has no way of telling anyways so this is a security bonus.
debug_warn!("Received request for media `{mxc}` on blocklisted server");
return Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."));
return Err!(Request(NotFound("Media not found.")));
}
let content_response = services
@@ -553,222 +537,3 @@ async fn get_remote_content(
cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_owned()),
})
}
async fn download_image(services: &Services, client: &reqwest::Client, url: &str) -> Result<UrlPreviewData> {
let image = client.get(url).send().await?.bytes().await?;
let mxc = format!("mxc://{}/{}", services.globals.server_name(), utils::random_string(MXC_LENGTH));
services
.media
.create(None, &mxc, None, None, &image)
.await?;
let (width, height) = match ImgReader::new(Cursor::new(&image)).with_guessed_format() {
Err(_) => (None, None),
Ok(reader) => match reader.into_dimensions() {
Err(_) => (None, None),
Ok((width, height)) => (Some(width), Some(height)),
},
};
Ok(UrlPreviewData {
image: Some(mxc),
image_size: Some(image.len()),
image_width: width,
image_height: height,
..Default::default()
})
}
async fn download_html(services: &Services, client: &reqwest::Client, url: &str) -> Result<UrlPreviewData> {
let mut response = client.get(url).send().await?;
let mut bytes: Vec<u8> = Vec::new();
while let Some(chunk) = response.chunk().await? {
bytes.extend_from_slice(&chunk);
if bytes.len() > services.globals.url_preview_max_spider_size() {
debug!(
"Response body from URL {} exceeds url_preview_max_spider_size ({}), not processing the rest of the \
response body and assuming our necessary data is in this range.",
url,
services.globals.url_preview_max_spider_size()
);
break;
}
}
let body = String::from_utf8_lossy(&bytes);
let Ok(html) = HTML::from_string(body.to_string(), Some(url.to_owned())) else {
return Err(Error::BadRequest(ErrorKind::Unknown, "Failed to parse HTML"));
};
let mut data = match html.opengraph.images.first() {
None => UrlPreviewData::default(),
Some(obj) => download_image(services, client, &obj.url).await?,
};
let props = html.opengraph.properties;
/* use OpenGraph title/description, but fall back to HTML if not available */
data.title = props.get("title").cloned().or(html.title);
data.description = props.get("description").cloned().or(html.description);
Ok(data)
}
async fn request_url_preview(services: &Services, url: &str) -> Result<UrlPreviewData> {
if let Ok(ip) = IPAddress::parse(url) {
if !services.globals.valid_cidr_range(&ip) {
return Err(Error::BadServerResponse("Requesting from this address is forbidden"));
}
}
let client = &services.client.url_preview;
let response = client.head(url).send().await?;
if let Some(remote_addr) = response.remote_addr() {
if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) {
if !services.globals.valid_cidr_range(&ip) {
return Err(Error::BadServerResponse("Requesting from this address is forbidden"));
}
}
}
let Some(content_type) = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|x| x.to_str().ok())
else {
return Err(Error::BadRequest(ErrorKind::Unknown, "Unknown Content-Type"));
};
let data = match content_type {
html if html.starts_with("text/html") => download_html(services, client, url).await?,
img if img.starts_with("image/") => download_image(services, client, url).await?,
_ => return Err(Error::BadRequest(ErrorKind::Unknown, "Unsupported Content-Type")),
};
services.media.set_url_preview(url, &data).await?;
Ok(data)
}
async fn get_url_preview(services: &Services, url: &str) -> Result<UrlPreviewData> {
if let Some(preview) = services.media.get_url_preview(url).await {
return Ok(preview);
}
// ensure that only one request is made per URL
let _request_lock = services.media.url_preview_mutex.lock(url).await;
match services.media.get_url_preview(url).await {
Some(preview) => Ok(preview),
None => request_url_preview(services, url).await,
}
}
fn url_preview_allowed(services: &Services, url_str: &str) -> bool {
let url: Url = match Url::parse(url_str) {
Ok(u) => u,
Err(e) => {
warn!("Failed to parse URL from a str: {}", e);
return false;
},
};
if ["http", "https"]
.iter()
.all(|&scheme| scheme != url.scheme().to_lowercase())
{
debug!("Ignoring non-HTTP/HTTPS URL to preview: {}", url);
return false;
}
let host = match url.host_str() {
None => {
debug!("Ignoring URL preview for a URL that does not have a host (?): {}", url);
return false;
},
Some(h) => h.to_owned(),
};
let allowlist_domain_contains = services.globals.url_preview_domain_contains_allowlist();
let allowlist_domain_explicit = services.globals.url_preview_domain_explicit_allowlist();
let denylist_domain_explicit = services.globals.url_preview_domain_explicit_denylist();
let allowlist_url_contains = services.globals.url_preview_url_contains_allowlist();
if allowlist_domain_contains.contains(&"*".to_owned())
|| allowlist_domain_explicit.contains(&"*".to_owned())
|| allowlist_url_contains.contains(&"*".to_owned())
{
debug!("Config key contains * which is allowing all URL previews. Allowing URL {}", url);
return true;
}
if !host.is_empty() {
if denylist_domain_explicit.contains(&host) {
debug!(
"Host {} is not allowed by url_preview_domain_explicit_denylist (check 1/4)",
&host
);
return false;
}
if allowlist_domain_explicit.contains(&host) {
debug!("Host {} is allowed by url_preview_domain_explicit_allowlist (check 2/4)", &host);
return true;
}
if allowlist_domain_contains
.iter()
.any(|domain_s| domain_s.contains(&host.clone()))
{
debug!("Host {} is allowed by url_preview_domain_contains_allowlist (check 3/4)", &host);
return true;
}
if allowlist_url_contains
.iter()
.any(|url_s| url.to_string().contains(&url_s.to_string()))
{
debug!("URL {} is allowed by url_preview_url_contains_allowlist (check 4/4)", &host);
return true;
}
// check root domain if available and if user has root domain checks
if services.globals.url_preview_check_root_domain() {
debug!("Checking root domain");
match host.split_once('.') {
None => return false,
Some((_, root_domain)) => {
if denylist_domain_explicit.contains(&root_domain.to_owned()) {
debug!(
"Root domain {} is not allowed by url_preview_domain_explicit_denylist (check 1/3)",
&root_domain
);
return true;
}
if allowlist_domain_explicit.contains(&root_domain.to_owned()) {
debug!(
"Root domain {} is allowed by url_preview_domain_explicit_allowlist (check 2/3)",
&root_domain
);
return true;
}
if allowlist_domain_contains
.iter()
.any(|domain_s| domain_s.contains(&root_domain.to_owned()))
{
debug!(
"Root domain {} is allowed by url_preview_domain_contains_allowlist (check 3/3)",
&root_domain
);
return true;
}
},
}
}
}
false
}
+24 -29
View File
@@ -8,11 +8,11 @@ use std::{
use axum::extract::State;
use axum_client_ip::InsecureClientIp;
use conduit::{
debug, debug_warn, error, info,
debug, debug_error, debug_warn, err, error, info,
pdu::{gen_event_id_canonical_json, PduBuilder},
trace, utils,
utils::math::continue_exponential_backoff_secs,
warn, Error, PduEvent, Result,
warn, Err, Error, PduEvent, Result,
};
use ruma::{
api::{
@@ -705,11 +705,11 @@ async fn join_room_by_id_helper_remote(
{
room_version
},
_ => return Err(Error::BadServerResponse("Room version is not supported")),
_ => return Err!(BadServerResponse("Room version is not supported")),
};
let mut join_event_stub: CanonicalJsonObject = serde_json::from_str(make_join_response.event.get())
.map_err(|_| Error::BadServerResponse("Invalid make_join event json received from server."))?;
.map_err(|e| err!(BadServerResponse("Invalid make_join event json received from server: {e:?}")))?;
let join_authorized_via_users_server = join_event_stub
.get("content")
@@ -876,15 +876,14 @@ async fn join_room_by_id_helper_remote(
info!("Parsing join event");
let parsed_join_pdu = PduEvent::from_id_val(event_id, join_event.clone())
.map_err(|_| Error::BadServerResponse("Invalid join event PDU."))?;
.map_err(|e| err!(BadServerResponse("Invalid join event PDU: {e:?}")))?;
let mut state = HashMap::new();
let pub_key_map = RwLock::new(BTreeMap::new());
info!("Fetching join signing keys");
services
.rooms
.event_handler
.server_keys
.fetch_join_signing_keys(&send_join_response, &room_version_id, &pub_key_map)
.await?;
@@ -900,8 +899,8 @@ async fn join_room_by_id_helper_remote(
};
let pdu = PduEvent::from_id_val(&event_id, value.clone()).map_err(|e| {
warn!("Invalid PDU in send_join response: {} {:?}", e, value);
Error::BadServerResponse("Invalid PDU in send_join response.")
debug_warn!("Invalid PDU in send_join response: {value:#?}");
err!(BadServerResponse("Invalid PDU in send_join response: {e:?}"))
})?;
services.rooms.outlier.add_pdu_outlier(&event_id, &value)?;
@@ -1123,10 +1122,10 @@ async fn join_room_by_id_helper_local(
{
room_version_id
},
_ => return Err(Error::BadServerResponse("Room version is not supported")),
_ => return Err!(BadServerResponse("Room version is not supported")),
};
let mut join_event_stub: CanonicalJsonObject = serde_json::from_str(make_join_response.event.get())
.map_err(|_| Error::BadServerResponse("Invalid make_join event json received from server."))?;
.map_err(|e| err!(BadServerResponse("Invalid make_join event json received from server: {e:?}")))?;
let join_authorized_via_users_server = join_event_stub
.get("content")
.map(|s| {
@@ -1230,8 +1229,7 @@ async fn join_room_by_id_helper_local(
drop(state_lock);
let pub_key_map = RwLock::new(BTreeMap::new());
services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys([&signed_value], &pub_key_map)
.await?;
services
@@ -1252,7 +1250,7 @@ async fn join_room_by_id_helper_local(
async fn make_join_request(
services: &Services, sender_user: &UserId, room_id: &RoomId, servers: &[OwnedServerName],
) -> Result<(federation::membership::prepare_join_event::v1::Response, OwnedServerName)> {
let mut make_join_response_and_server = Err(Error::BadServerResponse("No server available to assist in joining."));
let mut make_join_response_and_server = Err!(BadServerResponse("No server available to assist in joining."));
let mut make_join_counter: u16 = 0;
let mut incompatible_room_version_count: u8 = 0;
@@ -1293,8 +1291,7 @@ async fn make_join_request(
"15 servers have responded with M_INCOMPATIBLE_ROOM_VERSION or M_UNSUPPORTED_ROOM_VERSION, \
assuming that Conduwuit does not support the room {room_id}: {e}"
);
make_join_response_and_server =
Err(Error::BadServerResponse("Room version is not supported by Conduwuit"));
make_join_response_and_server = Err!(BadServerResponse("Room version is not supported by Conduwuit"));
return make_join_response_and_server;
}
@@ -1302,8 +1299,7 @@ async fn make_join_request(
warn!(
"50 servers failed to provide valid make_join response, assuming no server can assist in joining."
);
make_join_response_and_server =
Err(Error::BadServerResponse("No server available to assist in joining."));
make_join_response_and_server = Err!(BadServerResponse("No server available to assist in joining."));
return make_join_response_and_server;
}
}
@@ -1323,8 +1319,8 @@ pub async fn validate_and_add_event_id(
pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
) -> Result<(OwnedEventId, CanonicalJsonObject)> {
let mut value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
error!("Invalid PDU in server response: {:?}: {:?}", pdu, e);
Error::BadServerResponse("Invalid PDU in server response")
debug_error!("Invalid PDU in server response: {pdu:#?}");
err!(BadServerResponse("Invalid PDU in server response: {e:?}"))
})?;
let event_id = EventId::parse(format!(
"${}",
@@ -1360,15 +1356,15 @@ pub async fn validate_and_add_event_id(
const MIN: u64 = 60 * 5;
const MAX: u64 = 60 * 60 * 24;
if continue_exponential_backoff_secs(MIN, MAX, time.elapsed(), *tries) {
debug!("Backing off from {event_id}");
return Err(Error::BadServerResponse("bad event, still backing off"));
return Err!(BadServerResponse("bad event {event_id:?}, still backing off"));
}
}
if let Err(e) = ruma::signatures::verify_event(&*pub_key_map.read().await, &value, room_version) {
warn!("Event {} failed verification {:?} {}", event_id, pdu, e);
debug_error!("Event {event_id} failed verification {pdu:#?}");
let e = Err!(BadServerResponse(debug_error!("Event {event_id} failed verification: {e:?}")));
back_off(event_id).await;
return Err(Error::BadServerResponse("Event failed verification."));
return e;
}
value.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.as_str().to_owned()));
@@ -1474,8 +1470,7 @@ pub(crate) async fn invite_helper(
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?;
services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys([&value], &pub_key_map)
.await?;
@@ -1651,7 +1646,7 @@ pub async fn leave_room(services: &Services, user_id: &UserId, room_id: &RoomId,
}
async fn remote_leave_room(services: &Services, user_id: &UserId, room_id: &RoomId) -> Result<()> {
let mut make_leave_response_and_server = Err(Error::BadServerResponse("No server available to assist in leaving."));
let mut make_leave_response_and_server = Err!(BadServerResponse("No server available to assist in leaving."));
let invite_state = services
.rooms
@@ -1708,11 +1703,11 @@ async fn remote_leave_room(services: &Services, user_id: &UserId, room_id: &Room
{
version
},
_ => return Err(Error::BadServerResponse("Room version is not supported")),
_ => return Err!(BadServerResponse("Room version is not supported")),
};
let mut leave_event_stub = serde_json::from_str::<CanonicalJsonObject>(make_leave_response.event.get())
.map_err(|_| Error::BadServerResponse("Invalid make_leave event json received from server."))?;
.map_err(|e| err!(BadServerResponse("Invalid make_leave event json received from server: {e:?}")))?;
// TODO: Is origin needed?
leave_event_stub.insert(
+17 -7
View File
@@ -1,4 +1,5 @@
use axum::extract::State;
use axum_client_ip::InsecureClientIp;
use ruma::{
api::client::{
error::ErrorKind,
@@ -33,8 +34,9 @@ struct Claims {
///
/// Get the supported login types of this server. One of these should be used as
/// the `type` field when logging in.
#[tracing::instrument(skip_all, fields(%client), name = "register")]
pub(crate) async fn get_login_types_route(
_body: Ruma<get_login_types::v3::Request>,
InsecureClientIp(client): InsecureClientIp, _body: Ruma<get_login_types::v3::Request>,
) -> Result<get_login_types::v3::Response> {
Ok(get_login_types::v3::Response::new(vec![
get_login_types::v3::LoginType::Password(PasswordLoginType::default()),
@@ -56,8 +58,9 @@ pub(crate) async fn get_login_types_route(
/// Note: You can use [`GET
/// /_matrix/client/r0/login`](fn.get_supported_versions_route.html) to see
/// supported login types.
#[tracing::instrument(skip_all, fields(%client), name = "register")]
pub(crate) async fn login_route(
State(services): State<crate::State>, body: Ruma<login::v3::Request>,
State(services): State<crate::State>, InsecureClientIp(client): InsecureClientIp, body: Ruma<login::v3::Request>,
) -> Result<login::v3::Response> {
// Validate login method
// TODO: Other login methods
@@ -176,9 +179,13 @@ pub(crate) async fn login_route(
if device_exists {
services.users.set_token(&user_id, &device_id, &token)?;
} else {
services
.users
.create_device(&user_id, &device_id, &token, body.initial_device_display_name.clone())?;
services.users.create_device(
&user_id,
&device_id,
&token,
body.initial_device_display_name.clone(),
Some(client.to_string()),
)?;
}
// send client well-known if specified so the client knows to reconfigure itself
@@ -214,8 +221,9 @@ pub(crate) async fn login_route(
/// last seen ts)
/// - Forgets to-device events
/// - Triggers device list updates
#[tracing::instrument(skip_all, fields(%client), name = "register")]
pub(crate) async fn logout_route(
State(services): State<crate::State>, body: Ruma<logout::v3::Request>,
State(services): State<crate::State>, InsecureClientIp(client): InsecureClientIp, body: Ruma<logout::v3::Request>,
) -> Result<logout::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let sender_device = body.sender_device.as_ref().expect("user is authenticated");
@@ -241,8 +249,10 @@ pub(crate) async fn logout_route(
/// Note: This is equivalent to calling [`GET
/// /_matrix/client/r0/logout`](fn.logout_route.html) from each device of this
/// user.
#[tracing::instrument(skip_all, fields(%client), name = "register")]
pub(crate) async fn logout_all_route(
State(services): State<crate::State>, body: Ruma<logout_all::v3::Request>,
State(services): State<crate::State>, InsecureClientIp(client): InsecureClientIp,
body: Ruma<logout_all::v3::Request>,
) -> Result<logout_all::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+5 -3
View File
@@ -106,11 +106,13 @@ pub(crate) async fn sync_events_route(
.unwrap_or_default(),
};
// some clients, at least element, seem to require knowledge of redundant
// members for "inline" profiles on the timeline to work properly
let (lazy_load_enabled, lazy_load_send_redundant) = match filter.room.state.lazy_load_options {
LazyLoadOptions::Enabled {
include_redundant_members: redundant,
} => (true, redundant),
LazyLoadOptions::Disabled => (false, false),
include_redundant_members,
} => (true, include_redundant_members),
LazyLoadOptions::Disabled => (false, cfg!(feature = "element_hacks")),
};
let full_state = body.full_state;
+1 -2
View File
@@ -10,8 +10,7 @@ extern crate conduit_service as service;
pub(crate) use conduit::{debug_info, pdu::PduEvent, utils, Error, Result};
pub(crate) use service::services;
pub use crate::router::State;
pub(crate) use crate::router::{Ruma, RumaResponse};
pub(crate) use self::router::{Ruma, RumaResponse, State};
conduit::mod_ctor! {}
conduit::mod_dtor! {}
+2 -5
View File
@@ -3,8 +3,7 @@ mod auth;
mod handler;
mod request;
mod response;
use std::sync::Arc;
pub mod state;
use axum::{
response::IntoResponse,
@@ -15,11 +14,9 @@ use conduit::{err, Server};
use http::Uri;
use self::handler::RouterExt;
pub(super) use self::{args::Args as Ruma, response::RumaResponse};
pub(super) use self::{args::Args as Ruma, response::RumaResponse, state::State};
use crate::{client, server};
pub type State = Arc<service::Services>;
pub fn build(router: Router<State>, server: &Server) -> Router<State> {
let config = &server.config;
let router = router
+1 -2
View File
@@ -250,8 +250,7 @@ async fn auth_server(
};
let keys_result = services
.rooms
.event_handler
.server_keys
.fetch_signing_keys_for_server(origin, vec![x_matrix.key.to_string()])
.await;
+79
View File
@@ -0,0 +1,79 @@
use std::{ops::Deref, sync::Arc};
use conduit_service::Services;
#[derive(Clone, Copy)]
pub struct State {
services: *const Services,
}
pub struct Guard {
services: Arc<Services>,
}
pub fn create(services: Arc<Services>) -> (State, Guard) {
let state = State {
services: Arc::into_raw(services.clone()),
};
let guard = Guard {
services,
};
(state, guard)
}
impl Drop for Guard {
fn drop(&mut self) {
let ptr = Arc::as_ptr(&self.services);
// SAFETY: Parity with Arc::into_raw() called in create(). This revivifies the
// Arc lost to State so it can be dropped, otherwise Services will leak.
let arc = unsafe { Arc::from_raw(ptr) };
debug_assert!(
Arc::strong_count(&arc) > 1,
"Services usually has more than one reference and is not dropped here"
);
}
}
impl Deref for State {
type Target = Services;
fn deref(&self) -> &Self::Target {
deref(&self.services).expect("dereferenced Services pointer in State must not be null")
}
}
/// SAFETY: State is a thin wrapper containing a raw const pointer to Services
/// in lieu of an Arc. Services is internally threadsafe. If State contains
/// additional fields this notice should be reevaluated.
unsafe impl Send for State {}
/// SAFETY: State is a thin wrapper containing a raw const pointer to Services
/// in lieu of an Arc. Services is internally threadsafe. If State contains
/// additional fields this notice should be reevaluated.
unsafe impl Sync for State {}
fn deref(services: &*const Services) -> Option<&Services> {
// SAFETY: We replaced Arc<Services> with *const Services in State. This is
// worth about 10 clones (20 reference count updates) for each request handled.
// Though this is not an incredibly large quantity, it's woefully unnecessary
// given the context as explained below; though it is not currently known to be
// a performance bottleneck, the front-line position justifies preempting it.
//
// Services is created prior to the axum/tower stack and Router, and prior
// to serving any requests through the handlers in this crate. It is then
// dropped only after all requests have completed, the listening sockets
// have been closed, axum/tower has been dropped. Thus Services is
// expected to live at least as long as any instance of State, making the
// constant updates to the prior Arc unnecessary to keep Services alive.
//
// Nevertheless if it is possible to accomplish this by annotating State
// with a lifetime to hold a reference (and be aware I have made a
// significant effort trying to make this work) this unsafety may not be
// necessary. It is either very difficult or impossible to get a
// lifetime'ed reference through Router / RumaHandler; though it is
// possible to pass a reference through axum's `with_state()` in trivial
// configurations as the only requirement of a State is Clone.
unsafe { services.as_ref() }
}
+1 -2
View File
@@ -105,8 +105,7 @@ async fn handle_pdus(
let pub_key_map = RwLock::new(BTreeMap::new());
if !parsed_pdus.is_empty() {
services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys(parsed_pdus.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
.await
.unwrap_or_else(|e| warn!("Could not fetch all signatures for PDUs from {origin}: {e:?}"));
+1 -2
View File
@@ -138,8 +138,7 @@ async fn create_join_event(
}
services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys([&value], &pub_key_map)
.await?;
+1 -2
View File
@@ -150,8 +150,7 @@ async fn create_leave_event(
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "origin is not a server name."))?;
services
.rooms
.event_handler
.server_keys
.fetch_required_signing_keys([&value], &pub_key_map)
.await?;
+2
View File
@@ -51,11 +51,13 @@ sha256_media = []
[dependencies]
argon2.workspace = true
arrayvec.workspace = true
axum.workspace = true
bytes.workspace = true
cargo_toml.workspace = true
checked_ops.workspace = true
chrono.workspace = true
clap.workspace = true
conduit-macros.workspace = true
const-str.workspace = true
ctor.workspace = true
+6 -1
View File
@@ -236,6 +236,8 @@ pub struct Config {
pub rocksdb_compaction_ioprio_idle: bool,
#[serde(default = "true_fn")]
pub rocksdb_compaction: bool,
#[serde(default = "default_rocksdb_stats_level")]
pub rocksdb_stats_level: u8,
pub emergency_password: Option<String>,
@@ -287,7 +289,7 @@ pub struct Config {
#[serde(default = "true_fn")]
pub media_startup_check: bool,
#[serde(default = "true_fn")]
#[serde(default)]
pub media_compat_file_link: bool,
#[serde(default = "Vec::new")]
pub prevent_media_downloads_from: Vec<OwnedServerName>,
@@ -718,6 +720,7 @@ impl fmt::Display for Config {
&self.rocksdb_compaction_ioprio_idle.to_string(),
);
line("RocksDB Compaction enabled", &self.rocksdb_compaction.to_string());
line("RocksDB Statistics level", &self.rocksdb_stats_level.to_string());
line("Media integrity checks on startup", &self.media_startup_check.to_string());
line("Media compatibility filesystem links", &self.media_compat_file_link.to_string());
line("Prevent Media Downloads From", {
@@ -1002,6 +1005,8 @@ fn default_rocksdb_compression_level() -> i32 { 32767 }
#[allow(clippy::doc_markdown)]
fn default_rocksdb_bottommost_compression_level() -> i32 { 32767 }
fn default_rocksdb_stats_level() -> u8 { 1 }
// I know, it's a great name
#[must_use]
pub fn default_default_room_version() -> RoomVersionId { RoomVersionId::V10 }
+5 -1
View File
@@ -14,7 +14,7 @@ pub use crate::utils::debug::*;
#[macro_export]
macro_rules! debug_event {
( $level:expr, $($x:tt)+ ) => {
if cfg!(debug_assertions) && cfg!(not(feature = "dev_release_log_level")) {
if $crate::debug::logging() {
::tracing::event!( $level, $($x)+ )
} else {
::tracing::debug!( $($x)+ )
@@ -88,3 +88,7 @@ pub fn panic_str(p: &Box<dyn Any + Send>) -> &'static str { p.downcast_ref::<&st
#[inline(always)]
#[must_use]
pub fn type_name<T>(_: &T) -> &'static str { std::any::type_name::<T>() }
#[must_use]
#[inline]
pub const fn logging() -> bool { cfg!(debug_assertions) && cfg!(not(feature = "dev_release_log_level")) }
+114 -25
View File
@@ -41,60 +41,149 @@ macro_rules! Err {
#[macro_export]
macro_rules! err {
(Config($item:literal, $($args:expr),*)) => {{
$crate::error!(config = %$item, $($args),*);
$crate::error::Error::Config($item, $crate::format_maybe!($($args),*))
}};
(Request(Forbidden($level:ident!($($args:expr),*)))) => {{
$crate::$level!($($args),*);
(Request(Forbidden($level:ident!($($args:tt)+)))) => {{
let mut buf = String::new();
$crate::error::Error::Request(
::ruma::api::client::error::ErrorKind::forbidden(),
$crate::format_maybe!($($args),*),
$crate::err_log!(buf, $level, $($args)+),
::http::StatusCode::BAD_REQUEST
)
}};
(Request(Forbidden($($args:expr),*))) => {
(Request(Forbidden($($args:tt)+))) => {
$crate::error::Error::Request(
::ruma::api::client::error::ErrorKind::forbidden(),
$crate::format_maybe!($($args),*),
$crate::format_maybe!($($args)+),
::http::StatusCode::BAD_REQUEST
)
};
(Request($variant:ident($level:ident!($($args:expr),*)))) => {{
$crate::$level!($($args),*);
(Request($variant:ident($level:ident!($($args:tt)+)))) => {{
let mut buf = String::new();
$crate::error::Error::Request(
::ruma::api::client::error::ErrorKind::$variant,
$crate::format_maybe!($($args),*),
$crate::err_log!(buf, $level, $($args)+),
::http::StatusCode::BAD_REQUEST
)
}};
(Request($variant:ident($($args:expr),*))) => {
(Request($variant:ident($($args:tt)+))) => {
$crate::error::Error::Request(
::ruma::api::client::error::ErrorKind::$variant,
$crate::format_maybe!($($args),*),
$crate::format_maybe!($($args)+),
::http::StatusCode::BAD_REQUEST
)
};
($variant:ident($level:ident!($($args:expr),*))) => {{
$crate::$level!($($args),*);
$crate::error::Error::$variant($crate::format_maybe!($($args),*))
(Config($item:literal, $($args:tt)+)) => {{
let mut buf = String::new();
$crate::error::Error::Config($item, $crate::err_log!(buf, error, config = %$item, $($args)+))
}};
($variant:ident($($args:expr),*)) => {
$crate::error::Error::$variant($crate::format_maybe!($($args),*))
($variant:ident($level:ident!($($args:tt)+))) => {{
let mut buf = String::new();
$crate::error::Error::$variant($crate::err_log!(buf, $level, $($args)+))
}};
($variant:ident($($args:tt)+)) => {
$crate::error::Error::$variant($crate::format_maybe!($($args)+))
};
($level:ident!($($args:expr),*)) => {{
$crate::$level!($($args),*);
$crate::error::Error::Err($crate::format_maybe!($($args),*))
($level:ident!($($args:tt)+)) => {{
let mut buf = String::new();
$crate::error::Error::Err($crate::err_log!(buf, $level, $($args)+))
}};
($($args:expr),*) => {
$crate::error::Error::Err($crate::format_maybe!($($args),*))
($($args:tt)+) => {
$crate::error::Error::Err($crate::format_maybe!($($args)+))
};
}
/// A trinity of integration between tracing, logging, and Error. This is a
/// customization of tracing::event! with the primary purpose of sharing the
/// error string, fieldset parsing and formatting. An added benefit is that we
/// can share the same callsite metadata for the source of our Error and the
/// associated logging and tracing event dispatches.
#[macro_export]
macro_rules! err_log {
($out:ident, $level:ident, $($fields:tt)+) => {{
use std::{fmt, fmt::Write};
use ::tracing::{
callsite, callsite2, level_enabled, metadata, valueset, Callsite, Event, __macro_support,
__tracing_log,
field::{Field, ValueSet, Visit},
Level,
};
const LEVEL: Level = $crate::err_lev!($level);
static __CALLSITE: callsite::DefaultCallsite = callsite2! {
name: std::concat! {
"event ",
std::file!(),
":",
std::line!(),
},
kind: metadata::Kind::EVENT,
target: std::module_path!(),
level: LEVEL,
fields: $($fields)+,
};
let visit = &mut |vs: ValueSet<'_>| {
struct Visitor<'a>(&'a mut String);
impl Visit for Visitor<'_> {
fn record_debug(&mut self, field: &Field, val: &dyn fmt::Debug) {
if field.name() == "message" {
write!(self.0, "{:?}", val).expect("stream error");
} else {
write!(self.0, " {}={:?}", field.name(), val).expect("stream error");
}
}
}
let meta = __CALLSITE.metadata();
let enabled = level_enabled!(LEVEL) && {
let interest = __CALLSITE.interest();
!interest.is_never() && __macro_support::__is_enabled(meta, interest)
};
if enabled {
Event::dispatch(meta, &vs);
}
__tracing_log!(LEVEL, __CALLSITE, &vs);
vs.record(&mut Visitor(&mut $out));
};
(visit)(valueset!(__CALLSITE.metadata().fields(), $($fields)+));
($out).into()
}}
}
#[macro_export]
macro_rules! err_lev {
(debug_warn) => {
if $crate::debug::logging() {
::tracing::Level::WARN
} else {
::tracing::Level::DEBUG
}
};
(debug_error) => {
if $crate::debug::logging() {
::tracing::Level::ERROR
} else {
::tracing::Level::DEBUG
}
};
(warn) => {
::tracing::Level::WARN
};
(error) => {
::tracing::Level::ERROR
};
}
+5 -4
View File
@@ -5,8 +5,7 @@ mod response;
use std::{any::Any, borrow::Cow, convert::Infallible, fmt};
pub use log::*;
pub use self::log::*;
use crate::error;
#[derive(thiserror::Error)]
@@ -63,6 +62,8 @@ pub enum Error {
TomlSerError(#[from] toml::ser::Error),
#[error("{0}")]
TomlDeError(#[from] toml::de::Error),
#[error("{0}")]
Clap(#[from] clap::error::Error),
// ruma
#[error("{0}")]
@@ -87,13 +88,13 @@ pub enum Error {
// conduwuit
#[error("Arithmetic operation failed: {0}")]
Arithmetic(Cow<'static, str>),
#[error("{0}")]
BadServerResponse(Cow<'static, str>),
#[error("There was a problem with the '{0}' directive in your configuration: {1}")]
Config(&'static str, Cow<'static, str>),
#[error("{0}")]
Database(Cow<'static, str>),
#[error("{0}")]
BadServerResponse(&'static str),
#[error("{0}")]
Conflict(&'static str), // This is only needed for when a room alias already exists
// unique / untyped
+3
View File
@@ -6,3 +6,6 @@ pub mod rustc;
pub mod version;
pub use conduit_macros::rustc_flags_capture;
pub const MODULE_ROOT: &str = const_str::split!(std::module_path!(), "::")[0];
pub const CRATE_PREFIX: &str = const_str::split!(MODULE_ROOT, '_')[0];
+8 -5
View File
@@ -2,15 +2,20 @@ use tracing::Level;
use tracing_core::{span::Current, Event};
use super::{layer::Value, Layer};
use crate::{info, utils::string::EMPTY};
pub struct Data<'a> {
pub layer: &'a Layer,
pub event: &'a Event<'a>,
pub current: &'a Current,
pub values: Option<&'a mut [Value]>,
pub values: &'a [Value],
pub scope: &'a [&'static str],
}
impl Data<'_> {
#[must_use]
pub fn our_modules(&self) -> bool { self.mod_name().starts_with(info::CRATE_PREFIX) }
#[must_use]
pub fn level(&self) -> Level { *self.event.metadata().level() }
@@ -18,15 +23,13 @@ impl Data<'_> {
pub fn mod_name(&self) -> &str { self.event.metadata().module_path().unwrap_or_default() }
#[must_use]
pub fn span_name(&self) -> &str { self.current.metadata().map_or("", |s| s.name()) }
pub fn span_name(&self) -> &str { self.current.metadata().map_or(EMPTY, |s| s.name()) }
#[must_use]
pub fn message(&self) -> &str {
self.values
.as_ref()
.expect("values are not composed for a filter")
.iter()
.find(|(k, _)| *k == "message")
.map_or("", |(_, v)| v.as_str())
.map_or(EMPTY, |(_, v)| v.as_str())
}
}
+21 -6
View File
@@ -1,21 +1,25 @@
use std::{fmt, sync::Arc};
use arrayvec::ArrayVec;
use tracing::field::{Field, Visit};
use tracing_core::{Event, Subscriber};
use tracing_subscriber::{layer::Context, registry::LookupSpan};
use super::{Capture, Data, State};
pub type Value = (&'static str, String);
pub struct Layer {
state: Arc<State>,
}
struct Visitor {
values: Vec<Value>,
values: Values,
}
type Values = ArrayVec<Value, 32>;
pub type Value = (&'static str, String);
type ScopeNames = ArrayVec<&'static str, 32>;
impl Layer {
#[inline]
pub fn new(state: &Arc<State>) -> Self {
@@ -51,8 +55,9 @@ fn handle<S>(layer: &Layer, capture: &Capture, event: &Event<'_>, ctx: &Context<
where
S: Subscriber + for<'a> LookupSpan<'a>,
{
let names = ScopeNames::new();
let mut visitor = Visitor {
values: Vec::new(),
values: Values::new(),
};
event.record(&mut visitor);
@@ -61,7 +66,8 @@ where
layer,
event,
current: &ctx.current_span(),
values: Some(&mut visitor.values),
values: &visitor.values,
scope: &names,
});
}
@@ -69,12 +75,21 @@ fn filter<S>(layer: &Layer, capture: &Capture, event: &Event<'_>, ctx: &Context<
where
S: Subscriber + for<'a> LookupSpan<'a>,
{
let values = Values::new();
let mut names = ScopeNames::new();
if let Some(scope) = ctx.event_scope(event) {
for span in scope {
names.push(span.name());
}
}
capture.filter.as_ref().map_or(true, |filter| {
filter(Data {
layer,
event,
current: &ctx.current_span(),
values: None,
values: &values,
scope: &names,
})
})
}
+6 -6
View File
@@ -5,7 +5,7 @@ use crate::Result;
pub fn html<S>(out: &mut S, level: &Level, span: &str, msg: &str) -> Result<()>
where
S: Write,
S: Write + ?Sized,
{
let color = color::code_tag(level);
let level = level.as_str().to_uppercase();
@@ -19,7 +19,7 @@ where
pub fn markdown<S>(out: &mut S, level: &Level, span: &str, msg: &str) -> Result<()>
where
S: Write,
S: Write + ?Sized,
{
let level = level.as_str().to_uppercase();
writeln!(out, "`{level:>5}` `{span:^12}` `{msg}`")?;
@@ -29,19 +29,19 @@ where
pub fn markdown_table<S>(out: &mut S, level: &Level, span: &str, msg: &str) -> Result<()>
where
S: Write,
S: Write + ?Sized,
{
let level = level.as_str().to_uppercase();
writeln!(out, "| `{level:>5}` | `{span:^12}` | `{msg} |")?;
writeln!(out, "| {level:>5} | {span:^12} | {msg} |")?;
Ok(())
}
pub fn markdown_table_head<S>(out: &mut S) -> Result<()>
where
S: Write,
S: Write + ?Sized,
{
write!(out, "| level | span | message |\n|------:|:----:|:--------|\n")?;
write!(out, "| level | span | message |\n| ------: | :-----: | :------- |\n")?;
Ok(())
}
+1 -1
View File
@@ -16,7 +16,7 @@ pub use error::Error;
pub use info::{rustc_flags_capture, version, version::version};
pub use pdu::{PduBuilder, PduCount, PduEvent};
pub use server::Server;
pub use utils::{ctor, dtor};
pub use utils::{ctor, dtor, implement};
pub use crate as conduit_core;
+2 -2
View File
@@ -1,7 +1,7 @@
use std::ffi::OsStr;
use super::{path, Library};
use crate::{Error, Result};
use crate::{Err, Result};
const OPEN_FLAGS: i32 = libloading::os::unix::RTLD_LAZY | libloading::os::unix::RTLD_GLOBAL;
@@ -16,7 +16,7 @@ pub fn from_path(path: &OsStr) -> Result<Library> {
let lib = unsafe { Library::open(Some(path), OPEN_FLAGS) };
if let Err(e) = lib {
let name = path::to_name(path)?;
return Err(Error::Err(format!("Loading module {name:?} failed: {e}")));
return Err!("Loading module {name:?} failed: {e}");
}
Ok(lib.expect("module loaded"))
+4 -6
View File
@@ -23,7 +23,7 @@ use serde_json::{
value::{to_raw_value, RawValue as RawJsonValue},
};
use crate::{warn, Error};
use crate::{err, warn, Error};
#[derive(Deserialize)]
struct ExtractRedactedBecause {
@@ -119,7 +119,7 @@ impl PduEvent {
// deliberately allowing for the possibility of negative age
let now: i128 = MilliSecondsSinceUnixEpoch::now().get().into();
let then: i128 = self.origin_server_ts.into();
let this_age: i128 = now.saturating_sub(then);
let this_age = now.saturating_sub(then);
unsigned.insert("age".to_owned(), to_raw_value(&this_age).unwrap());
self.unsigned = Some(to_raw_value(&unsigned).expect("unsigned is valid"));
@@ -396,10 +396,8 @@ impl Ord for PduEvent {
pub fn gen_event_id_canonical_json(
pdu: &RawJsonValue, room_version_id: &RoomVersionId,
) -> crate::Result<(OwnedEventId, CanonicalJsonObject)> {
let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
warn!("Error parsing incoming event {:?}: {:?}", pdu, e);
Error::BadServerResponse("Invalid PDU in server response")
})?;
let value: CanonicalJsonObject = serde_json::from_str(pdu.get())
.map_err(|e| err!(BadServerResponse(warn!("Error parsing incoming event: {e:?}"))))?;
let event_id = format!(
"${}",
+1
View File
@@ -17,6 +17,7 @@ use std::cmp::{self, Ordering};
pub use ::ctor::{ctor, dtor};
pub use bytes::{increment, u64_from_bytes, u64_from_u8, u64_from_u8x8};
pub use conduit_macros::implement;
pub use debug::slice_truncated as debug_slice_truncated;
pub use hash::calculate_hash;
pub use html::Escape as HtmlEscape;
+17 -3
View File
@@ -8,12 +8,12 @@ pub const EMPTY: &str = "";
/// arguments are provided the first is assumed to be a format string.
#[macro_export]
macro_rules! format_maybe {
($s:literal) => {
($s:literal $(,)?) => {
if $crate::is_format!($s) { std::format!($s).into() } else { $s.into() }
};
($($args:expr),*) => {
std::format!($($args),*).into()
($s:literal, $($args:tt)+) => {
std::format!($s, $($args)+).into()
};
}
@@ -24,6 +24,20 @@ macro_rules! is_format {
($s:literal) => {
::const_str::contains!($s, "{") && ::const_str::contains!($s, "}")
};
($($s:tt)+) => {
false
};
}
#[inline]
pub fn collect_stream<F>(func: F) -> Result<String>
where
F: FnOnce(&mut dyn std::fmt::Write) -> Result<()>,
{
let mut out = String::new();
func(&mut out)?;
Ok(out)
}
#[inline]
+9 -1
View File
@@ -2,7 +2,12 @@ use std::{ops::Index, sync::Arc};
use conduit::{Result, Server};
use crate::{cork::Cork, maps, maps::Maps, Engine, Map};
use crate::{
cork::Cork,
maps,
maps::{Maps, MapsKey, MapsVal},
Engine, Map,
};
pub struct Database {
pub db: Arc<Engine>,
@@ -30,6 +35,9 @@ impl Database {
#[inline]
#[must_use]
pub fn cork_and_sync(&self) -> Cork { Cork::new(&self.db, true, true) }
#[inline]
pub fn iter_maps(&self) -> impl Iterator<Item = (&MapsKey, &MapsVal)> + '_ { self.map.iter() }
}
impl Index<&str> for Database {
+36 -6
View File
@@ -1,14 +1,17 @@
use std::{
collections::{BTreeSet, HashMap},
ffi::CStr,
fmt::Write,
path::PathBuf,
sync::{atomic::AtomicU32, Arc, Mutex, RwLock},
};
use conduit::{debug, error, info, utils::time::rfc2822_from_seconds, warn, Result, Server};
use conduit::{debug, error, info, utils::time::rfc2822_from_seconds, warn, Err, Result, Server};
use rocksdb::{
backup::{BackupEngine, BackupEngineOptions},
perf::get_memory_usage_stats,
BoundColumnFamily, Cache, ColumnFamilyDescriptor, DBCommon, DBWithThreadMode, Env, MultiThreaded, Options,
AsColumnFamilyRef, BoundColumnFamily, Cache, ColumnFamilyDescriptor, DBCommon, DBWithThreadMode, Env,
MultiThreaded, Options,
};
use crate::{
@@ -50,10 +53,7 @@ impl Engine {
let load_time = std::time::Instant::now();
if config.rocksdb_repair {
warn!("Starting database repair. This may take a long time...");
if let Err(e) = Db::repair(&db_opts, &config.database_path) {
error!("Repair failed: {e:?}");
}
repair(&db_opts, &config.database_path)?;
}
debug!("Listing column families in database");
@@ -77,6 +77,7 @@ impl Engine {
let db = res.or_else(or_else)?;
info!(
columns = cfs.len(),
sequence = %db.latest_sequence_number(),
time = ?load_time.elapsed(),
"Opened database."
@@ -241,6 +242,30 @@ impl Engine {
},
}
}
/// Query for database property by null-terminated name which is expected to
/// have a result with an integer representation. This is intended for
/// low-overhead programmatic use.
pub(crate) fn property_integer(&self, cf: &impl AsColumnFamilyRef, name: &CStr) -> Result<u64> {
result(self.db.property_int_value_cf(cf, name))
.and_then(|val| val.map_or_else(|| Err!("Property {name:?} not found."), Ok))
}
/// Query for database property by name receiving the result in a string.
pub(crate) fn property(&self, cf: &impl AsColumnFamilyRef, name: &str) -> Result<String> {
result(self.db.property_value_cf(cf, name))
.and_then(|val| val.map_or_else(|| Err!("Property {name:?} not found."), Ok))
}
}
pub(crate) fn repair(db_opts: &Options, path: &PathBuf) -> Result<()> {
warn!("Starting database repair. This may take a long time...");
match Db::repair(db_opts, path) {
Ok(()) => info!("Database repair successful."),
Err(e) => return Err!("Repair failed: {e:?}"),
}
Ok(())
}
impl Drop for Engine {
@@ -259,5 +284,10 @@ impl Drop for Engine {
debug!("Joining background threads...");
self.env.join_all_threads();
info!(
sequence = %self.db.latest_sequence_number(),
"Closing database..."
);
}
}
+5 -1
View File
@@ -1,4 +1,4 @@
use std::{future::Future, mem::size_of, pin::Pin, sync::Arc};
use std::{ffi::CStr, future::Future, mem::size_of, pin::Pin, sync::Arc};
use conduit::{utils, Result};
use rocksdb::{
@@ -189,6 +189,10 @@ impl Map {
self.watchers.watch(prefix)
}
pub fn property_integer(&self, name: &CStr) -> Result<u64> { self.db.property_integer(&self.cf(), name) }
pub fn property(&self, name: &str) -> Result<String> { self.db.property(&self.cf(), name) }
#[inline]
pub fn name(&self) -> &str { &self.name }
+3 -1
View File
@@ -4,7 +4,9 @@ use conduit::Result;
use crate::{Engine, Map};
pub type Maps = BTreeMap<String, Arc<Map>>;
pub type Maps = BTreeMap<MapsKey, MapsVal>;
pub(crate) type MapsVal = Arc<Map>;
pub(crate) type MapsKey = String;
pub(crate) fn open(db: &Arc<Engine>) -> Result<Maps> { open_list(db, MAPS) }
+18 -3
View File
@@ -2,8 +2,8 @@ use std::{cmp, collections::HashMap};
use conduit::{utils, Config};
use rocksdb::{
BlockBasedOptions, Cache, DBCompactionStyle, DBCompressionType, DBRecoveryMode, Env, LogLevel, Options,
UniversalCompactOptions, UniversalCompactionStopStyle,
statistics::StatsLevel, BlockBasedOptions, Cache, DBCompactionStyle, DBCompressionType, DBRecoveryMode, Env,
LogLevel, Options, UniversalCompactOptions, UniversalCompactionStopStyle,
};
/// Create database-wide options suitable for opening the database. This also
@@ -13,6 +13,11 @@ use rocksdb::{
/// through cf_options().
pub(crate) fn db_options(config: &Config, env: &mut Env, row_cache: &Cache, col_cache: &Cache) -> Options {
const MIN_PARALLELISM: usize = 2;
const DEFAULT_STATS_LEVEL: StatsLevel = if cfg!(debug_assertions) {
StatsLevel::ExceptDetailedTimers
} else {
StatsLevel::DisableAll
};
let mut opts = Options::default();
@@ -68,8 +73,18 @@ pub(crate) fn db_options(config: &Config, env: &mut Env, row_cache: &Cache, col_
set_compression_defaults(&mut opts, config);
// Misc
opts.set_disable_auto_compactions(!config.rocksdb_compaction);
opts.create_if_missing(true);
opts.set_disable_auto_compactions(!config.rocksdb_compaction);
opts.set_statistics_level(match config.rocksdb_stats_level {
0 => StatsLevel::DisableAll,
1 => DEFAULT_STATS_LEVEL,
2 => StatsLevel::ExceptHistogramOrTimers,
3 => StatsLevel::ExceptTimers,
4 => StatsLevel::ExceptDetailedTimers,
5 => StatsLevel::ExceptTimeForMutex,
6_u8..=u8::MAX => StatsLevel::All,
});
// Default: https://github.com/facebook/rocksdb/wiki/WAL-Recovery-Modes#ktoleratecorruptedtailrecords
//
+4 -2
View File
@@ -9,11 +9,13 @@ pub(super) fn flags_capture(args: TokenStream) -> TokenStream {
};
let flag = std::env::args().collect::<Vec<_>>();
let flag_len = flag.len();
let ret = quote! {
pub static RUSTC_FLAGS: [&str; #flag_len] = [#( #flag ),*];
#[conduit_core::ctor]
fn _set_rustc_flags() {
let flags = &[#( #flag ),*];
conduit_core::info::rustc::FLAGS.lock().expect("locked").insert(#crate_name, flags);
conduit_core::info::rustc::FLAGS.lock().expect("locked").insert(#crate_name, &RUSTC_FLAGS);
}
// static strings have to be yanked on module unload
+41 -13
View File
@@ -1,18 +1,27 @@
#![cfg(conduit_mods)]
#[no_link]
extern crate conduit_service;
use std::{
future::Future,
pin::Pin,
sync::{atomic::Ordering, Arc},
};
use conduit::{mods, Error, Result};
use tracing::{debug, error};
use conduit::{debug, error, mods, Error, Result};
use conduit_service::Services;
use crate::Server;
type RunFuncResult = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
type RunFuncProto = fn(&Arc<conduit::Server>) -> RunFuncResult;
type StartFuncResult = Pin<Box<dyn Future<Output = Result<Arc<Services>>> + Send>>;
type StartFuncProto = fn(&Arc<conduit::Server>) -> StartFuncResult;
type RunFuncResult = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
type RunFuncProto = fn(&Arc<Services>) -> RunFuncResult;
type StopFuncResult = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
type StopFuncProto = fn(Arc<Services>) -> StopFuncResult;
const RESTART_THRESH: &str = "conduit_service";
const MODULE_NAMES: &[&str] = &[
@@ -33,15 +42,25 @@ pub(crate) async fn run(server: &Arc<Server>, starts: bool) -> Result<(bool, boo
let main_lock = server.mods.read().await;
let main_mod = (*main_lock).last().expect("main module loaded");
if starts {
let start = main_mod.get::<RunFuncProto>("start")?;
if let Err(error) = start(&server.server).await {
error!("Starting server: {error}");
return Err(error);
}
let start = main_mod.get::<StartFuncProto>("start")?;
match start(&server.server).await {
Ok(services) => server.services.lock().await.insert(services),
Err(error) => {
error!("Starting server: {error}");
return Err(error);
},
};
}
server.server.stopping.store(false, Ordering::Release);
let run = main_mod.get::<RunFuncProto>("run")?;
if let Err(error) = run(&server.server).await {
if let Err(error) = run(server
.services
.lock()
.await
.as_ref()
.expect("services initialized"))
.await
{
error!("Running server: {error}");
return Err(error);
}
@@ -49,8 +68,17 @@ pub(crate) async fn run(server: &Arc<Server>, starts: bool) -> Result<(bool, boo
let stops = !reloads || stale(server).await? <= restart_thresh();
let starts = reloads && stops;
if stops {
let stop = main_mod.get::<RunFuncProto>("stop")?;
if let Err(error) = stop(&server.server).await {
let stop = main_mod.get::<StopFuncProto>("stop")?;
if let Err(error) = stop(
server
.services
.lock()
.await
.take()
.expect("services initialized"),
)
.await
{
error!("Stopping server: {error}");
return Err(error);
}
@@ -103,7 +131,7 @@ pub(crate) async fn close(server: &Arc<Server>, force: bool) -> Result<usize, Er
async fn stale_count(server: &Arc<Server>) -> usize {
let watermark = stale(server).await.unwrap_or(available());
available() - watermark
available().saturating_sub(watermark)
}
async fn stale(server: &Arc<Server>) -> Result<usize, Error> {
+6 -4
View File
@@ -6,6 +6,7 @@ use axum::{
};
use axum_client_ip::SecureClientIpSource;
use conduit::{error, Result, Server};
use conduit_api::router::state::Guard;
use conduit_service::Services;
use http::{
header::{self, HeaderName},
@@ -35,7 +36,7 @@ const CONDUWUIT_CSP: &[&str] = &[
const CONDUWUIT_PERMISSIONS_POLICY: &[&str] = &["interest-cohort=()", "browsing-topics=()"];
pub(crate) fn build(services: &Arc<Services>) -> Result<Router> {
pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
let server = &services.server;
let layers = ServiceBuilder::new();
@@ -47,7 +48,7 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<Router> {
let layers = layers
.sensitive_headers([header::AUTHORIZATION])
.layer(axum::middleware::from_fn_with_state(Arc::clone(server), request::spawn))
.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::spawn))
.layer(
TraceLayer::new_for_http()
.make_span_with(tracing_span::<_>)
@@ -55,7 +56,7 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<Router> {
.on_request(DefaultOnRequest::new().level(Level::TRACE))
.on_response(DefaultOnResponse::new().level(Level::DEBUG)),
)
.layer(axum::middleware::from_fn_with_state(Arc::clone(server), request::handle))
.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::handle))
.layer(SecureClientIpSource::ConnectInfo.into_extension())
.layer(SetResponseHeaderLayer::if_not_present(
HeaderName::from_static("origin-agent-cluster"), // https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Origin-Agent-Cluster
@@ -85,7 +86,8 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<Router> {
.layer(body_limit_layer(server))
.layer(CatchPanicLayer::custom(catch_panic));
Ok(router::build(services).layer(layers))
let (router, guard) = router::build(services);
Ok((router.layer(layers), guard))
}
#[cfg(any(feature = "zstd_compression", feature = "gzip_compression", feature = "brotli_compression"))]
+6 -3
View File
@@ -4,13 +4,15 @@ use axum::{
extract::State,
response::{IntoResponse, Response},
};
use conduit::{debug, debug_error, debug_warn, defer, err, error, trace, Result, Server};
use conduit::{debug, debug_error, debug_warn, defer, err, error, trace, Result};
use conduit_service::Services;
use http::{Method, StatusCode, Uri};
#[tracing::instrument(skip_all, level = "debug")]
pub(crate) async fn spawn(
State(server): State<Arc<Server>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
State(services): State<Arc<Services>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
) -> Result<Response, StatusCode> {
let server = &services.server;
if !server.running() {
debug_warn!("unavailable pending shutdown");
return Err(StatusCode::SERVICE_UNAVAILABLE);
@@ -34,8 +36,9 @@ pub(crate) async fn spawn(
#[tracing::instrument(skip_all, name = "handle")]
pub(crate) async fn handle(
State(server): State<Arc<Server>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
State(services): State<Arc<Services>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
) -> Result<Response, StatusCode> {
let server = &services.server;
if !server.running() {
debug_warn!(
method = %req.method(),
+8 -7
View File
@@ -2,19 +2,20 @@ use std::sync::Arc;
use axum::{response::IntoResponse, routing::get, Router};
use conduit::Error;
use conduit_api::State;
use conduit_api::router::{state, state::Guard};
use conduit_service::Services;
use http::{StatusCode, Uri};
use ruma::api::client::error::ErrorKind;
pub(crate) fn build(services: &Arc<Services>) -> Router {
let router = Router::<State>::new();
let state = services.clone();
conduit_api::router::build(router, &services.server)
pub(crate) fn build(services: &Arc<Services>) -> (Router, Guard) {
let router = Router::<state::State>::new();
let (state, guard) = state::create(services.clone());
let router = conduit_api::router::build(router, &services.server)
.route("/", get(it_works))
.fallback(not_found)
.with_state(state)
.with_state(state);
(router, guard)
}
async fn not_found(_uri: Uri) -> impl IntoResponse {
+1 -1
View File
@@ -18,7 +18,7 @@ pub(super) async fn serve(
let server = &services.server;
let config = &server.config;
let addrs = config.get_bind_addrs();
let app = layers::build(&services)?;
let (app, _guard) = layers::build(&services)?;
if cfg!(unix) && config.unix_socket_path.is_some() {
unix::serve(server, app, shutdown).await
+1
View File
@@ -69,6 +69,7 @@ termimad.optional = true
tokio.workspace = true
tracing.workspace = true
url.workspace = true
webpage.workspace = true
[lints]
workspace = true
+18 -9
View File
@@ -5,7 +5,7 @@ mod grant;
use std::{
future::Future,
pin::Pin,
sync::{Arc, RwLock as StdRwLock},
sync::{Arc, RwLock as StdRwLock, Weak},
};
use async_trait::async_trait;
@@ -41,7 +41,7 @@ struct Services {
timeline: Dep<rooms::timeline::Service>,
state: Dep<rooms::state::Service>,
state_cache: Dep<rooms::state_cache::Service>,
services: StdRwLock<Option<Arc<crate::Services>>>,
services: StdRwLock<Option<Weak<crate::Services>>>,
}
#[derive(Debug)]
@@ -51,9 +51,9 @@ pub struct CommandInput {
}
pub type Completer = fn(&str) -> String;
pub type Handler = fn(Arc<crate::Services>, CommandInput) -> HandlerResult;
pub type HandlerResult = Pin<Box<dyn Future<Output = CommandResult> + Send>>;
pub type CommandResult = Result<CommandOutput, Error>;
pub type Handler = fn(Arc<crate::Services>, CommandInput) -> HandlerFuture;
pub type HandlerFuture = Pin<Box<dyn Future<Output = HandlerResult> + Send>>;
pub type HandlerResult = Result<CommandOutput>;
pub type CommandOutput = Option<RoomMessageEventContent>;
const COMMAND_QUEUE_LIMIT: usize = 512;
@@ -173,8 +173,15 @@ impl Service {
}
}
async fn process_command(&self, command: CommandInput) -> CommandResult {
let Some(services) = self.services.services.read().expect("locked").clone() else {
async fn process_command(&self, command: CommandInput) -> HandlerResult {
let Some(services) = self
.services
.services
.read()
.expect("locked")
.as_ref()
.and_then(Weak::upgrade)
else {
return Err!("Services self-reference not initialized.");
};
@@ -365,7 +372,9 @@ impl Service {
/// Sets the self-reference to crate::Services which will provide context to
/// the admin commands.
pub(super) fn set_services(&self, services: Option<Arc<crate::Services>>) {
*self.services.services.write().expect("locked for writing") = services;
pub(super) fn set_services(&self, services: &Option<Arc<crate::Services>>) {
let receiver = &mut *self.services.services.write().expect("locked for writing");
let weak = services.as_ref().map(Arc::downgrade);
*receiver = weak;
}
}
+30 -2
View File
@@ -743,6 +743,7 @@ async fn db_lt_13(services: &Services) -> Result<()> {
/// again.
async fn migrate_sha256_media(services: &Services) -> Result<()> {
let db = &services.db;
let config = &services.server.config;
warn!("Migrating legacy base64 file names to sha256 file names");
let mediaid_file = &db["mediaid_file"];
@@ -759,7 +760,9 @@ async fn migrate_sha256_media(services: &Services) -> Result<()> {
for (old_path, path) in changes {
if old_path.exists() {
tokio::fs::rename(&old_path, &path).await?;
tokio::fs::symlink(&path, &old_path).await?;
if config.media_compat_file_link {
tokio::fs::symlink(&path, &old_path).await?;
}
}
}
@@ -822,8 +825,14 @@ async fn handle_media_check(
let (mediaid_file, mediaid_user) = dbs;
let old_exists = files.contains(old_path);
let new_exists = files.contains(new_path);
let old_exists = files.contains(old_path);
let old_is_symlink = || async {
tokio::fs::symlink_metadata(old_path)
.await
.map_or(false, |md| md.is_symlink())
};
if !old_exists && !new_exists {
error!(
media_id = ?encode_key(key), ?new_path, ?old_path,
@@ -849,10 +858,29 @@ async fn handle_media_check(
"Legacy media found without sha256 migration. Fixing..."
);
debug_assert!(
old_is_symlink().await,
"Legacy media not expected to be a symlink without an existing sha256 migration."
);
tokio::fs::rename(&old_path, &new_path).await?;
tokio::fs::symlink(&new_path, &old_path).await?;
}
if !config.media_compat_file_link && old_exists && old_is_symlink().await {
debug_warn!(
media_id = ?encode_key(key), ?new_path, ?old_path,
"Legacy link found but compat disabled. Cleansing symlink..."
);
debug_assert!(
new_exists,
"sha256 migration into new file expected prior to cleaning legacy symlink here."
);
tokio::fs::remove_file(&old_path).await?;
}
Ok(())
}
+2 -1
View File
@@ -60,7 +60,8 @@ impl Manager {
.read()
.expect("locked for reading")
.values()
.map(|v| v.0.clone())
.map(|val| val.0.upgrade())
.map(|arc| arc.expect("services available for manager startup"))
.collect();
debug!("Starting service workers...");
+1 -1
View File
@@ -4,7 +4,7 @@ use conduit::{debug, debug_info, utils::string_from_bytes, Error, Result};
use database::{Database, Map};
use ruma::api::client::error::ErrorKind;
use crate::media::UrlPreviewData;
use super::preview::UrlPreviewData;
pub(crate) struct Data {
mediaid_file: Arc<Map>,
+20 -42
View File
@@ -1,4 +1,5 @@
mod data;
mod preview;
mod tests;
mod thumbnail;
@@ -6,16 +7,15 @@ use std::{path::PathBuf, sync::Arc, time::SystemTime};
use async_trait::async_trait;
use base64::{engine::general_purpose, Engine as _};
use conduit::{debug, debug_error, err, error, utils, utils::MutexMap, Err, Result, Server};
use conduit::{debug, debug_error, err, error, trace, utils, utils::MutexMap, Err, Result, Server};
use data::{Data, Metadata};
use ruma::{OwnedMxcUri, OwnedUserId};
use serde::Serialize;
use tokio::{
fs,
io::{AsyncReadExt, AsyncWriteExt, BufReader},
};
use crate::{globals, Dep};
use crate::{client, globals, Dep};
#[derive(Debug)]
pub struct FileMeta {
@@ -24,43 +24,32 @@ pub struct FileMeta {
pub content_disposition: Option<String>,
}
#[derive(Serialize, Default)]
pub struct UrlPreviewData {
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:title"))]
pub title: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:description"))]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image"))]
pub image: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "matrix:image:size"))]
pub image_size: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:width"))]
pub image_width: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:height"))]
pub image_height: Option<u32>,
}
pub struct Service {
services: Services,
url_preview_mutex: MutexMap<String, ()>,
pub(crate) db: Data,
pub url_preview_mutex: MutexMap<String, ()>,
services: Services,
}
struct Services {
server: Arc<Server>,
client: Dep<client::Service>,
globals: Dep<globals::Service>,
}
/// generated MXC ID (`media-id`) length
pub const MXC_LENGTH: usize = 32;
#[async_trait]
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
Ok(Arc::new(Self {
url_preview_mutex: MutexMap::new(),
db: Data::new(args.db),
services: Services {
server: args.server.clone(),
client: args.depend::<client::Service>("client"),
globals: args.depend::<globals::Service>("globals"),
},
db: Data::new(args.db),
url_preview_mutex: MutexMap::new(),
}))
}
@@ -99,10 +88,15 @@ impl Service {
pub async fn delete(&self, mxc: &str) -> Result<()> {
if let Ok(keys) = self.db.search_mxc_metadata_prefix(mxc) {
for key in keys {
self.remove_media_file(&key).await?;
trace!(?mxc, ?key, "Deleting from filesystem");
if let Err(e) = self.remove_media_file(&key).await {
error!(?mxc, ?key, "Failed to remove media file: {e}");
}
debug!("Deleting MXC {mxc} from database");
self.db.delete_file_mxc(mxc)?;
trace!(?mxc, ?key, "Deleting from database");
if let Err(e) = self.db.delete_file_mxc(mxc) {
error!(?mxc, ?key, "Failed to remove media from database: {e}");
}
}
Ok(())
@@ -224,22 +218,6 @@ impl Service {
Ok(deletion_count)
}
pub async fn get_url_preview(&self, url: &str) -> Option<UrlPreviewData> { self.db.get_url_preview(url) }
/// TODO: use this?
#[allow(dead_code)]
pub async fn remove_url_preview(&self, url: &str) -> Result<()> {
// TODO: also remove the downloaded image
self.db.remove_url_preview(url)
}
pub async fn set_url_preview(&self, url: &str, data: &UrlPreviewData) -> Result<()> {
let now = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.expect("valid system time");
self.db.set_url_preview(url, data, now)
}
pub async fn create_media_dir(&self) -> Result<()> {
let dir = self.get_media_dir();
Ok(fs::create_dir_all(dir).await?)
+274
View File
@@ -0,0 +1,274 @@
use std::{io::Cursor, time::SystemTime};
use conduit::{debug, utils, warn, Err, Result};
use conduit_core::implement;
use image::ImageReader as ImgReader;
use ipaddress::IPAddress;
use serde::Serialize;
use url::Url;
use webpage::HTML;
use super::{Service, MXC_LENGTH};
#[derive(Serialize, Default)]
pub struct UrlPreviewData {
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:title"))]
pub title: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:description"))]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image"))]
pub image: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "matrix:image:size"))]
pub image_size: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:width"))]
pub image_width: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:height"))]
pub image_height: Option<u32>,
}
#[implement(Service)]
pub async fn remove_url_preview(&self, url: &str) -> Result<()> {
// TODO: also remove the downloaded image
self.db.remove_url_preview(url)
}
#[implement(Service)]
pub async fn set_url_preview(&self, url: &str, data: &UrlPreviewData) -> Result<()> {
let now = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.expect("valid system time");
self.db.set_url_preview(url, data, now)
}
#[implement(Service)]
pub async fn download_image(&self, url: &str) -> Result<UrlPreviewData> {
let client = &self.services.client.url_preview;
let image = client.get(url).send().await?.bytes().await?;
let mxc = format!(
"mxc://{}/{}",
self.services.globals.server_name(),
utils::random_string(MXC_LENGTH)
);
self.create(None, &mxc, None, None, &image).await?;
let (width, height) = match ImgReader::new(Cursor::new(&image)).with_guessed_format() {
Err(_) => (None, None),
Ok(reader) => match reader.into_dimensions() {
Err(_) => (None, None),
Ok((width, height)) => (Some(width), Some(height)),
},
};
Ok(UrlPreviewData {
image: Some(mxc),
image_size: Some(image.len()),
image_width: width,
image_height: height,
..Default::default()
})
}
#[implement(Service)]
pub async fn get_url_preview(&self, url: &str) -> Result<UrlPreviewData> {
if let Some(preview) = self.db.get_url_preview(url) {
return Ok(preview);
}
// ensure that only one request is made per URL
let _request_lock = self.url_preview_mutex.lock(url).await;
match self.db.get_url_preview(url) {
Some(preview) => Ok(preview),
None => self.request_url_preview(url).await,
}
}
#[implement(Service)]
async fn request_url_preview(&self, url: &str) -> Result<UrlPreviewData> {
if let Ok(ip) = IPAddress::parse(url) {
if !self.services.globals.valid_cidr_range(&ip) {
return Err!(BadServerResponse("Requesting from this address is forbidden"));
}
}
let client = &self.services.client.url_preview;
let response = client.head(url).send().await?;
if let Some(remote_addr) = response.remote_addr() {
if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) {
if !self.services.globals.valid_cidr_range(&ip) {
return Err!(BadServerResponse("Requesting from this address is forbidden"));
}
}
}
let Some(content_type) = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|x| x.to_str().ok())
else {
return Err!(Request(Unknown("Unknown Content-Type")));
};
let data = match content_type {
html if html.starts_with("text/html") => self.download_html(url).await?,
img if img.starts_with("image/") => self.download_image(url).await?,
_ => return Err!(Request(Unknown("Unsupported Content-Type"))),
};
self.set_url_preview(url, &data).await?;
Ok(data)
}
#[implement(Service)]
async fn download_html(&self, url: &str) -> Result<UrlPreviewData> {
let client = &self.services.client.url_preview;
let mut response = client.get(url).send().await?;
let mut bytes: Vec<u8> = Vec::new();
while let Some(chunk) = response.chunk().await? {
bytes.extend_from_slice(&chunk);
if bytes.len() > self.services.globals.url_preview_max_spider_size() {
debug!(
"Response body from URL {} exceeds url_preview_max_spider_size ({}), not processing the rest of the \
response body and assuming our necessary data is in this range.",
url,
self.services.globals.url_preview_max_spider_size()
);
break;
}
}
let body = String::from_utf8_lossy(&bytes);
let Ok(html) = HTML::from_string(body.to_string(), Some(url.to_owned())) else {
return Err!(Request(Unknown("Failed to parse HTML")));
};
let mut data = match html.opengraph.images.first() {
None => UrlPreviewData::default(),
Some(obj) => self.download_image(&obj.url).await?,
};
let props = html.opengraph.properties;
/* use OpenGraph title/description, but fall back to HTML if not available */
data.title = props.get("title").cloned().or(html.title);
data.description = props.get("description").cloned().or(html.description);
Ok(data)
}
#[implement(Service)]
pub fn url_preview_allowed(&self, url_str: &str) -> bool {
let url: Url = match Url::parse(url_str) {
Ok(u) => u,
Err(e) => {
warn!("Failed to parse URL from a str: {}", e);
return false;
},
};
if ["http", "https"]
.iter()
.all(|&scheme| scheme != url.scheme().to_lowercase())
{
debug!("Ignoring non-HTTP/HTTPS URL to preview: {}", url);
return false;
}
let host = match url.host_str() {
None => {
debug!("Ignoring URL preview for a URL that does not have a host (?): {}", url);
return false;
},
Some(h) => h.to_owned(),
};
let allowlist_domain_contains = self
.services
.globals
.url_preview_domain_contains_allowlist();
let allowlist_domain_explicit = self
.services
.globals
.url_preview_domain_explicit_allowlist();
let denylist_domain_explicit = self.services.globals.url_preview_domain_explicit_denylist();
let allowlist_url_contains = self.services.globals.url_preview_url_contains_allowlist();
if allowlist_domain_contains.contains(&"*".to_owned())
|| allowlist_domain_explicit.contains(&"*".to_owned())
|| allowlist_url_contains.contains(&"*".to_owned())
{
debug!("Config key contains * which is allowing all URL previews. Allowing URL {}", url);
return true;
}
if !host.is_empty() {
if denylist_domain_explicit.contains(&host) {
debug!(
"Host {} is not allowed by url_preview_domain_explicit_denylist (check 1/4)",
&host
);
return false;
}
if allowlist_domain_explicit.contains(&host) {
debug!("Host {} is allowed by url_preview_domain_explicit_allowlist (check 2/4)", &host);
return true;
}
if allowlist_domain_contains
.iter()
.any(|domain_s| domain_s.contains(&host.clone()))
{
debug!("Host {} is allowed by url_preview_domain_contains_allowlist (check 3/4)", &host);
return true;
}
if allowlist_url_contains
.iter()
.any(|url_s| url.to_string().contains(&url_s.to_string()))
{
debug!("URL {} is allowed by url_preview_url_contains_allowlist (check 4/4)", &host);
return true;
}
// check root domain if available and if user has root domain checks
if self.services.globals.url_preview_check_root_domain() {
debug!("Checking root domain");
match host.split_once('.') {
None => return false,
Some((_, root_domain)) => {
if denylist_domain_explicit.contains(&root_domain.to_owned()) {
debug!(
"Root domain {} is not allowed by url_preview_domain_explicit_denylist (check 1/3)",
&root_domain
);
return true;
}
if allowlist_domain_explicit.contains(&root_domain.to_owned()) {
debug!(
"Root domain {} is allowed by url_preview_domain_explicit_allowlist (check 2/3)",
&root_domain
);
return true;
}
if allowlist_domain_contains
.iter()
.any(|domain_s| domain_s.contains(&root_domain.to_owned()))
{
debug!(
"Root domain {} is allowed by url_preview_domain_contains_allowlist (check 3/3)",
&root_domain
);
return true;
}
},
}
}
}
false
}
+1
View File
@@ -18,6 +18,7 @@ pub mod pusher;
pub mod resolver;
pub mod rooms;
pub mod sending;
pub mod server_keys;
pub mod transaction_ids;
pub mod uiaa;
pub mod updates;
+21 -4
View File
@@ -4,7 +4,8 @@ use conduit::{debug_warn, utils, Error, Result};
use database::Map;
use ruma::{events::presence::PresenceEvent, presence::PresenceState, OwnedUserId, UInt, UserId};
use crate::{globals, presence::Presence, users, Dep};
use super::Presence;
use crate::{globals, users, Dep};
pub struct Data {
presenceid_presence: Arc<Map>,
@@ -60,6 +61,22 @@ impl Data {
Some(ref presence) => presence.1.content.presence != *presence_state,
};
let status_msg_changed = match last_presence {
None => true,
Some(ref last_presence) => {
let old_msg = last_presence
.1
.content
.status_msg
.clone()
.unwrap_or_default();
let new_msg = status_msg.clone().unwrap_or_default();
new_msg != old_msg
},
};
let now = utils::millis_since_unix_epoch();
let last_last_active_ts = match last_presence {
None => 0,
@@ -71,10 +88,10 @@ impl Data {
Some(last_active_ago) => now.saturating_sub(last_active_ago.into()),
};
// tighten for state flicker?
if !state_changed && last_active_ts <= last_last_active_ts {
// TODO: tighten for state flicker?
if !status_msg_changed && !state_changed && last_active_ts < last_last_active_ts {
debug_warn!(
"presence spam {:?} last_active_ts:{:?} <= {:?}",
"presence spam {:?} last_active_ts:{:?} < {:?}",
user_id,
last_active_ts,
last_last_active_ts
+16 -70
View File
@@ -1,81 +1,25 @@
mod data;
mod presence;
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
use conduit::{checked, debug, error, utils, Error, Result, Server};
use conduit::{checked, debug, error, Error, Result, Server};
use futures_util::{stream::FuturesUnordered, StreamExt};
use ruma::{
events::presence::{PresenceEvent, PresenceEventContent},
presence::PresenceState,
OwnedUserId, UInt, UserId,
};
use serde::{Deserialize, Serialize};
use ruma::{events::presence::PresenceEvent, presence::PresenceState, OwnedUserId, UInt, UserId};
use tokio::{sync::Mutex, time::sleep};
use self::data::Data;
use self::{data::Data, presence::Presence};
use crate::{globals, users, Dep};
/// Represents data required to be kept in order to implement the presence
/// specification.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Presence {
state: PresenceState,
currently_active: bool,
last_active_ts: u64,
status_msg: Option<String>,
}
impl Presence {
#[must_use]
pub fn new(state: PresenceState, currently_active: bool, last_active_ts: u64, status_msg: Option<String>) -> Self {
Self {
state,
currently_active,
last_active_ts,
status_msg,
}
}
pub fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
serde_json::from_slice(bytes).map_err(|_| Error::bad_database("Invalid presence data in database"))
}
pub fn to_json_bytes(&self) -> Result<Vec<u8>> {
serde_json::to_vec(self).map_err(|_| Error::bad_database("Could not serialize Presence to JSON"))
}
/// Creates a PresenceEvent from available data.
pub fn to_presence_event(&self, user_id: &UserId, users: &users::Service) -> Result<PresenceEvent> {
let now = utils::millis_since_unix_epoch();
let last_active_ago = if self.currently_active {
None
} else {
Some(UInt::new_saturating(now.saturating_sub(self.last_active_ts)))
};
Ok(PresenceEvent {
sender: user_id.to_owned(),
content: PresenceEventContent {
presence: self.state.clone(),
status_msg: self.status_msg.clone(),
currently_active: Some(self.currently_active),
last_active_ago,
displayname: users.displayname(user_id)?,
avatar_url: users.avatar_url(user_id)?,
},
})
}
}
pub struct Service {
services: Services,
pub db: Data,
pub timer_sender: loole::Sender<(OwnedUserId, Duration)>,
timer_receiver: Mutex<loole::Receiver<(OwnedUserId, Duration)>>,
timer_sender: loole::Sender<TimerType>,
timer_receiver: Mutex<loole::Receiver<TimerType>>,
timeout_remote_users: bool,
idle_timeout: u64,
offline_timeout: u64,
pub db: Data,
services: Services,
}
struct Services {
@@ -84,6 +28,8 @@ struct Services {
users: Dep<users::Service>,
}
type TimerType = (OwnedUserId, Duration);
#[async_trait]
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@@ -92,17 +38,17 @@ impl crate::Service for Service {
let offline_timeout_s = config.presence_offline_timeout_s;
let (timer_sender, timer_receiver) = loole::unbounded();
Ok(Arc::new(Self {
services: Services {
server: args.server.clone(),
globals: args.depend::<globals::Service>("globals"),
users: args.depend::<users::Service>("users"),
},
db: Data::new(&args),
timer_sender,
timer_receiver: Mutex::new(timer_receiver),
timeout_remote_users: config.presence_timeout_remote_users,
idle_timeout: checked!(idle_timeout_s * 1_000)?,
offline_timeout: checked!(offline_timeout_s * 1_000)?,
db: Data::new(&args),
services: Services {
server: args.server.clone(),
globals: args.depend::<globals::Service>("globals"),
users: args.depend::<users::Service>("users"),
},
}))
}
+65
View File
@@ -0,0 +1,65 @@
use std::sync::Arc;
use conduit::{utils, Error, Result};
use ruma::{
events::presence::{PresenceEvent, PresenceEventContent},
presence::PresenceState,
UInt, UserId,
};
use serde::{Deserialize, Serialize};
use crate::users;
/// Represents data required to be kept in order to implement the presence
/// specification.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub(super) struct Presence {
state: PresenceState,
currently_active: bool,
last_active_ts: u64,
status_msg: Option<String>,
}
impl Presence {
#[must_use]
pub(super) fn new(
state: PresenceState, currently_active: bool, last_active_ts: u64, status_msg: Option<String>,
) -> Self {
Self {
state,
currently_active,
last_active_ts,
status_msg,
}
}
pub(super) fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
serde_json::from_slice(bytes).map_err(|_| Error::bad_database("Invalid presence data in database"))
}
pub(super) fn to_json_bytes(&self) -> Result<Vec<u8>> {
serde_json::to_vec(self).map_err(|_| Error::bad_database("Could not serialize Presence to JSON"))
}
/// Creates a PresenceEvent from available data.
pub(super) fn to_presence_event(&self, user_id: &UserId, users: &Arc<users::Service>) -> Result<PresenceEvent> {
let now = utils::millis_since_unix_epoch();
let last_active_ago = if self.currently_active {
None
} else {
Some(UInt::new_saturating(now.saturating_sub(self.last_active_ts)))
};
Ok(PresenceEvent {
sender: user_id.to_owned(),
content: PresenceEventContent {
presence: self.state.clone(),
status_msg: self.status_msg.clone(),
currently_active: Some(self.currently_active),
last_active_ago,
displayname: users.displayname(user_id)?,
avatar_url: users.avatar_url(user_id)?,
},
})
}
}
+16 -17
View File
@@ -3,7 +3,7 @@ mod data;
use std::{fmt::Debug, mem, sync::Arc};
use bytes::BytesMut;
use conduit::{debug_info, info, trace, utils::string_from_bytes, warn, Error, PduEvent, Result};
use conduit::{debug_error, err, trace, utils::string_from_bytes, warn, Err, PduEvent, Result};
use ipaddress::IPAddress;
use ruma::{
api::{
@@ -84,8 +84,9 @@ impl Service {
let http_request = request
.try_into_http_request::<BytesMut>(&dest, SendAccessToken::IfRequired(""), &VERSIONS)
.map_err(|e| {
warn!("Failed to find destination {dest} for push gateway: {e}");
Error::BadServerResponse("Invalid push gateway destination")
err!(BadServerResponse(warn!(
"Failed to find destination {dest} for push gateway: {e}"
)))
})?
.map(BytesMut::freeze);
@@ -95,7 +96,7 @@ impl Service {
trace!("Checking request URL for IP");
if let Ok(ip) = IPAddress::parse(url_host) {
if !self.services.globals.valid_cidr_range(&ip) {
return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
return Err!(BadServerResponse("Not allowed to send requests to this IP"));
}
}
}
@@ -110,7 +111,7 @@ impl Service {
if let Some(remote_addr) = response.remote_addr() {
if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) {
if !self.services.globals.valid_cidr_range(&ip) {
return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
return Err!(BadServerResponse("Not allowed to send requests to this IP"));
}
}
}
@@ -129,10 +130,10 @@ impl Service {
let body = response.bytes().await?; // TODO: handle timeout
if !status.is_success() {
info!("Push gateway {dest} returned unsuccessful HTTP response ({status})");
debug_info!("Push gateway response body: {:?}", string_from_bytes(&body));
return Err(Error::BadServerResponse("Push gateway returned unsuccessful response"));
debug_error!("Push gateway response body: {:?}", string_from_bytes(&body));
return Err!(BadServerResponse(error!(
"Push gateway {dest} returned unsuccessful HTTP response: {status}"
)));
}
let response = T::IncomingResponse::try_from_http_response(
@@ -140,13 +141,11 @@ impl Service {
.body(body)
.expect("reqwest body is valid http body"),
);
response.map_err(|e| {
warn!("Push gateway {dest} returned invalid response bytes: {e}");
Error::BadServerResponse("Push gateway returned bad/invalid response")
})
response
.map_err(|e| err!(BadServerResponse(error!("Push gateway {dest} returned invalid response: {e}"))))
},
Err(e) => {
warn!("Could not send request to pusher {dest}: {e}");
debug_error!("Could not send request to pusher {dest}: {e}");
Err(e.into())
},
}
@@ -165,7 +164,7 @@ impl Service {
.room_state_get(&pdu.room_id, &StateEventType::RoomPowerLevels, "")?
.map(|ev| {
serde_json::from_str(ev.content.get())
.map_err(|_| Error::bad_database("invalid m.room.power_levels event"))
.map_err(|e| err!(Database("invalid m.room.power_levels event: {e:?}")))
})
.transpose()?
.unwrap_or_default();
@@ -181,8 +180,8 @@ impl Service {
};
if notify.is_some() {
return Err(Error::bad_database(
r#"Malformed pushrule contains more than one of these actions: ["dont_notify", "notify", "coalesce"]"#,
return Err!(Database(
r#"Malformed pushrule contains more than one of these actions: ["dont_notify", "notify", "coalesce"]"#
));
}
+5 -7
View File
@@ -4,7 +4,7 @@ use std::{
sync::Arc,
};
use conduit::{debug, debug_error, debug_info, debug_warn, trace, Err, Error, Result};
use conduit::{debug, debug_error, debug_info, debug_warn, err, trace, Err, Result};
use hickory_resolver::{error::ResolveError, lookup::SrvLookup};
use ipaddress::IPAddress;
use ruma::ServerName;
@@ -189,7 +189,7 @@ impl super::Service {
.services
.client
.well_known
.get(&format!("https://{dest}/.well-known/matrix/server"))
.get(format!("https://{dest}/.well-known/matrix/server"))
.send()
.await;
@@ -329,10 +329,8 @@ impl super::Service {
dest.is_ip_literal() || !IPAddress::is_valid(dest.host()),
"Destination is not an IP literal."
);
let ip = IPAddress::parse(dest.host()).map_err(|e| {
debug_error!("Failed to parse IP literal from string: {}", e);
Error::BadServerResponse("Invalid IP address")
})?;
let ip = IPAddress::parse(dest.host())
.map_err(|e| err!(BadServerResponse(debug_error!("Failed to parse IP literal from string: {e}"))))?;
self.validate_ip(&ip)?;
@@ -341,7 +339,7 @@ impl super::Service {
pub(crate) fn validate_ip(&self, ip: &IPAddress) -> Result<()> {
if !self.services.globals.valid_cidr_range(ip) {
return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
return Err!(BadServerResponse("Not allowed to send requests to this IP"));
}
Ok(())
+16 -13
View File
@@ -1,5 +1,4 @@
mod parse_incoming_pdu;
mod signing_keys;
use std::{
collections::{hash_map, BTreeMap, HashMap, HashSet},
@@ -35,7 +34,7 @@ use ruma::{
use tokio::sync::RwLock;
use super::state_compressor::CompressedStateEvent;
use crate::{globals, rooms, sending, Dep};
use crate::{globals, rooms, sending, server_keys, Dep};
pub struct Service {
services: Services,
@@ -50,6 +49,7 @@ struct Services {
metadata: Dep<rooms::metadata::Service>,
outlier: Dep<rooms::outlier::Service>,
pdu_metadata: Dep<rooms::pdu_metadata::Service>,
server_keys: Dep<server_keys::Service>,
short: Dep<rooms::short::Service>,
state: Dep<rooms::state::Service>,
state_accessor: Dep<rooms::state_accessor::Service>,
@@ -77,6 +77,7 @@ impl crate::Service for Service {
auth_chain: args.depend::<rooms::auth_chain::Service>("rooms::auth_chain"),
metadata: args.depend::<rooms::metadata::Service>("rooms::metadata"),
outlier: args.depend::<rooms::outlier::Service>("rooms::outlier"),
server_keys: args.depend::<server_keys::Service>("server_keys"),
pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
short: args.depend::<rooms::short::Service>("rooms::short"),
state: args.depend::<rooms::state::Service>("rooms::state"),
@@ -1187,17 +1188,19 @@ impl Service {
// We go through all the signatures we see on the PDUs and their unresolved
// dependencies and fetch the corresponding signing keys
self.fetch_required_signing_keys(
events_with_auth_events
.iter()
.flat_map(|(_id, _local_pdu, events)| events)
.map(|(_event_id, event)| event),
pub_key_map,
)
.await
.unwrap_or_else(|e| {
warn!("Could not fetch all signatures for PDUs from {}: {:?}", origin, e);
});
self.services
.server_keys
.fetch_required_signing_keys(
events_with_auth_events
.iter()
.flat_map(|(_id, _local_pdu, events)| events)
.map(|(_event_id, event)| event),
pub_key_map,
)
.await
.unwrap_or_else(|e| {
warn!("Could not fetch all signatures for PDUs from {}: {:?}", origin, e);
});
let mut pdus = Vec::with_capacity(events_with_auth_events.len());
for (id, local_pdu, events_in_reverse_order) in events_with_auth_events {
@@ -1,18 +1,18 @@
use conduit::{pdu::gen_event_id_canonical_json, warn, Err, Error, Result};
use ruma::{api::client::error::ErrorKind, CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId};
use conduit::{debug_warn, err, pdu::gen_event_id_canonical_json, Err, Result};
use ruma::{CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId};
use serde_json::value::RawValue as RawJsonValue;
impl super::Service {
pub fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<(OwnedEventId, CanonicalJsonObject, OwnedRoomId)> {
let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
warn!("Error parsing incoming event {pdu:?}: {e:?}");
Error::BadServerResponse("Invalid PDU in server response")
debug_warn!("Error parsing incoming event {pdu:#?}");
err!(BadServerResponse("Error parsing incoming event {e:?}"))
})?;
let room_id: OwnedRoomId = value
.get("room_id")
.and_then(|id| RoomId::parse(id.as_str()?).ok())
.ok_or(Error::BadRequest(ErrorKind::InvalidParam, "Invalid room id in pdu"))?;
.ok_or(err!(Request(InvalidParam("Invalid room id in pdu"))))?;
let Ok(room_version_id) = self.services.state.get_room_version(&room_id) else {
return Err!("Server is not in room {room_id}");
+6 -4
View File
@@ -12,17 +12,19 @@ use ruma::{DeviceId, OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId};
use self::data::Data;
pub struct Service {
pub lazy_load_waiting: Mutex<LazyLoadWaiting>,
db: Data,
#[allow(clippy::type_complexity)]
pub lazy_load_waiting: Mutex<HashMap<(OwnedUserId, OwnedDeviceId, OwnedRoomId, PduCount), HashSet<OwnedUserId>>>,
}
type LazyLoadWaiting = HashMap<LazyLoadWaitingKey, LazyLoadWaitingVal>;
type LazyLoadWaitingKey = (OwnedUserId, OwnedDeviceId, OwnedRoomId, PduCount);
type LazyLoadWaitingVal = HashSet<OwnedUserId>;
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
Ok(Arc::new(Self {
db: Data::new(args.db),
lazy_load_waiting: Mutex::new(HashMap::new()),
db: Data::new(args.db),
}))
}
+4 -2
View File
@@ -41,7 +41,7 @@ use tokio::sync::RwLock;
use self::data::Data;
use crate::{
account_data, admin, appservice, appservice::NamespaceRegex, globals, pusher, rooms,
rooms::state_compressor::CompressedStateEvent, sending, Dep,
rooms::state_compressor::CompressedStateEvent, sending, server_keys, Dep,
};
// Update Relationships
@@ -86,6 +86,7 @@ struct Services {
pdu_metadata: Dep<rooms::pdu_metadata::Service>,
read_receipt: Dep<rooms::read_receipt::Service>,
sending: Dep<sending::Service>,
server_keys: Dep<server_keys::Service>,
user: Dep<rooms::user::Service>,
pusher: Dep<pusher::Service>,
threads: Dep<rooms::threads::Service>,
@@ -114,6 +115,7 @@ impl crate::Service for Service {
pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
read_receipt: args.depend::<rooms::read_receipt::Service>("rooms::read_receipt"),
sending: args.depend::<sending::Service>("sending"),
server_keys: args.depend::<server_keys::Service>("server_keys"),
user: args.depend::<rooms::user::Service>("rooms::user"),
pusher: args.depend::<pusher::Service>("pusher"),
threads: args.depend::<rooms::threads::Service>("rooms::threads"),
@@ -1181,7 +1183,7 @@ impl Service {
}
self.services
.event_handler
.server_keys
.fetch_required_signing_keys([&value], pub_key_map)
.await?;
+9 -12
View File
@@ -1,7 +1,7 @@
use std::{fmt::Debug, mem};
use bytes::BytesMut;
use conduit::{debug_error, trace, utils, warn, Error, Result};
use conduit::{debug_error, err, trace, utils, warn, Err, Result};
use reqwest::Client;
use ruma::api::{appservice::Registration, IncomingResponse, MatrixVersion, OutgoingRequest, SendAccessToken};
@@ -26,10 +26,7 @@ where
let hs_token = registration.hs_token.as_str();
let mut http_request = request
.try_into_http_request::<BytesMut>(&dest, SendAccessToken::IfRequired(hs_token), &VERSIONS)
.map_err(|e| {
warn!("Failed to find destination {dest}: {e}");
Error::BadServerResponse("Invalid appservice destination")
})?
.map_err(|e| err!(BadServerResponse(warn!("Failed to find destination {dest}: {e}"))))?
.map(BytesMut::freeze);
let mut parts = http_request.uri().clone().into_parts();
@@ -69,13 +66,11 @@ where
let body = response.bytes().await?; // TODO: handle timeout
if !status.is_success() {
warn!(
debug_error!("Appservice response bytes: {:?}", utils::string_from_bytes(&body));
return Err!(BadServerResponse(error!(
"Appservice \"{}\" returned unsuccessful HTTP response {status} at {dest}",
registration.id
);
debug_error!("Appservice response bytes: {:?}", utils::string_from_bytes(&body));
return Err(Error::BadServerResponse("Appservice returned unsuccessful HTTP response"));
)));
}
let response = T::IncomingResponse::try_from_http_response(
@@ -85,7 +80,9 @@ where
);
response.map(Some).map_err(|e| {
warn!("Appservice \"{}\" returned invalid response bytes {dest}: {e}", registration.id);
Error::BadServerResponse("Appservice returned bad/invalid response")
err!(BadServerResponse(error!(
"Appservice \"{}\" returned invalid response bytes {dest}: {e}",
registration.id
)))
})
}
-343
View File
@@ -1,343 +0,0 @@
use std::{
fmt::Debug,
net::{IpAddr, SocketAddr},
};
use conduit::{debug, debug_error, debug_info, debug_warn, trace, Err, Error, Result};
use hickory_resolver::{error::ResolveError, lookup::SrvLookup};
use ipaddress::IPAddress;
use ruma::ServerName;
use crate::{
resolver::{add_port_to_hostname, get_ip_with_port, CachedDest, CachedOverride, FedDest},
services,
};
#[derive(Clone, Debug)]
pub(crate) struct ActualDest {
pub(crate) dest: FedDest,
pub(crate) host: String,
pub(crate) string: String,
pub(crate) cached: bool,
}
#[tracing::instrument(skip_all, name = "resolve")]
pub(crate) async fn get_actual_dest(server_name: &ServerName) -> Result<ActualDest> {
let cached;
let cached_result = services().resolver.get_cached_destination(server_name);
let CachedDest {
dest,
host,
..
} = if let Some(result) = cached_result {
cached = true;
result
} else {
cached = false;
validate_dest(server_name)?;
resolve_actual_dest(server_name, true).await?
};
let string = dest.clone().into_https_string();
Ok(ActualDest {
dest,
host,
string,
cached,
})
}
/// Returns: `actual_destination`, host header
/// Implemented according to the specification at <https://matrix.org/docs/spec/server_server/r0.1.4#resolving-server-names>
/// Numbers in comments below refer to bullet points in linked section of
/// specification
#[tracing::instrument(skip_all, name = "actual")]
pub async fn resolve_actual_dest(dest: &ServerName, cache: bool) -> Result<CachedDest> {
trace!("Finding actual destination for {dest}");
let mut host = dest.as_str().to_owned();
let actual_dest = match get_ip_with_port(dest.as_str()) {
Some(host_port) => actual_dest_1(host_port)?,
None => {
if let Some(pos) = dest.as_str().find(':') {
actual_dest_2(dest, cache, pos).await?
} else if let Some(delegated) = request_well_known(dest.as_str()).await? {
actual_dest_3(&mut host, cache, delegated).await?
} else if let Some(overrider) = query_srv_record(dest.as_str()).await? {
actual_dest_4(&host, cache, overrider).await?
} else {
actual_dest_5(dest, cache).await?
}
},
};
// Can't use get_ip_with_port here because we don't want to add a port
// to an IP address if it wasn't specified
let host = if let Ok(addr) = host.parse::<SocketAddr>() {
FedDest::Literal(addr)
} else if let Ok(addr) = host.parse::<IpAddr>() {
FedDest::Named(addr.to_string(), ":8448".to_owned())
} else if let Some(pos) = host.find(':') {
let (host, port) = host.split_at(pos);
FedDest::Named(host.to_owned(), port.to_owned())
} else {
FedDest::Named(host, ":8448".to_owned())
};
debug!("Actual destination: {actual_dest:?} hostname: {host:?}");
Ok(CachedDest {
dest: actual_dest,
host: host.into_uri_string(),
expire: CachedDest::default_expire(),
})
}
fn actual_dest_1(host_port: FedDest) -> Result<FedDest> {
debug!("1: IP literal with provided or default port");
Ok(host_port)
}
async fn actual_dest_2(dest: &ServerName, cache: bool, pos: usize) -> Result<FedDest> {
debug!("2: Hostname with included port");
let (host, port) = dest.as_str().split_at(pos);
conditional_query_and_cache_override(host, host, port.parse::<u16>().unwrap_or(8448), cache).await?;
Ok(FedDest::Named(host.to_owned(), port.to_owned()))
}
async fn actual_dest_3(host: &mut String, cache: bool, delegated: String) -> Result<FedDest> {
debug!("3: A .well-known file is available");
*host = add_port_to_hostname(&delegated).into_uri_string();
match get_ip_with_port(&delegated) {
Some(host_and_port) => actual_dest_3_1(host_and_port),
None => {
if let Some(pos) = delegated.find(':') {
actual_dest_3_2(cache, delegated, pos).await
} else {
trace!("Delegated hostname has no port in this branch");
if let Some(overrider) = query_srv_record(&delegated).await? {
actual_dest_3_3(cache, delegated, overrider).await
} else {
actual_dest_3_4(cache, delegated).await
}
}
},
}
}
fn actual_dest_3_1(host_and_port: FedDest) -> Result<FedDest> {
debug!("3.1: IP literal in .well-known file");
Ok(host_and_port)
}
async fn actual_dest_3_2(cache: bool, delegated: String, pos: usize) -> Result<FedDest> {
debug!("3.2: Hostname with port in .well-known file");
let (host, port) = delegated.split_at(pos);
conditional_query_and_cache_override(host, host, port.parse::<u16>().unwrap_or(8448), cache).await?;
Ok(FedDest::Named(host.to_owned(), port.to_owned()))
}
async fn actual_dest_3_3(cache: bool, delegated: String, overrider: FedDest) -> Result<FedDest> {
debug!("3.3: SRV lookup successful");
let force_port = overrider.port();
conditional_query_and_cache_override(&delegated, &overrider.hostname(), force_port.unwrap_or(8448), cache).await?;
if let Some(port) = force_port {
Ok(FedDest::Named(delegated, format!(":{port}")))
} else {
Ok(add_port_to_hostname(&delegated))
}
}
async fn actual_dest_3_4(cache: bool, delegated: String) -> Result<FedDest> {
debug!("3.4: No SRV records, just use the hostname from .well-known");
conditional_query_and_cache_override(&delegated, &delegated, 8448, cache).await?;
Ok(add_port_to_hostname(&delegated))
}
async fn actual_dest_4(host: &str, cache: bool, overrider: FedDest) -> Result<FedDest> {
debug!("4: No .well-known; SRV record found");
let force_port = overrider.port();
conditional_query_and_cache_override(host, &overrider.hostname(), force_port.unwrap_or(8448), cache).await?;
if let Some(port) = force_port {
Ok(FedDest::Named(host.to_owned(), format!(":{port}")))
} else {
Ok(add_port_to_hostname(host))
}
}
async fn actual_dest_5(dest: &ServerName, cache: bool) -> Result<FedDest> {
debug!("5: No SRV record found");
conditional_query_and_cache_override(dest.as_str(), dest.as_str(), 8448, cache).await?;
Ok(add_port_to_hostname(dest.as_str()))
}
#[tracing::instrument(skip_all, name = "well-known")]
async fn request_well_known(dest: &str) -> Result<Option<String>> {
trace!("Requesting well known for {dest}");
if !services().resolver.has_cached_override(dest) {
query_and_cache_override(dest, dest, 8448).await?;
}
let response = services()
.client
.well_known
.get(&format!("https://{dest}/.well-known/matrix/server"))
.send()
.await;
trace!("response: {:?}", response);
if let Err(e) = &response {
debug!("error: {e:?}");
return Ok(None);
}
let response = response?;
if !response.status().is_success() {
debug!("response not 2XX");
return Ok(None);
}
let text = response.text().await?;
trace!("response text: {:?}", text);
if text.len() >= 12288 {
debug_warn!("response contains junk");
return Ok(None);
}
let body: serde_json::Value = serde_json::from_str(&text).unwrap_or_default();
let m_server = body
.get("m.server")
.unwrap_or(&serde_json::Value::Null)
.as_str()
.unwrap_or_default();
if ruma::identifiers_validation::server_name::validate(m_server).is_err() {
debug_error!("response content missing or invalid");
return Ok(None);
}
debug_info!("{:?} found at {:?}", dest, m_server);
Ok(Some(m_server.to_owned()))
}
#[inline]
async fn conditional_query_and_cache_override(overname: &str, hostname: &str, port: u16, cache: bool) -> Result<()> {
if cache {
query_and_cache_override(overname, hostname, port).await
} else {
Ok(())
}
}
#[tracing::instrument(skip_all, name = "ip")]
async fn query_and_cache_override(overname: &'_ str, hostname: &'_ str, port: u16) -> Result<()> {
match services()
.resolver
.resolver
.lookup_ip(hostname.to_owned())
.await
{
Err(e) => handle_resolve_error(&e),
Ok(override_ip) => {
if hostname != overname {
debug_info!("{overname:?} overriden by {hostname:?}");
}
services().resolver.set_cached_override(
overname.to_owned(),
CachedOverride {
ips: override_ip.iter().collect(),
port,
expire: CachedOverride::default_expire(),
},
);
Ok(())
},
}
}
#[tracing::instrument(skip_all, name = "srv")]
async fn query_srv_record(hostname: &'_ str) -> Result<Option<FedDest>> {
fn handle_successful_srv(srv: &SrvLookup) -> Option<FedDest> {
srv.iter().next().map(|result| {
FedDest::Named(
result.target().to_string().trim_end_matches('.').to_owned(),
format!(":{}", result.port()),
)
})
}
async fn lookup_srv(hostname: &str) -> Result<SrvLookup, ResolveError> {
debug!("querying SRV for {:?}", hostname);
let hostname = hostname.trim_end_matches('.');
services()
.resolver
.resolver
.srv_lookup(hostname.to_owned())
.await
}
let hostnames = [format!("_matrix-fed._tcp.{hostname}."), format!("_matrix._tcp.{hostname}.")];
for hostname in hostnames {
match lookup_srv(&hostname).await {
Ok(result) => return Ok(handle_successful_srv(&result)),
Err(e) => handle_resolve_error(&e)?,
}
}
Ok(None)
}
#[allow(clippy::single_match_else)]
fn handle_resolve_error(e: &ResolveError) -> Result<()> {
use hickory_resolver::error::ResolveErrorKind;
match *e.kind() {
ResolveErrorKind::NoRecordsFound {
..
} => {
// Raise to debug_warn if we can find out the result wasn't from cache
debug!("{e}");
Ok(())
},
_ => Err!(error!("DNS {e}")),
}
}
fn validate_dest(dest: &ServerName) -> Result<()> {
if dest == services().globals.server_name() {
return Err!("Won't send federation request to ourselves");
}
if dest.is_ip_literal() || IPAddress::is_valid(dest.host()) {
validate_dest_ip_literal(dest)?;
}
Ok(())
}
fn validate_dest_ip_literal(dest: &ServerName) -> Result<()> {
trace!("Destination is an IP literal, checking against IP range denylist.",);
debug_assert!(
dest.is_ip_literal() || !IPAddress::is_valid(dest.host()),
"Destination is not an IP literal."
);
let ip = IPAddress::parse(dest.host()).map_err(|e| {
debug_error!("Failed to parse IP literal from string: {}", e);
Error::BadServerResponse("Invalid IP address")
})?;
validate_ip(&ip)?;
Ok(())
}
pub(crate) fn validate_ip(ip: &IPAddress) -> Result<()> {
if !services().globals.valid_cidr_range(ip) {
return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
}
Ok(())
}
+3 -6
View File
@@ -1,7 +1,7 @@
use std::{fmt::Debug, mem};
use conduit::{
debug, debug_error, debug_warn, error::inspect_debug_log, trace, utils::string::EMPTY, Err, Error, Result,
debug, debug_error, debug_warn, err, error::inspect_debug_log, trace, utils::string::EMPTY, Err, Error, Result,
};
use http::{header::AUTHORIZATION, HeaderValue};
use ipaddress::IPAddress;
@@ -62,7 +62,7 @@ impl super::Service {
trace!("Preparing request");
let mut http_request = req
.try_into_http_request::<Vec<u8>>(&actual.string, SATIR, &VERSIONS)
.map_err(|_| Error::BadServerResponse("Invalid destination"))?;
.map_err(|e| err!(BadServerResponse("Invalid destination: {e:?}")))?;
sign_request::<T>(&self.services.globals, dest, &mut http_request);
@@ -139,10 +139,7 @@ where
);
}
match response {
Err(_) => Err(Error::BadServerResponse("Server returned bad 200 response.")),
Ok(response) => Ok(response),
}
response.map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}")))
}
fn handle_error<T>(
@@ -1,9 +1,10 @@
use std::{
collections::{BTreeMap, HashMap, HashSet},
sync::Arc,
time::{Duration, SystemTime},
};
use conduit::{debug, error, info, trace, warn, Error, Result};
use conduit::{debug, debug_error, debug_warn, err, error, info, trace, warn, Err, Result};
use futures_util::{stream::FuturesUnordered, StreamExt};
use ruma::{
api::federation::{
@@ -21,7 +22,31 @@ use ruma::{
use serde_json::value::RawValue as RawJsonValue;
use tokio::sync::{RwLock, RwLockWriteGuard};
impl super::Service {
use crate::{globals, sending, Dep};
pub struct Service {
services: Services,
}
struct Services {
globals: Dep<globals::Service>,
sending: Dep<sending::Service>,
}
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
Ok(Arc::new(Self {
services: Services {
globals: args.depend::<globals::Service>("globals"),
sending: args.depend::<sending::Service>("sending"),
},
}))
}
fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
}
impl Service {
pub async fn fetch_required_signing_keys<'a, E>(
&'a self, events: E, pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
) -> Result<()>
@@ -32,13 +57,13 @@ impl super::Service {
for event in events {
for (signature_server, signature) in event
.get("signatures")
.ok_or(Error::BadServerResponse("No signatures in server response pdu."))?
.ok_or(err!(BadServerResponse("No signatures in server response pdu.")))?
.as_object()
.ok_or(Error::BadServerResponse("Invalid signatures object in server response pdu."))?
.ok_or(err!(BadServerResponse("Invalid signatures object in server response pdu.")))?
{
let signature_object = signature.as_object().ok_or(Error::BadServerResponse(
let signature_object = signature.as_object().ok_or(err!(BadServerResponse(
"Invalid signatures content object in server response pdu.",
))?;
)))?;
for signature_id in signature_object.keys() {
server_key_ids
@@ -69,10 +94,12 @@ impl super::Service {
.map(|(signature_server, signature_ids)| async {
let fetch_res = self
.fetch_signing_keys_for_server(
signature_server.as_str().try_into().map_err(|_| {
signature_server.as_str().try_into().map_err(|e| {
(
signature_server.clone(),
Error::BadServerResponse("Invalid servername in signatures of server response pdu."),
err!(BadServerResponse(
"Invalid servername in signatures of server response pdu: {e:?}"
)),
)
})?,
signature_ids.into_iter().collect(), // HashSet to Vec
@@ -82,7 +109,9 @@ impl super::Service {
match fetch_res {
Ok(keys) => Ok((signature_server, keys)),
Err(e) => {
warn!("Signature verification failed: Could not fetch signing key for {signature_server}: {e}",);
debug_error!(
"Signature verification failed: Could not fetch signing key for {signature_server}: {e}",
);
Err((signature_server, e))
},
}
@@ -98,7 +127,7 @@ impl super::Service {
.insert(signature_server.clone(), keys);
},
Err((signature_server, e)) => {
warn!("Failed to fetch keys for {}: {:?}", signature_server, e);
debug_warn!("Failed to fetch keys for {signature_server}: {e:?}");
},
}
}
@@ -116,35 +145,37 @@ impl super::Service {
pub_key_map: &mut RwLockWriteGuard<'_, BTreeMap<String, BTreeMap<String, Base64>>>,
) -> Result<()> {
let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
error!("Invalid PDU in server response: {:?}: {:?}", pdu, e);
Error::BadServerResponse("Invalid PDU in server response")
debug_error!("Invalid PDU in server response: {pdu:#?}");
err!(BadServerResponse(error!("Invalid PDU in server response: {e:?}")))
})?;
let signatures = value
.get("signatures")
.ok_or(Error::BadServerResponse("No signatures in server response pdu."))?
.ok_or(err!(BadServerResponse("No signatures in server response pdu.")))?
.as_object()
.ok_or(Error::BadServerResponse("Invalid signatures object in server response pdu."))?;
.ok_or(err!(BadServerResponse("Invalid signatures object in server response pdu.")))?;
for (signature_server, signature) in signatures {
let signature_object = signature.as_object().ok_or(Error::BadServerResponse(
let signature_object = signature.as_object().ok_or(err!(BadServerResponse(
"Invalid signatures content object in server response pdu.",
))?;
)))?;
let signature_ids = signature_object.keys().cloned().collect::<Vec<_>>();
let contains_all_ids =
|keys: &BTreeMap<String, Base64>| signature_ids.iter().all(|id| keys.contains_key(id));
let origin = <&ServerName>::try_from(signature_server.as_str())
.map_err(|_| Error::BadServerResponse("Invalid servername in signatures of server response pdu."))?;
let origin = <&ServerName>::try_from(signature_server.as_str()).map_err(|e| {
err!(BadServerResponse(
"Invalid servername in signatures of server response pdu: {e:?}"
))
})?;
if servers.contains_key(origin) || pub_key_map.contains_key(origin.as_str()) {
continue;
}
debug!("Loading signing keys for {}", origin);
debug!("Loading signing keys for {origin}");
let result: BTreeMap<_, _> = self
.services
.globals
@@ -154,7 +185,7 @@ impl super::Service {
.collect();
if !contains_all_ids(&result) {
debug!("Signing key not loaded for {}", origin);
debug_warn!("Signing key not loaded for {origin}");
servers.insert(origin.to_owned(), BTreeMap::new());
}
@@ -171,7 +202,7 @@ impl super::Service {
pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
) -> Result<()> {
for server in self.services.globals.trusted_servers() {
debug!("Asking batch signing keys from trusted server {}", server);
debug!("Asking batch signing keys from trusted server {server}");
match self
.services
.sending
@@ -184,14 +215,16 @@ impl super::Service {
.await
{
Ok(keys) => {
debug!("Got signing keys: {:?}", keys);
debug!("Got signing keys: {keys:?}");
let mut pkm = pub_key_map.write().await;
for k in keys.server_keys {
let k = match k.deserialize() {
Ok(key) => key,
Err(e) => {
warn!("Received error {e} while fetching keys from trusted server {server}");
warn!("{}", k.into_json());
warn!(
"Received error {e} while fetching keys from trusted server {server}: {:#?}",
k.into_json()
);
continue;
},
};
@@ -211,13 +244,10 @@ impl super::Service {
pkm.insert(k.server_name.to_string(), result);
}
},
Err(e) => {
warn!(
"Failed sending batched key request to trusted key server {server} for the remote servers \
{:?}: {e}",
servers
);
},
Err(e) => error!(
"Failed sending batched key request to trusted key server {server} for the remote servers \
{servers:?}: {e}"
),
}
}
@@ -453,7 +483,6 @@ impl super::Service {
}
} else {
info!("query_trusted_key_servers_first is set to false, querying {origin} first");
debug!("Asking {origin} for their signing keys over federation");
if let Some(server_key) = self
.services
@@ -511,7 +540,7 @@ impl super::Service {
.filter_map(|e| e.deserialize().ok())
.collect::<Vec<_>>()
}) {
debug!("Got signing keys: {:?}", server_keys);
debug!("Got signing keys: {server_keys:?}");
for k in server_keys {
self.services
.globals
@@ -536,7 +565,6 @@ impl super::Service {
}
}
warn!("Failed to find public key for server: {origin}");
Err(Error::BadServerResponse("Failed to find public key for server"))
Err!(BadServerResponse(warn!("Failed to find public key for server {origin:?}")))
}
}
+47 -32
View File
@@ -3,7 +3,7 @@ use std::{
collections::BTreeMap,
fmt::Write,
ops::Deref,
sync::{Arc, OnceLock, RwLock},
sync::{Arc, OnceLock, RwLock, Weak},
};
use async_trait::async_trait;
@@ -53,77 +53,92 @@ pub(crate) struct Args<'a> {
/// Circular-dependencies between services require this indirection.
pub(crate) struct Dep<T> {
dep: OnceLock<Arc<T>>,
service: Arc<Map>,
service: Weak<Map>,
name: &'static str,
}
pub(crate) type Map = RwLock<BTreeMap<String, MapVal>>;
pub(crate) type MapVal = (Arc<dyn Service>, Arc<dyn Any + Send + Sync>);
pub(crate) type Map = RwLock<MapType>;
pub(crate) type MapType = BTreeMap<MapKey, MapVal>;
pub(crate) type MapVal = (Weak<dyn Service>, Weak<dyn Any + Send + Sync>);
pub(crate) type MapKey = String;
impl<T: Send + Sync + 'static> Deref for Dep<T> {
type Target = Arc<T>;
/// Dereference a dependency. The dependency must be ready or panics.
fn deref(&self) -> &Self::Target {
self.dep
.get_or_init(|| require::<T>(&self.service, self.name))
self.dep.get_or_init(|| {
let service = self
.service
.upgrade()
.expect("services map exists for dependency initialization.");
require::<T>(&service, self.name)
})
}
}
impl Args<'_> {
impl<'a> Args<'a> {
/// Create a lazy-reference to a service when constructing another Service.
pub(crate) fn depend<T: Send + Sync + 'static>(&self, name: &'static str) -> Dep<T> {
pub(crate) fn depend<T: Send + Sync + 'a + 'static>(&'a self, name: &'static str) -> Dep<T> {
Dep::<T> {
dep: OnceLock::new(),
service: self.service.clone(),
service: Arc::downgrade(self.service),
name,
}
}
/// Create a reference immediately to a service when constructing another
/// Service. The other service must be constructed.
pub(crate) fn require<T: Send + Sync + 'static>(&self, name: &str) -> Arc<T> { require::<T>(self.service, name) }
pub(crate) fn require<T: Send + Sync + 'a + 'static>(&'a self, name: &'static str) -> Arc<T> {
require::<T>(self.service, name)
}
}
/// Reference a Service by name. Panics if the Service does not exist or was
/// incorrectly cast.
pub(crate) fn require<T: Send + Sync + 'static>(map: &Map, name: &str) -> Arc<T> {
pub(crate) fn require<'a, 'b, T: Send + Sync + 'a + 'b + 'static>(map: &'b Map, name: &'a str) -> Arc<T> {
try_get::<T>(map, name)
.inspect_err(inspect_log)
.expect("Failure to reference service required by another service.")
}
/// Reference a Service by name. Returns Err if the Service does not exist or
/// was incorrectly cast.
pub(crate) fn try_get<T: Send + Sync + 'static>(map: &Map, name: &str) -> Result<Arc<T>> {
map.read()
.expect("locked for reading")
.get(name)
.map_or_else(
|| Err!("Service {name:?} does not exist or has not been built yet."),
|(_, s)| {
s.clone()
.downcast::<T>()
.map_err(|_| err!("Service {name:?} must be correctly downcast."))
},
)
}
/// Reference a Service by name. Returns None if the Service does not exist, but
/// panics if incorrectly cast.
///
/// # Panics
/// Incorrect type is not a silent failure (None) as the type never has a reason
/// to be incorrect.
pub(crate) fn get<T: Send + Sync + 'static>(map: &Map, name: &str) -> Option<Arc<T>> {
pub(crate) fn get<'a, 'b, T: Send + Sync + 'a + 'b + 'static>(map: &'b Map, name: &'a str) -> Option<Arc<T>> {
map.read()
.expect("locked for reading")
.get(name)
.map(|(_, s)| {
s.clone()
.downcast::<T>()
.expect("Service must be correctly downcast.")
})
s.upgrade().map(|s| {
s.downcast::<T>()
.expect("Service must be correctly downcast.")
})
})?
}
/// Reference a Service by name. Returns Err if the Service does not exist or
/// was incorrectly cast.
pub(crate) fn try_get<'a, 'b, T: Send + Sync + 'a + 'b + 'static>(map: &'b Map, name: &'a str) -> Result<Arc<T>> {
map.read()
.expect("locked for reading")
.get(name)
.map_or_else(
|| Err!("Service {name:?} does not exist or has not been built yet."),
|(_, s)| {
s.upgrade().map_or_else(
|| Err!("Service {name:?} no longer exists."),
|s| {
s.downcast::<T>()
.map_err(|_| err!("Service {name:?} must be correctly downcast."))
},
)
},
)
}
/// Utility for service implementations; see Service::name() in the trait.
+21 -12
View File
@@ -12,7 +12,7 @@ use tokio::sync::Mutex;
use crate::{
account_data, admin, appservice, client, emergency, globals, key_backups,
manager::Manager,
media, presence, pusher, resolver, rooms, sending, service,
media, presence, pusher, resolver, rooms, sending, server_keys, service,
service::{Args, Map, Service},
transaction_ids, uiaa, updates, users,
};
@@ -31,6 +31,7 @@ pub struct Services {
pub resolver: Arc<resolver::Service>,
pub rooms: rooms::Service,
pub sending: Arc<sending::Service>,
pub server_keys: Arc<server_keys::Service>,
pub transaction_ids: Arc<transaction_ids::Service>,
pub uiaa: Arc<uiaa::Service>,
pub updates: Arc<updates::Service>,
@@ -94,6 +95,7 @@ impl Services {
user: build!(rooms::user::Service),
},
sending: build!(sending::Service),
server_keys: build!(server_keys::Service),
transaction_ids: build!(transaction_ids::Service),
uiaa: build!(uiaa::Service),
updates: build!(updates::Service),
@@ -109,7 +111,7 @@ impl Services {
pub async fn start(self: &Arc<Self>) -> Result<Arc<Self>> {
debug_info!("Starting services...");
self.admin.set_services(Some(Arc::clone(self)));
self.admin.set_services(&Some(Arc::clone(self)));
globals::migrations::migrations(self).await?;
self.manager
.lock()
@@ -131,7 +133,7 @@ impl Services {
manager.stop().await;
}
self.admin.set_services(None);
self.admin.set_services(&None);
debug_info!("Services shutdown complete.");
}
@@ -146,7 +148,9 @@ impl Services {
pub async fn clear_cache(&self) {
for (service, ..) in self.service.read().expect("locked for reading").values() {
service.clear_cache();
if let Some(service) = service.upgrade() {
service.clear_cache();
}
}
//TODO
@@ -161,7 +165,9 @@ impl Services {
pub async fn memory_usage(&self) -> Result<String> {
let mut out = String::new();
for (service, ..) in self.service.read().expect("locked for reading").values() {
service.memory_usage(&mut out)?;
if let Some(service) = service.upgrade() {
service.memory_usage(&mut out)?;
}
}
//TODO
@@ -179,27 +185,30 @@ impl Services {
fn interrupt(&self) {
debug!("Interrupting services...");
for (name, (service, ..)) in self.service.read().expect("locked for reading").iter() {
trace!("Interrupting {name}");
service.interrupt();
if let Some(service) = service.upgrade() {
trace!("Interrupting {name}");
service.interrupt();
}
}
}
pub fn try_get<T: Send + Sync + 'static>(&self, name: &str) -> Result<Arc<T>> {
pub fn try_get<'a, 'b, T: Send + Sync + 'a + 'b + 'static>(&'b self, name: &'a str) -> Result<Arc<T>> {
service::try_get::<T>(&self.service, name)
}
pub fn get<T: Send + Sync + 'static>(&self, name: &str) -> Option<Arc<T>> { service::get::<T>(&self.service, name) }
pub fn get<'a, 'b, T: Send + Sync + 'a + 'b + 'static>(&'b self, name: &'a str) -> Option<Arc<T>> {
service::get::<T>(&self.service, name)
}
}
#[allow(clippy::needless_pass_by_value)]
fn add_service(map: &Arc<Map>, s: Arc<dyn Service>, a: Arc<dyn Any + Send + Sync>) {
let name = s.name();
let len = map.read().expect("locked for reading").len();
trace!("built service #{len}: {name:?}");
map.write()
.expect("locked for writing")
.insert(name.to_owned(), (s, a));
.insert(name.to_owned(), (Arc::downgrade(&s), Arc::downgrade(&a)));
}
+2 -1
View File
@@ -246,6 +246,7 @@ impl Data {
/// Adds a new device to a user.
pub(super) fn create_device(
&self, user_id: &UserId, device_id: &DeviceId, token: &str, initial_device_display_name: Option<String>,
client_ip: Option<String>,
) -> Result<()> {
// This method should never be called for nonexistent users. We shouldn't assert
// though...
@@ -266,7 +267,7 @@ impl Data {
&serde_json::to_vec(&Device {
device_id: device_id.into(),
display_name: initial_device_display_name,
last_seen_ip: None, // TODO
last_seen_ip: client_ip,
last_seen_ts: Some(MilliSecondsSinceUnixEpoch::now()),
})
.expect("Device::to_string never fails."),
+17 -14
View File
@@ -26,19 +26,10 @@ use ruma::{
use self::data::Data;
use crate::{admin, rooms, Dep};
pub struct SlidingSyncCache {
lists: BTreeMap<String, SyncRequestList>,
subscriptions: BTreeMap<OwnedRoomId, sync_events::v4::RoomSubscription>,
known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>, // For every room, the roomsince number
extensions: ExtensionsConfig,
}
type DbConnections = Mutex<BTreeMap<(OwnedUserId, OwnedDeviceId, String), Arc<Mutex<SlidingSyncCache>>>>;
pub struct Service {
services: Services,
connections: DbConnections,
pub db: Data,
pub connections: DbConnections,
services: Services,
}
struct Services {
@@ -49,18 +40,29 @@ struct Services {
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
Ok(Arc::new(Self {
connections: StdMutex::new(BTreeMap::new()),
db: Data::new(&args),
services: Services {
admin: args.depend::<admin::Service>("admin"),
state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
},
db: Data::new(&args),
connections: StdMutex::new(BTreeMap::new()),
}))
}
fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
}
type DbConnections = Mutex<BTreeMap<DbConnectionsKey, DbConnectionsVal>>;
type DbConnectionsKey = (OwnedUserId, OwnedDeviceId, String);
type DbConnectionsVal = Arc<Mutex<SlidingSyncCache>>;
struct SlidingSyncCache {
lists: BTreeMap<String, SyncRequestList>,
subscriptions: BTreeMap<OwnedRoomId, sync_events::v4::RoomSubscription>,
known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>, // For every room, the roomsince number
extensions: ExtensionsConfig,
}
impl Service {
/// Check if a user has an account on this homeserver.
#[inline]
@@ -326,9 +328,10 @@ impl Service {
/// Adds a new device to a user.
pub fn create_device(
&self, user_id: &UserId, device_id: &DeviceId, token: &str, initial_device_display_name: Option<String>,
client_ip: Option<String>,
) -> Result<()> {
self.db
.create_device(user_id, device_id, token, initial_device_display_name)
.create_device(user_id, device_id, token, initial_device_display_name, client_ip)
}
/// Removes a device from a user.