Compare commits

..

1 Commits

Author SHA1 Message Date
serinko 1ef33ce797 initialise branch - leaving for later focus now 2025-01-29 12:08:24 +01:00
453 changed files with 14003 additions and 13571 deletions
@@ -26,7 +26,6 @@ jobs:
runs-on: ${{ matrix.platform }}
env:
CARGO_TERM_COLOR: always
RUSTUP_PERMIT_COPY_RENAME: 1
steps:
- uses: actions/checkout@v4
@@ -12,7 +12,6 @@ jobs:
runs-on: arc-ubuntu-22.04
env:
CARGO_TERM_COLOR: always
RUSTUP_PERMIT_COPY_RENAME: 1
steps:
- name: Check out repository code
uses: actions/checkout@v4
-15
View File
@@ -37,7 +37,6 @@ jobs:
env:
CARGO_TERM_COLOR: always
IPINFO_API_TOKEN: ${{ secrets.IPINFO_API_TOKEN }}
RUSTUP_PERMIT_COPY_RENAME: 1
steps:
- name: Install Dependencies (Linux)
run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler
@@ -55,20 +54,6 @@ jobs:
override: true
components: rustfmt, clippy
# To avoid running out of disk space, skip generating debug symbols
- name: Set debug to false (unix)
if: contains(matrix.os, 'ubuntu') || contains(matrix.os, 'mac')
run: |
sed -i.bak 's/\[profile.dev\]/\[profile.dev\]\ndebug = false/' Cargo.toml
git diff
- name: Set debug to false (win)
if: contains(matrix.os, 'windows')
shell: pwsh
run: |
(Get-Content Cargo.toml) -replace '\[profile.dev\]', "`$&`ndebug = false" | Set-Content Cargo.toml
git diff
- name: Check formatting
uses: actions-rs/cargo@v1
with:
-2
View File
@@ -9,8 +9,6 @@ on:
paths:
- 'contracts/**'
- 'common/**'
- 'Cargo.lock'
- 'Cargo.toml'
- '.github/workflows/ci-contracts.yml'
jobs:
-6
View File
@@ -30,12 +30,6 @@ jobs:
override: true
components: rustfmt, clippy
- name: Set debug to false
working-directory: nym-wallet
run: |
sed -i.bak '1s/^/\[profile.dev\]\ndebug = false\n\n/' Cargo.toml
git diff
- name: Build all binaries
uses: actions-rs/cargo@v1
with:
-5
View File
@@ -45,11 +45,6 @@ jobs:
- name: Install wasm-bindgen-cli
run: cargo install wasm-bindgen-cli
- name: Set debug to false
run: |
sed -i.bak 's/\[profile.dev\]/\[profile.dev\]\ndebug = false/' Cargo.toml
git diff
- name: "Build"
run: make sdk-wasm-build
-206
View File
@@ -4,212 +4,6 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
## [Unreleased]
## [2025.4-dorina] (2025-03-04)
- fixed sphinx version metrics registration ([#5546])
- Feature/chain status api ([#5539])
- Add SURBs soft threshold ([#5535])
- Simplify IPR v8 ([#5532])
- Shared instance for DNS AsyncResolver ([#5523])
- merge #5512 again after reverting due to incorrect rebase ([#5520])
- cherry-pick 17d3ff2d775f61aee381d90a304ed416c08f33fc onto dorina ([#5519])
- cherry-pick 6e5d0dac1b75413c5f09122b0d953f8ec6ef48df onto dorina ([#5518])
- chore: workspace global panic preventing lints ([#5512])
- bugfix: dont query for ecash apis unless necessary when spending ticketbooks ([#5508])
- bugfix: bound check when recovering a reply SURB ([#5502])
- chore: removed all old coconut code ([#5500])
- IPR request types v8 ([#5498])
- Support static routes for HTTP requests ([#5487])
- build(deps): bump the patch-updates group across 1 directory with 3 updates ([#5482])
- added missing import to doctest ([#5480])
- adjusted TestSetup::new_complex to ensure bonded node's existence ([#5478])
- Trigger contracts CI on main workspace Cargo changes ([#5477])
- build(deps): bump http from 1.1.0 to 1.2.0 ([#5472])
- build(deps): bump utoipa-swagger-ui from 8.0.3 to 8.1.0 ([#5471])
- build(deps): bump colored from 2.1.0 to 2.2.0 ([#5470])
- build(deps): bump celes from 2.4.0 to 2.5.0 ([#5469])
- build(deps): bump the patch-updates group with 2 updates ([#5467])
- build(deps): bump elliptic from 6.5.4 to 6.6.1 in /docker/typescript_client/upload_contract ([#5463])
- Run cargo autoinherit ([#5460])
- Fix clippy::precedence ([#5457])
- Provide Interval context with node descriptor endpoints ([#5456])
- fix: update fx average rate calcs to ignore 0 values ([#5454])
- Feature/add gbp currency ([#5453])
- Add helper to extract a list of sqlite files with journal files wal/shm ([#5452])
- Add a middleware layer to the nym api allowing for data compression ([#5451])
- Condense core API functionalities and enable gzip decompression for reqwest payloads ([#5450])
- build(deps): bump uniffi_build from 0.25.3 to 0.29.0 ([#5448])
- Upgrade tower to 0.5.2 ([#5446])
- build(deps): bump hickory-proto from 0.24.2 to 0.24.3 ([#5444])
- Seedable clients ([#5440])
- build(deps): bump the patch-updates group across 1 directory with 10 updates ([#5439])
- Remove all recv_with_delay and add shutdown condition to loops in client-core ([#5435])
- Disable the test for checking the remaining bandwidth in nym-node-status-api ([#5425])
- Dz nym node stats ([#5418])
- build(deps): bump hyper from 1.4.1 to 1.6.0 ([#5416])
- build(deps): bump publicsuffix from 2.2.3 to 2.3.0 ([#5367])
- Nymnode entrypoint docker ([#5300])
[#5546]: https://github.com/nymtech/nym/pull/5546
[#5539]: https://github.com/nymtech/nym/pull/5539
[#5535]: https://github.com/nymtech/nym/pull/5535
[#5532]: https://github.com/nymtech/nym/pull/5532
[#5523]: https://github.com/nymtech/nym/pull/5523
[#5520]: https://github.com/nymtech/nym/pull/5520
[#5519]: https://github.com/nymtech/nym/pull/5519
[#5518]: https://github.com/nymtech/nym/pull/5518
[#5512]: https://github.com/nymtech/nym/pull/5512
[#5508]: https://github.com/nymtech/nym/pull/5508
[#5502]: https://github.com/nymtech/nym/pull/5502
[#5500]: https://github.com/nymtech/nym/pull/5500
[#5498]: https://github.com/nymtech/nym/pull/5498
[#5487]: https://github.com/nymtech/nym/pull/5487
[#5482]: https://github.com/nymtech/nym/pull/5482
[#5480]: https://github.com/nymtech/nym/pull/5480
[#5478]: https://github.com/nymtech/nym/pull/5478
[#5477]: https://github.com/nymtech/nym/pull/5477
[#5472]: https://github.com/nymtech/nym/pull/5472
[#5471]: https://github.com/nymtech/nym/pull/5471
[#5470]: https://github.com/nymtech/nym/pull/5470
[#5469]: https://github.com/nymtech/nym/pull/5469
[#5467]: https://github.com/nymtech/nym/pull/5467
[#5463]: https://github.com/nymtech/nym/pull/5463
[#5460]: https://github.com/nymtech/nym/pull/5460
[#5457]: https://github.com/nymtech/nym/pull/5457
[#5456]: https://github.com/nymtech/nym/pull/5456
[#5454]: https://github.com/nymtech/nym/pull/5454
[#5453]: https://github.com/nymtech/nym/pull/5453
[#5452]: https://github.com/nymtech/nym/pull/5452
[#5451]: https://github.com/nymtech/nym/pull/5451
[#5450]: https://github.com/nymtech/nym/pull/5450
[#5448]: https://github.com/nymtech/nym/pull/5448
[#5446]: https://github.com/nymtech/nym/pull/5446
[#5444]: https://github.com/nymtech/nym/pull/5444
[#5440]: https://github.com/nymtech/nym/pull/5440
[#5439]: https://github.com/nymtech/nym/pull/5439
[#5435]: https://github.com/nymtech/nym/pull/5435
[#5425]: https://github.com/nymtech/nym/pull/5425
[#5418]: https://github.com/nymtech/nym/pull/5418
[#5416]: https://github.com/nymtech/nym/pull/5416
[#5367]: https://github.com/nymtech/nym/pull/5367
[#5300]: https://github.com/nymtech/nym/pull/5300
## [2025.3-ruta] (2025-02-10)
- Push down forget me to client configs ([#5431])
- Fix statistics shutdown ([#5426])
- Make wait_for_graceful_shutdown to be pub ([#5424])
- Upgrade to thiserror 2.0 ([#5414])
- build(deps): bump the patch-updates group across 1 directory with 9 updates ([#5406])
- Relocate a validator api function ([#5401])
- Send shutdown instead of panic when reaching max fail ([#5398])
- Change Explorer URL to new smooshed nodes ([#5396])
- reduce log severity for checking topology validity ([#5395])
- MixnetClient can send ClientRequests ([#5381])
- Fix missing path triggers for CI ([#5380])
- Uncouple storage reference for bandwidth client ([#5372])
- build(deps): bump tokio from 1.40.0 to 1.43.0 ([#5370])
- DNS resolver configuration for internal HTTP client lookups ([#5355])
- Update README.md ([#5328])
- Update README.md ([#5327])
[#5431]: https://github.com/nymtech/nym/pull/5431
[#5426]: https://github.com/nymtech/nym/pull/5426
[#5424]: https://github.com/nymtech/nym/pull/5424
[#5414]: https://github.com/nymtech/nym/pull/5414
[#5406]: https://github.com/nymtech/nym/pull/5406
[#5401]: https://github.com/nymtech/nym/pull/5401
[#5398]: https://github.com/nymtech/nym/pull/5398
[#5396]: https://github.com/nymtech/nym/pull/5396
[#5395]: https://github.com/nymtech/nym/pull/5395
[#5381]: https://github.com/nymtech/nym/pull/5381
[#5380]: https://github.com/nymtech/nym/pull/5380
[#5372]: https://github.com/nymtech/nym/pull/5372
[#5370]: https://github.com/nymtech/nym/pull/5370
[#5355]: https://github.com/nymtech/nym/pull/5355
[#5328]: https://github.com/nymtech/nym/pull/5328
[#5327]: https://github.com/nymtech/nym/pull/5327
## [2025.2-hu] (2025-02-04)
- Feature/remove double spending bloomfilter ([#5417])
- HU - Downgrade harmless log message from info to debug ([#5405])
- lower default ticket verification quorum to 0.7 ([#5404])
- Downgrade harmless log message from info to debug ([#5403])
- Redirect from mixnode page to nodes page ([#5397])
- chore :update version of chain watcher and validator rewarder ([#5394])
- bugfix: correctly handle ingore epoch roles flag ([#5390])
- bugfix: terminate mixnet socket listener on shutdown ([#5389])
- feat: make client ignore dual mode nodes by default ([#5388])
- Handle ecash network errors differently ([#5378])
- Remove empty ephemeral keys ([#5376])
- fixed sql migration for adding default message timestamp ([#5374])
- Bind to [::] on nym-node for both IP versions ([#5361])
- exposed NymApiClient method for obtaining node performance history ([#5360])
- Client gateway selection ([#5358])
- chore: refresh wasm sdk ([#5353])
- chore: update indexed_db_futures ([#5347])
- build(deps): bump mikefarah/yq from 4.44.6 to 4.45.1 ([#5342])
- updated cosmrs and tendermint-rpc to their most recent versions ([#5339])
- build(deps): bump ts-rs from 10.0.0 to 10.1.0 ([#5338])
- build(deps): bump tempfile from 3.14.0 to 3.15.0 ([#5337])
- build(deps): bump the patch-updates group with 8 updates ([#5336])
- feature: introduce /load endpoint for self-reported quantised NymNode load ([#5326])
- feature: `CancellationToken`-based shutdowns ([#5325])
- Use expect in geodata test to give error message on failure ([#5314])
- feature: periodically remove stale gateway messages ([#5312])
- build(deps): bump the patch-updates group across 1 directory with 35 updates ([#5310])
- Add dependabot assignes for the root cargo ecosystem ([#5297])
- Move tun constants to network defaults ([#5286])
- Include IPINFO_API_TOKEN in nightly CI ([#5285])
- Nyx Chain Watcher ([#5274])
- bugfix: remove unnecessary arguments for nym-api swagger endpoints ([#5272])
- feature: nym topology revamp ([#5271])
- Add windows to CI builds ([#5269])
- http-api-client: deduplicate code ([#5267])
- build(deps): bump http from 1.1.0 to 1.2.0 ([#5228])
- NS API: add mixnet scraper ([#5200])
- build(deps): bump criterion from 0.4.0 to 0.5.1 ([#4911])
[#5417]: https://github.com/nymtech/nym/pull/5417
[#5405]: https://github.com/nymtech/nym/pull/5405
[#5404]: https://github.com/nymtech/nym/pull/5404
[#5403]: https://github.com/nymtech/nym/pull/5403
[#5397]: https://github.com/nymtech/nym/pull/5397
[#5394]: https://github.com/nymtech/nym/pull/5394
[#5390]: https://github.com/nymtech/nym/pull/5390
[#5389]: https://github.com/nymtech/nym/pull/5389
[#5388]: https://github.com/nymtech/nym/pull/5388
[#5378]: https://github.com/nymtech/nym/pull/5378
[#5376]: https://github.com/nymtech/nym/pull/5376
[#5374]: https://github.com/nymtech/nym/pull/5374
[#5361]: https://github.com/nymtech/nym/pull/5361
[#5360]: https://github.com/nymtech/nym/pull/5360
[#5358]: https://github.com/nymtech/nym/pull/5358
[#5353]: https://github.com/nymtech/nym/pull/5353
[#5347]: https://github.com/nymtech/nym/pull/5347
[#5342]: https://github.com/nymtech/nym/pull/5342
[#5339]: https://github.com/nymtech/nym/pull/5339
[#5338]: https://github.com/nymtech/nym/pull/5338
[#5337]: https://github.com/nymtech/nym/pull/5337
[#5336]: https://github.com/nymtech/nym/pull/5336
[#5326]: https://github.com/nymtech/nym/pull/5326
[#5325]: https://github.com/nymtech/nym/pull/5325
[#5314]: https://github.com/nymtech/nym/pull/5314
[#5312]: https://github.com/nymtech/nym/pull/5312
[#5310]: https://github.com/nymtech/nym/pull/5310
[#5297]: https://github.com/nymtech/nym/pull/5297
[#5286]: https://github.com/nymtech/nym/pull/5286
[#5285]: https://github.com/nymtech/nym/pull/5285
[#5274]: https://github.com/nymtech/nym/pull/5274
[#5272]: https://github.com/nymtech/nym/pull/5272
[#5271]: https://github.com/nymtech/nym/pull/5271
[#5269]: https://github.com/nymtech/nym/pull/5269
[#5267]: https://github.com/nymtech/nym/pull/5267
[#5228]: https://github.com/nymtech/nym/pull/5228
[#5200]: https://github.com/nymtech/nym/pull/5200
[#4911]: https://github.com/nymtech/nym/pull/4911
## [2025.1-reeses] (2025-01-15)
- Feture/legacy alert ([#5346])
Generated
+1133 -1290
View File
File diff suppressed because it is too large Load Diff
+39 -54
View File
@@ -48,6 +48,7 @@ members = [
"common/credentials-interface",
"common/crypto",
"common/dkg",
"common/ecash-double-spending",
"common/ecash-time",
"common/execute",
"common/exit-policy",
@@ -66,6 +67,7 @@ members = [
"common/nym-id",
"common/nym-metrics",
"common/nym_offline_compact_ecash",
"common/nymcoconut",
"common/nymsphinx",
"common/nymsphinx/acknowledgements",
"common/nymsphinx/addressing",
@@ -186,43 +188,40 @@ readme = "README.md"
[workspace.dependencies]
addr = "0.15.6"
aead = "0.5.2"
aes = "0.8.1"
aes-gcm = "0.10.1"
aes-gcm-siv = "0.11.1"
ammonia = "4"
anyhow = "1.0.97"
aead = "0.5.2"
anyhow = "1.0.95"
arc-swap = "1.7.1"
argon2 = "0.5.0"
async-trait = "0.1.87"
axum = "0.7.5"
async-trait = "0.1.85"
axum-client-ip = "0.6.1"
axum = "0.7.5"
axum-extra = "0.9.4"
axum-test = "16.2.0"
base64 = "0.22.1"
base85rs = "0.1.3"
bincode = "1.3.3"
bip39 = { version = "2.0.0", features = ["zeroize"] }
bit-vec = "0.7.0" # can we unify those?
bitvec = "1.0.0"
blake3 = "1.6.1"
blake3 = "1.5.5"
bloomfilter = "1.0.14"
bs58 = "0.5.1"
bytecodec = "0.4.15"
bytes = "1.7.2"
cargo_metadata = "0.18.1"
celes = "2.5.0"
celes = "2.4.0"
cfg-if = "1.0.0"
chacha20 = "0.9.0"
chacha20poly1305 = "0.10.1"
chrono = "0.4.40"
chrono = "0.4.39"
cipher = "0.4.3"
clap = "4.5.31"
clap = "4.5.26"
clap_complete = "4.5"
clap_complete_fig = "4.5"
colored = "2.2"
comfy-table = "7.1.4"
console = "0.15.11"
colored = "2.0"
comfy-table = "7.1.3"
console = "0.15.10"
console-subscriber = "0.1.1"
console_error_panic_hook = "0.1"
const-str = "0.5.6"
@@ -241,50 +240,50 @@ doc-comment = "0.3"
dotenvy = "0.15.6"
ecdsa = "0.16"
ed25519-dalek = "2.1"
env_logger = "0.11.6"
envy = "0.4"
etherparse = "0.13.0"
envy = "0.4"
eyre = "0.6.9"
fastrand = "2.1.1"
flate2 = "1.1.0"
flate2 = "1.0.35"
futures = "0.3.31"
futures-util = "0.3"
generic-array = "0.14.7"
getrandom = "0.2.10"
getset = "0.1.5"
getset = "0.1.3"
handlebars = "3.5.5"
headers = "0.4.0"
hex = "0.4.3"
hex-literal = "0.3.3"
hickory-resolver = "0.24.4"
hickory-resolver = "0.24.2"
hkdf = "0.12.3"
hmac = "0.12.1"
http = "1"
http-body-util = "0.1"
httpcodec = "0.2.3"
human-repr = "1.1.0"
humantime = "2.1.0"
humantime-serde = "1.1.1"
hyper = "1.6.0"
human-repr = "1.1.0"
hyper = "1.4.1"
hyper-util = "0.1"
indicatif = "0.17.11"
indicatif = "0.17.9"
inquire = "0.6.2"
ip_network = "0.4.1"
ipnetwork = "0.20"
isocountry = "0.3.2"
itertools = "0.14.0"
itertools = "0.13.0"
k256 = "0.13"
lazy_static = "1.5.0"
ledger-transport = "0.10.0"
ledger-transport-hid = "0.10.0"
log = "0.4"
maxminddb = "0.23.0"
rs_merkle = "1.4.2"
mime = "0.3.17"
moka = { version = "0.12", features = ["future"] }
nix = "0.27.1"
notify = "5.1.0"
okapi = "0.7.0"
once_cell = "1.20.3"
once_cell = "1.20.2"
opentelemetry = "0.19.0"
opentelemetry-jaeger = "0.18.0"
parking_lot = "0.12.3"
@@ -293,7 +292,7 @@ petgraph = "0.6.5"
pin-project = "1.1"
pin-project-lite = "0.2.16"
pretty_env_logger = "0.4.0"
publicsuffix = "2.3.0"
publicsuffix = "2.2.3"
quote = "1"
rand = "0.8.5"
rand_chacha = "0.3"
@@ -307,21 +306,20 @@ reqwest = { version = "0.12.4", default-features = false }
rocket = "0.5.0"
rocket_cors = "0.6.0"
rocket_okapi = "0.8.0"
rs_merkle = "1.4.2"
safer-ffi = "0.1.13"
schemars = "0.8.22"
semver = "1.0.25"
schemars = "0.8.21"
semver = "1.0.24"
serde = "1.0.217"
serde_bytes = "0.11.16"
serde_bytes = "0.11.15"
serde_derive = "1.0"
serde_json = "1.0.140"
serde_json_path = "0.7.2"
serde_json = "1.0.135"
serde_json_path = "0.7.1"
serde_repr = "0.1"
serde_with = "3.9.0"
serde_yaml = "0.9.25"
sha2 = "0.10.8"
si-scale = "0.2.3"
sphinx-packet = "=0.4.0" # make sure to use version 0.4.0 (or higher) that has removed backwards compatibility
sphinx-packet = "0.1.1"
sqlx = "0.7.4"
strum = "0.26"
strum_macros = "0.26"
@@ -329,32 +327,29 @@ subtle-encoding = "0.5"
syn = "1"
sysinfo = "0.33.0"
tap = "1.0.1"
tar = "0.4.44"
tar = "0.4.43"
tempfile = "3.15"
thiserror = "2.0"
thiserror = "1.0.64"
time = "0.3.37"
tokio = "1.43"
tokio-postgres = "0.7"
tokio = "1.39"
tokio-stream = "0.1.17"
tokio-test = "0.4.4"
tokio-tun = "0.11.5"
tokio-tungstenite = { version = "0.20.1" }
tokio-util = "0.7.13"
toml = "0.8.20"
tower = "0.5.2"
toml = "0.8.19"
tower = "0.4.13"
tower-http = "0.5.2"
tracing = "0.1.41"
tracing-log = "0.2"
tracing-opentelemetry = "0.19.0"
tracing-subscriber = "0.3.19"
tracing-tree = "0.2.2"
tracing-log = "0.2"
ts-rs = "10.1.0"
tungstenite = { version = "0.20.1", default-features = false }
uniffi = "0.29.0"
uniffi_build = "0.29.0"
url = "2.5"
utoipa = "5.2"
utoipa-swagger-ui = "8.1"
utoipa-swagger-ui = "8.0"
utoipauto = "0.2"
uuid = "*"
vergen = { version = "=8.3.1", default-features = false }
@@ -390,10 +385,10 @@ cw4 = { version = "=1.1.2" }
cw-controllers = { version = "=1.1.0" }
# cosmrs-related
bip32 = { version = "0.5.3", default-features = false }
bip32 = { version = "0.5.2", default-features = false }
cosmrs = { version = "0.21.1" }
cosmrs = { version = "0.21.0" }
tendermint = "0.40.0"
tendermint-rpc = "0.40.0"
prost = { version = "0.13", default-features = false }
@@ -437,13 +432,3 @@ opt-level = 'z'
[profile.release.package.mix-fetch-wasm]
# lto = true
opt-level = 'z'
[workspace.lints.clippy]
unwrap_used = "deny"
expect_used = "deny"
todo = "deny"
dbg_macro = "deny"
exit = "deny"
panic = "deny"
unimplemented = "deny"
unreachable = "deny"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-client"
version = "1.1.49"
version = "1.1.46"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
description = "Implementation of the Nym Client"
edition = "2021"
+2 -6
View File
@@ -56,7 +56,7 @@ pub fn default_data_directory<P: AsRef<Path>>(id: P) -> PathBuf {
.join(DEFAULT_DATA_DIR)
}
#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)]
#[derive(Debug, Deserialize, PartialEq, Serialize)]
pub struct Config {
#[serde(flatten)]
pub base: BaseClientConfig,
@@ -94,10 +94,6 @@ impl CliClientConfig for Config {
}
impl Config {
pub fn base(&self) -> BaseClientConfig {
self.base.clone()
}
pub fn new<S: AsRef<str>>(id: S) -> Self {
Config {
base: BaseClientConfig::new(id.as_ref(), env!("CARGO_PKG_VERSION")),
@@ -213,7 +209,7 @@ impl SocketType {
}
}
#[derive(Debug, Deserialize, PartialEq, Eq, Serialize, Clone)]
#[derive(Debug, Deserialize, PartialEq, Eq, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct Socket {
pub socket_type: SocketType,
@@ -107,8 +107,5 @@ enabled = {{ debug.stats_reporting.enabled }}
provider_address = '{{ debug.stats_reporting.provider_address }}'
reporting_interval = '{{ debug.stats_reporting.reporting_interval }}'
[debug.forget_me]
client = {{ debug.forget_me.client }}
stats = {{ debug.forget_me.stats }}
"#;
+6 -16
View File
@@ -20,7 +20,7 @@ pub use nym_sphinx::addressing::clients::Recipient;
pub mod config;
type NativeClientBuilder = BaseClientBuilder<QueryHttpRpcNyxdClient, OnDiskPersistent>;
type NativeClientBuilder<'a> = BaseClientBuilder<'a, QueryHttpRpcNyxdClient, OnDiskPersistent>;
pub struct SocketClient {
/// Client configuration options, including, among other things, packet sending rates,
@@ -32,10 +32,6 @@ pub struct SocketClient {
}
impl SocketClient {
pub fn config(&self) -> Config {
self.config.clone()
}
pub fn new(config: Config, custom_mixnet: Option<PathBuf>) -> Self {
SocketClient {
config,
@@ -49,7 +45,7 @@ impl SocketClient {
client_output: ClientOutput,
client_state: ClientState,
self_address: &Recipient,
task_client: nym_task::TaskClient,
shutdown: nym_task::TaskClient,
packet_type: PacketType,
) {
info!("Starting websocket listener...");
@@ -77,15 +73,10 @@ impl SocketClient {
shared_lane_queue_lengths,
reply_controller_sender,
Some(packet_type),
task_client.fork("websocket_handler"),
);
websocket::Listener::new(
config.socket.host,
config.socket.listening_port,
task_client.with_suffix("websocket_listener"),
)
.start(websocket_handler);
websocket::Listener::new(config.socket.host, config.socket.listening_port)
.start(websocket_handler, shutdown);
}
/// blocking version of `start_socket` method. Will run forever (or until SIGINT is sent)
@@ -117,9 +108,8 @@ impl SocketClient {
let storage = self.initialise_storage().await?;
let user_agent = nym_bin_common::bin_info!().into();
let mut base_client =
BaseClientBuilder::new(self.config().base(), storage, dkg_query_client)
.with_user_agent(user_agent);
let mut base_client = BaseClientBuilder::new(&self.config.base, storage, dkg_query_client)
.with_user_agent(user_agent);
if let Some(custom_mixnet) = &self.custom_mixnet {
base_client = base_client.with_stored_topology(custom_mixnet)?;
-1
View File
@@ -82,7 +82,6 @@ impl From<Init> for OverrideConfig {
nyxd_urls: init_config.common_args.nyxd_urls,
enabled_credentials_mode: init_config.common_args.enabled_credentials_mode,
stats_reporting_address: init_config.common_args.stats_reporting_address,
forget_me: init_config.common_args.forget_me.into(),
}
}
}
-3
View File
@@ -16,7 +16,6 @@ use nym_bin_common::completions::{fig_generate, ArgShell};
use nym_client::client::Recipient;
use nym_client_core::cli_helpers::CliClient;
use nym_client_core::client::base_client::storage::migration_helpers::v1_1_33;
use nym_client_core::config::ForgetMe;
use nym_config::OptionalSet;
use std::error::Error;
use std::net::IpAddr;
@@ -107,7 +106,6 @@ pub(crate) struct OverrideConfig {
nyxd_urls: Option<Vec<url::Url>>,
enabled_credentials_mode: Option<bool>,
stats_reporting_address: Option<Recipient>,
forget_me: ForgetMe,
}
pub(crate) async fn execute(args: Cli) -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -135,7 +133,6 @@ pub(crate) fn override_config(config: Config, args: OverrideConfig) -> Config {
args.fastmode,
)
.with_base(BaseClientConfig::with_disabled_cover_traffic, args.no_cover)
.with_base(BaseClientConfig::with_forget_me, args.forget_me)
.with_optional(Config::with_port, args.port)
.with_optional(Config::with_host, args.host)
.with_optional_custom_env_ext(
-1
View File
@@ -41,7 +41,6 @@ impl From<Run> for OverrideConfig {
nyxd_urls: run_config.common_args.nyxd_urls,
enabled_credentials_mode: run_config.common_args.enabled_credentials_mode,
stats_reporting_address: run_config.common_args.stats_reporting_address,
forget_me: run_config.common_args.forget_me.into(),
}
}
}
+40 -66
View File
@@ -19,7 +19,6 @@ use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::connections::{
ConnectionCommand, ConnectionCommandSender, ConnectionId, LaneQueueLengths, TransmissionLane,
};
use nym_task::TaskClient;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::time::Instant;
@@ -44,11 +43,9 @@ pub(crate) struct HandlerBuilder {
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
task_client: TaskClient,
}
impl HandlerBuilder {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
msg_input: InputMessageSender,
client_connection_tx: ConnectionCommandSender,
@@ -57,7 +54,6 @@ impl HandlerBuilder {
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
task_client: TaskClient,
) -> Self {
Self {
msg_input,
@@ -67,14 +63,11 @@ impl HandlerBuilder {
lane_queue_lengths,
reply_controller_sender,
packet_type,
task_client,
}
}
// TODO: make sure we only ever have one active handler
pub fn create_active_handler(&self) -> Handler {
let mut task_client = self.task_client.fork("active_handler");
task_client.disarm();
Handler {
msg_input: self.msg_input.clone(),
client_connection_tx: self.client_connection_tx.clone(),
@@ -85,7 +78,6 @@ impl HandlerBuilder {
lane_queue_lengths: self.lane_queue_lengths.clone(),
reply_controller_sender: self.reply_controller_sender.clone(),
packet_type: self.packet_type,
task_client,
}
}
}
@@ -100,18 +92,16 @@ pub(crate) struct Handler {
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
task_client: TaskClient,
}
impl Drop for Handler {
fn drop(&mut self) {
if let Err(err) = self
if self
.buffer_requester
.unbounded_send(ReceivedBufferMessage::ReceiverDisconnect)
.is_err()
{
if !self.task_client.is_shutdown_poll() {
error!("failed to disconnect the receiver from the buffer: {err}");
}
error!("we failed to disconnect the receiver from the buffer! presumably the shutdown procedure has been initiated!")
}
}
}
@@ -135,23 +125,10 @@ impl Handler {
};
// get the number of pending replies waiting for reply surbs
let reply_queue_length = match self
let reply_queue_length = self
.reply_controller_sender
.get_lane_queue_length(connection_id)
.await
{
Ok(length) => length,
Err(err) => {
if !self.task_client.is_shutdown_poll() {
error!(
"Failed to get reply queue length for connection {connection_id}: {err}"
);
}
// We're just going to assume that the queue is empty, and I think that's okay
// during shutdown.
0
}
};
.await;
let queue_length = base_length + reply_queue_length;
@@ -191,11 +168,10 @@ impl Handler {
// the ack control is now responsible for chunking, etc.
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send message to the input buffer: {err}");
}
}
self.msg_input
.send(input_msg)
.await
.expect("InputMessageReceiver has stopped receiving!");
// Only reply back with a `LaneQueueLength` if the sender providided a connection id
let TransmissionLane::ConnectionId(connection_id) = lane else {
@@ -224,11 +200,10 @@ impl Handler {
let input_msg =
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send anonymous message to the input buffer: {err}");
}
}
self.msg_input
.send(input_msg)
.await
.expect("InputMessageReceiver has stopped receiving!");
// Only reply back with a `LaneQueueLength` if the sender providided a connection id
let TransmissionLane::ConnectionId(connection_id) = lane else {
@@ -252,11 +227,10 @@ impl Handler {
});
let input_msg = InputMessage::new_reply(recipient_tag, message, lane, self.packet_type);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send reply message to the input buffer: {err}");
}
}
self.msg_input
.send(input_msg)
.await
.expect("InputMessageReceiver has stopped receiving!");
// Only reply back with a `LaneQueueLength` if the sender providided a connection id
let TransmissionLane::ConnectionId(connection_id) = lane else {
@@ -271,14 +245,9 @@ impl Handler {
}
fn handle_closed_connection(&self, connection_id: u64) -> Option<ServerResponse> {
if let Err(err) = self
.client_connection_tx
self.client_connection_tx
.unbounded_send(ConnectionCommand::Close(connection_id))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send close connection command: {err}");
}
}
.unwrap();
None
}
@@ -393,10 +362,11 @@ impl Handler {
}
}
async fn listen_for_requests(&mut self, mut msg_receiver: ReconstructedMessagesReceiver) {
let mut task_client = self.task_client.fork("select");
task_client.disarm();
async fn listen_for_requests(
&mut self,
mut msg_receiver: ReconstructedMessagesReceiver,
mut task_client: nym_task::TaskClient,
) {
while !task_client.is_shutdown() {
tokio::select! {
// we can either get a client request from the websocket
@@ -445,7 +415,15 @@ impl Handler {
}
// consume self to make sure `drop` is called after this is done
pub(crate) async fn handle_connection(mut self, socket: TcpStream) {
pub(crate) async fn handle_connection(
mut self,
socket: TcpStream,
mut task_client: nym_task::TaskClient,
) {
// We don't want a crash in the connection handler to trigger a shutdown of the whole
// process.
task_client.disarm();
let ws_stream = match accept_async(socket).await {
Ok(ws_stream) => ws_stream,
Err(err) => {
@@ -458,18 +436,14 @@ impl Handler {
let (reconstructed_sender, reconstructed_receiver) = mpsc::unbounded();
// tell the buffer to start sending stuff to us
if let Err(err) =
self.buffer_requester
.unbounded_send(ReceivedBufferMessage::ReceiverAnnounce(
reconstructed_sender,
))
{
if !self.task_client.is_shutdown_poll() {
error!("failed to announce the receiver to the buffer: {err}");
}
}
self.buffer_requester
.unbounded_send(ReceivedBufferMessage::ReceiverAnnounce(
reconstructed_sender,
))
.expect("the buffer request failed!");
self.listen_for_requests(reconstructed_receiver).await;
self.listen_for_requests(reconstructed_receiver, task_client)
.await;
}
}
+17 -11
View File
@@ -3,7 +3,6 @@
use super::handler::HandlerBuilder;
use log::*;
use nym_task::TaskClient;
use std::net::IpAddr;
use std::{net::SocketAddr, process, sync::Arc};
use tokio::io::AsyncWriteExt;
@@ -23,19 +22,21 @@ impl State {
pub(crate) struct Listener {
address: SocketAddr,
state: State,
task_client: TaskClient,
}
impl Listener {
pub(crate) fn new(host: IpAddr, port: u16, task_client: TaskClient) -> Self {
pub(crate) fn new(host: IpAddr, port: u16) -> Self {
Listener {
address: SocketAddr::new(host, port),
state: State::AwaitingConnection,
task_client,
}
}
pub(crate) async fn run(&mut self, handler: HandlerBuilder) {
pub(crate) async fn run(
&mut self,
handler: HandlerBuilder,
mut task_client: nym_task::TaskClient,
) {
let tcp_listener = match tokio::net::TcpListener::bind(self.address).await {
Ok(listener) => listener,
Err(err) => {
@@ -46,11 +47,11 @@ impl Listener {
let notify = Arc::new(Notify::new());
while !self.task_client.is_shutdown() {
loop {
tokio::select! {
// When the handler finishes we check if shutdown is signalled
_ = notify.notified() => {
if self.task_client.is_shutdown() {
if task_client.is_shutdown() {
log::trace!("Websocket listener: detected shutdown after connection closed");
break;
}
@@ -59,7 +60,7 @@ impl Listener {
}
// ... but when there is no connected client at the time of shutdown being
// signalled, we handle it here.
_ = self.task_client.recv() => {
_ = task_client.recv() => {
if !self.state.is_connected() {
log::trace!("Not connected: shutting down");
break;
@@ -87,8 +88,9 @@ impl Listener {
// hanging because the executor doesn't come back here
let notify_clone = Arc::clone(&notify);
let fresh_handler = handler.create_active_handler();
let task_client_handler = task_client.clone();
tokio::spawn(async move {
fresh_handler.handle_connection(socket).await;
fresh_handler.handle_connection(socket, task_client_handler).await;
notify_clone.notify_one();
});
self.state = State::Connected;
@@ -102,9 +104,13 @@ impl Listener {
log::debug!("Websocket listener: Exiting");
}
pub(crate) fn start(mut self, handler: HandlerBuilder) -> JoinHandle<()> {
pub(crate) fn start(
mut self,
handler: HandlerBuilder,
shutdown: nym_task::TaskClient,
) -> JoinHandle<()> {
info!("Running websocket on {:?}", self.address.to_string());
tokio::spawn(async move { self.run(handler).await })
tokio::spawn(async move { self.run(handler, shutdown).await })
}
}
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-socks5-client"
version = "1.1.49"
version = "1.1.46"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
description = "A SOCKS5 localhost proxy that converts incoming messages to Sphinx and sends them to a Nym address"
edition = "2021"
-1
View File
@@ -93,7 +93,6 @@ impl From<Init> for OverrideConfig {
enabled_credentials_mode: init_config.common_args.enabled_credentials_mode,
outfox: false,
stats_reporting_address: init_config.common_args.stats_reporting_address,
forget_me: init_config.common_args.forget_me.into(),
}
}
}
+1 -3
View File
@@ -17,7 +17,7 @@ use nym_bin_common::completions::{fig_generate, ArgShell};
use nym_client_core::cli_helpers::CliClient;
use nym_client_core::client::base_client::storage::migration_helpers::v1_1_33;
use nym_client_core::client::topology_control::geo_aware_provider::CountryGroup;
use nym_client_core::config::{ForgetMe, GroupBy, TopologyStructure};
use nym_client_core::config::{GroupBy, TopologyStructure};
use nym_config::OptionalSet;
use nym_sphinx::addressing::Recipient;
use nym_sphinx::params::{PacketSize, PacketType};
@@ -113,7 +113,6 @@ pub(crate) struct OverrideConfig {
enabled_credentials_mode: Option<bool>,
outfox: bool,
stats_reporting_address: Option<Recipient>,
forget_me: ForgetMe,
}
pub(crate) async fn execute(args: Cli) -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -180,7 +179,6 @@ pub(crate) fn override_config(config: Config, args: OverrideConfig) -> Config {
BaseClientConfig::with_topology_structure,
topology_structure,
)
.with_base(BaseClientConfig::with_forget_me, args.forget_me)
.with_optional(Config::with_anonymous_replies, args.use_anonymous_replies)
.with_optional(Config::with_port, args.port)
.with_optional(Config::with_ip, args.ip)
-1
View File
@@ -65,7 +65,6 @@ impl From<Run> for OverrideConfig {
enabled_credentials_mode: run_config.common_args.enabled_credentials_mode,
outfox: run_config.outfox,
stats_reporting_address: run_config.common_args.stats_reporting_address,
forget_me: run_config.common_args.forget_me.into(),
}
}
}
-4
View File
@@ -113,8 +113,4 @@ enabled = {{ core.debug.stats_reporting.enabled }}
provider_address = '{{ core.debug.stats_reporting.provider_address }}'
reporting_interval = '{{ core.debug.stats_reporting.reporting_interval }}'
[core.debug.forget_me]
client = {{ core.debug.forget_me.client }}
stats = {{ core.debug.forget_me.stats }}
"#;
-1
View File
@@ -1,3 +1,2 @@
allow-unwrap-in-tests = true
allow-expect-in-tests = true
allow-panic-in-tests = true
@@ -60,7 +60,7 @@ impl From<IpAddr> for IpPair {
std::net::IpAddr::V4(ipv4_addr) => (ipv4_addr.octets()[2], ipv4_addr.octets()[3]),
std::net::IpAddr::V6(ipv6_addr) => (ipv6_addr.octets()[14], ipv6_addr.octets()[15]),
};
let last_bytes = ((before_last_byte as u16) << 8) | last_byte as u16;
let last_bytes = (before_last_byte as u16) << 8 | last_byte as u16;
let ipv4 = Ipv4Addr::new(
WG_TUN_DEVICE_IP_ADDRESS_V4.octets()[0],
WG_TUN_DEVICE_IP_ADDRESS_V4.octets()[1],
+7 -4
View File
@@ -105,24 +105,26 @@ impl<C, St: Storage> BandwidthController<C, St> {
async fn get_aggregate_verification_key(
&self,
epoch_id: EpochId,
ecash_apis: &mut ApiClientsWrapper<'_, C>,
apis: &mut ApiClientsWrapper,
) -> Result<VerificationKeyAuth, BandwidthControllerError>
where
C: DkgQueryClient + Sync + Send,
<St as Storage>::StorageError: Send + Sync + 'static,
{
let ecash_apis = apis.get_or_init(epoch_id, &self.client).await?;
get_aggregate_verification_key(&self.storage, epoch_id, ecash_apis).await
}
async fn get_coin_index_signatures(
&self,
epoch_id: EpochId,
ecash_apis: &mut ApiClientsWrapper<'_, C>,
apis: &mut ApiClientsWrapper,
) -> Result<Vec<AnnotatedCoinIndexSignature>, BandwidthControllerError>
where
C: DkgQueryClient + Sync + Send,
<St as Storage>::StorageError: Send + Sync + 'static,
{
let ecash_apis = apis.get_or_init(epoch_id, &self.client).await?;
get_coin_index_signatures(&self.storage, epoch_id, ecash_apis).await
}
@@ -130,12 +132,13 @@ impl<C, St: Storage> BandwidthController<C, St> {
&self,
epoch_id: EpochId,
expiration_date: Date,
ecash_apis: &mut ApiClientsWrapper<'_, C>,
apis: &mut ApiClientsWrapper,
) -> Result<Vec<AnnotatedExpirationDateSignature>, BandwidthControllerError>
where
C: DkgQueryClient + Sync + Send,
<St as Storage>::StorageError: Send + Sync + 'static,
{
let ecash_apis = apis.get_or_init(epoch_id, &self.client).await?;
get_expiration_date_signatures(&self.storage, epoch_id, expiration_date, ecash_apis).await
}
@@ -151,7 +154,7 @@ impl<C, St: Storage> BandwidthController<C, St> {
{
let epoch_id = retrieved_ticketbook.ticketbook.epoch_id();
let expiration_date = retrieved_ticketbook.ticketbook.expiration_date();
let mut api_clients = ApiClientsWrapper::new(&self.client, epoch_id);
let mut api_clients = Default::default();
let verification_key = self
.get_aggregate_verification_key(epoch_id, &mut api_clients)
+20 -63
View File
@@ -21,67 +21,30 @@ use rand::thread_rng;
use std::fmt::Display;
use std::future::Future;
pub(crate) trait EcashClientsProvider {
async fn try_get_ecash_clients(
&mut self,
) -> Result<Vec<EcashApiClient>, BandwidthControllerError>;
}
// it really doesn't need the RwLock because it's never moved across tasks,
// but we need all the Send/Sync action
#[derive(Default)]
pub(crate) struct ApiClientsWrapper(Option<Vec<EcashApiClient>>);
impl EcashClientsProvider for Vec<EcashApiClient> {
async fn try_get_ecash_clients(
impl ApiClientsWrapper {
pub(crate) async fn get_or_init<C>(
&mut self,
) -> Result<Vec<EcashApiClient>, BandwidthControllerError> {
Ok(self.clone())
}
}
impl<C> EcashClientsProvider for &mut ApiClientsWrapper<'_, C>
where
C: DkgQueryClient + Sync + Send,
{
async fn try_get_ecash_clients(
&mut self,
) -> Result<Vec<EcashApiClient>, BandwidthControllerError> {
self.clients().await
}
}
pub(crate) enum ApiClientsWrapper<'a, C> {
Uninitialised {
query_client: &'a C,
epoch_id: EpochId,
},
Cached {
clients: Vec<EcashApiClient>,
},
}
impl<'a, C> ApiClientsWrapper<'a, C> {
pub(crate) fn new(query_client: &'a C, epoch_id: EpochId) -> Self {
ApiClientsWrapper::Uninitialised {
query_client,
epoch_id,
}
}
async fn clients(&mut self) -> Result<Vec<EcashApiClient>, BandwidthControllerError>
dkg_client: &C,
) -> Result<Vec<EcashApiClient>, BandwidthControllerError>
where
C: DkgQueryClient + Sync + Send,
{
match self {
ApiClientsWrapper::Uninitialised {
query_client,
epoch_id,
} => {
let clients = all_ecash_api_clients(*query_client, *epoch_id).await?;
*self = ApiClientsWrapper::Cached {
clients: clients.clone(),
};
Ok(clients)
}
ApiClientsWrapper::Cached { clients } => Ok(clients.clone()),
if let Some(cached) = &self.0 {
return Ok(cached.clone());
}
let clients = all_ecash_api_clients(dkg_client, epoch_id).await?;
// technically we don't have to be cloning all the clients here, but it's way simpler than
// dealing with locking and whatnot given the performance penalty is negligible
self.0 = Some(clients.clone());
Ok(clients)
}
}
@@ -113,7 +76,7 @@ where
pub(crate) async fn get_aggregate_verification_key<St>(
storage: &St,
epoch_id: EpochId,
mut ecash_apis: impl EcashClientsProvider,
ecash_apis: Vec<EcashApiClient>,
) -> Result<VerificationKeyAuth, BandwidthControllerError>
where
St: Storage,
@@ -127,8 +90,6 @@ where
return Ok(stored);
};
let ecash_apis = ecash_apis.try_get_ecash_clients().await?;
let master_vk = query_random_apis_until_success(
ecash_apis,
|api| async move { api.api_client.master_verification_key(Some(epoch_id)).await },
@@ -154,7 +115,7 @@ where
pub(crate) async fn get_coin_index_signatures<St>(
storage: &St,
epoch_id: EpochId,
mut ecash_apis: impl EcashClientsProvider,
ecash_apis: Vec<EcashApiClient>,
) -> Result<Vec<AnnotatedCoinIndexSignature>, BandwidthControllerError>
where
St: Storage,
@@ -168,8 +129,6 @@ where
return Ok(stored);
};
let ecash_apis = ecash_apis.try_get_ecash_clients().await?;
let index_sigs = query_random_apis_until_success(
ecash_apis,
|api| async move {
@@ -200,7 +159,7 @@ pub(crate) async fn get_expiration_date_signatures<St>(
storage: &St,
epoch_id: EpochId,
expiration_date: Date,
mut ecash_apis: impl EcashClientsProvider,
ecash_apis: Vec<EcashApiClient>,
) -> Result<Vec<AnnotatedExpirationDateSignature>, BandwidthControllerError>
where
St: Storage,
@@ -214,8 +173,6 @@ where
return Ok(stored);
};
let ecash_apis = ecash_apis.try_get_ecash_clients().await?;
let expiration_sigs = query_random_apis_until_success(
ecash_apis,
|api| async move {
-1
View File
@@ -40,7 +40,6 @@ nym-crypto = { path = "../crypto" }
nym-explorer-client = { path = "../../explorer-api/explorer-client" }
nym-gateway-client = { path = "../client-libs/gateway-client" }
nym-gateway-requests = { path = "../gateway-requests" }
nym-http-api-client = { path = "../http-api-client" }
nym-metrics = { path = "../nym-metrics" }
nym-nonexhaustive-delayqueue = { path = "../nonexhaustive-delayqueue" }
nym-sphinx = { path = "../nymsphinx" }
+2 -85
View File
@@ -45,7 +45,6 @@ const DEFAULT_COVER_TRAFFIC_PRIMARY_SIZE_RATIO: f64 = 0.70;
// clients/client-core/src/client/replies/reply_storage/surb_storage.rs
const DEFAULT_MINIMUM_REPLY_SURB_STORAGE_THRESHOLD: usize = 10;
const DEFAULT_MAXIMUM_REPLY_SURB_STORAGE_THRESHOLD: usize = 200;
const DEFAULT_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER: usize = 0;
// define how much to request at once
// clients/client-core/src/client/replies/reply_controller.rs
@@ -146,11 +145,6 @@ impl Config {
self
}
pub fn with_forget_me(mut self, forget_me: ForgetMe) -> Self {
self.debug.forget_me = forget_me;
self
}
// TODO: this should be refactored properly
// as of 12.09.23 the below is true (not sure how this comment will rot in the future)
// medium_toggle:
@@ -523,7 +517,7 @@ impl Default for Acknowledgements {
}
#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Serialize)]
#[serde(default)]
#[serde(default, deny_unknown_fields)]
pub struct Topology {
/// The uniform delay every which clients are querying the directory server
/// to try to obtain a compatible network topology to send sphinx packets through.
@@ -564,10 +558,6 @@ pub struct Topology {
/// Specifies whether this client should ignore the current epoch role of the target egress node
/// when constructing the final hop packets.
pub ignore_egress_epoch_role: bool,
/// Specifies whether this client should ignore the current epoch role of the ingress node
/// when attempting to establish new connection
pub ignore_ingress_epoch_role: bool,
}
#[allow(clippy::large_enum_variant)]
@@ -605,9 +595,7 @@ impl Default for Topology {
minimum_mixnode_performance: DEFAULT_MIN_MIXNODE_PERFORMANCE,
minimum_gateway_performance: DEFAULT_MIN_GATEWAY_PERFORMANCE,
use_extended_topology: false,
ignore_egress_epoch_role: true,
ignore_ingress_epoch_role: true,
ignore_egress_epoch_role: false,
}
}
}
@@ -622,10 +610,6 @@ pub struct ReplySurbs {
/// Defines the maximum number of reply surbs the client wants to keep in its storage at any times.
pub maximum_reply_surb_storage_threshold: usize,
/// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client
/// should proactively request additional reply surbs.
pub minimum_reply_surb_threshold_buffer: usize,
/// Defines the minimum number of reply surbs the client would request.
pub minimum_reply_surb_request_size: u32,
@@ -665,7 +649,6 @@ impl Default for ReplySurbs {
ReplySurbs {
minimum_reply_surb_storage_threshold: DEFAULT_MINIMUM_REPLY_SURB_STORAGE_THRESHOLD,
maximum_reply_surb_storage_threshold: DEFAULT_MAXIMUM_REPLY_SURB_STORAGE_THRESHOLD,
minimum_reply_surb_threshold_buffer: DEFAULT_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER,
minimum_reply_surb_request_size: DEFAULT_MINIMUM_REPLY_SURB_REQUEST_SIZE,
maximum_reply_surb_request_size: DEFAULT_MAXIMUM_REPLY_SURB_REQUEST_SIZE,
maximum_allowed_reply_surb_request_size: DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE,
@@ -730,9 +713,6 @@ pub struct DebugConfig {
/// Defines all configuration options related to stats reporting.
pub stats_reporting: StatsReporting,
/// Defines all configuration options related to the forget me flag.
pub forget_me: ForgetMe,
}
impl DebugConfig {
@@ -755,69 +735,6 @@ impl Default for DebugConfig {
topology: Default::default(),
reply_surbs: Default::default(),
stats_reporting: Default::default(),
forget_me: Default::default(),
}
}
}
#[derive(Clone, Default, Debug, Deserialize, PartialEq, Serialize, Copy)]
pub struct ForgetMe {
client: bool,
stats: bool,
}
impl From<bool> for ForgetMe {
fn from(value: bool) -> Self {
if value {
Self::new_all()
} else {
Self::new_none()
}
}
}
impl ForgetMe {
pub fn new_all() -> Self {
Self {
client: true,
stats: true,
}
}
pub fn new_client() -> Self {
Self {
client: true,
stats: false,
}
}
pub fn new_stats() -> Self {
Self {
client: false,
stats: true,
}
}
pub fn new(client: bool, stats: bool) -> Self {
Self { client, stats }
}
pub fn any(&self) -> bool {
self.client || self.stats
}
pub fn client(&self) -> bool {
self.client
}
pub fn stats(&self) -> bool {
self.stats
}
pub fn new_none() -> Self {
Self {
client: false,
stats: false,
}
}
}
@@ -181,9 +181,8 @@ impl From<ConfigV5> for Config {
maximum_reply_surb_age: value.debug.reply_surbs.maximum_reply_surb_age,
maximum_reply_key_age: value.debug.reply_surbs.maximum_reply_key_age,
surb_mix_hops: value.debug.reply_surbs.surb_mix_hops,
..Default::default()
},
..Default::default()
stats_reporting: Default::default(),
},
}
}
@@ -1,24 +0,0 @@
/*
* Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: GPL-3.0-only
*/
-- make aes256gcm column non-nullable and drop any gateways that still use the legacy keys
-- (since they'd be unusable after this change)
CREATE TABLE remote_gateway_details_tmp
(
gateway_id_bs58 TEXT NOT NULL UNIQUE PRIMARY KEY REFERENCES registered_gateway (gateway_id_bs58),
derived_aes256_gcm_siv_key BLOB NOT NULL,
gateway_owner_address TEXT,
gateway_listener TEXT NOT NULL
);
INSERT INTO remote_gateway_details_tmp (gateway_id_bs58, derived_aes256_gcm_siv_key, gateway_owner_address,
gateway_listener)
SELECT gateway_id_bs58, derived_aes256_gcm_siv_key, gateway_owner_address, gateway_listener
FROM remote_gateway_details
WHERE derived_aes256_gcm_siv_key IS NOT NULL;
DROP TABLE remote_gateway_details;
ALTER TABLE remote_gateway_details_tmp
RENAME TO remote_gateway_details;
@@ -156,26 +156,48 @@ impl StorageManager {
pub(crate) async fn set_remote_gateway_details(
&self,
gateway_id_bs58: String,
derived_aes256_gcm_siv_key: &[u8],
gateway_owner_address: Option<String>,
gateway_listener: String,
remote: &RawRemoteGatewayDetails,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO remote_gateway_details(gateway_id_bs58, derived_aes256_gcm_siv_key, gateway_owner_address, gateway_listener)
VALUES (?, ?, ?, ?)
INSERT INTO remote_gateway_details(gateway_id_bs58, derived_aes128_ctr_blake3_hmac_keys_bs58, derived_aes256_gcm_siv_key, gateway_owner_address, gateway_listener)
VALUES (?, ?, ?, ?, ?)
"#,
gateway_id_bs58,
derived_aes256_gcm_siv_key,
gateway_owner_address,
gateway_listener,
remote.gateway_id_bs58,
remote.derived_aes128_ctr_blake3_hmac_keys_bs58,
remote.derived_aes256_gcm_siv_key,
remote.gateway_owner_address,
remote.gateway_listener,
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn update_remote_gateway_key(
&self,
gateway_id_bs58: &str,
derived_aes128_ctr_blake3_hmac_keys_bs58: Option<&str>,
derived_aes256_gcm_siv_key: Option<&[u8]>,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
UPDATE remote_gateway_details
SET
derived_aes128_ctr_blake3_hmac_keys_bs58 = ?,
derived_aes256_gcm_siv_key = ?
WHERE gateway_id_bs58 = ?
"#,
derived_aes128_ctr_blake3_hmac_keys_bs58,
derived_aes256_gcm_siv_key,
gateway_id_bs58
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn remove_remote_gateway_details(
&self,
gateway_id: &str,
@@ -8,6 +8,7 @@ use crate::{
use async_trait::async_trait;
use manager::StorageManager;
use nym_crypto::asymmetric::ed25519;
use nym_gateway_requests::SharedSymmetricKey;
use std::path::Path;
pub mod error;
@@ -118,16 +119,9 @@ impl GatewaysDetailsStore for OnDiskGatewaysDetails {
match &details.details {
GatewayDetails::Remote(remote_details) => {
let raw_details = remote_details.into();
self.manager
.set_remote_gateway_details(
remote_details.gateway_id.to_base58_string(),
remote_details.shared_key.as_bytes(),
remote_details
.gateway_owner_address
.as_ref()
.map(|o| o.to_string()),
remote_details.gateway_listener.to_string(),
)
.set_remote_gateway_details(&raw_details)
.await?;
}
GatewayDetails::Custom(custom_details) => {
@@ -140,6 +134,21 @@ impl GatewaysDetailsStore for OnDiskGatewaysDetails {
Ok(())
}
async fn upgrade_stored_remote_gateway_key(
&self,
gateway_id: ed25519::PublicKey,
updated_key: &SharedSymmetricKey,
) -> Result<(), Self::StorageError> {
self.manager
.update_remote_gateway_key(
&gateway_id.to_base58_string(),
None,
Some(updated_key.as_bytes()),
)
.await?;
Ok(())
}
// ideally all of those should be run under a storage tx to ensure storage consistency,
// but at that point it's fine
async fn remove_gateway_details(&self, gateway_id: &str) -> Result<(), Self::StorageError> {
@@ -2,8 +2,10 @@
// SPDX-License-Identifier: Apache-2.0
use crate::types::{ActiveGateway, GatewayRegistration};
use crate::{BadGateway, GatewaysDetailsStore};
use crate::{BadGateway, GatewayDetails, GatewaysDetailsStore};
use async_trait::async_trait;
use nym_crypto::asymmetric::ed25519::PublicKey;
use nym_gateway_requests::{SharedGatewayKey, SharedSymmetricKey};
use std::collections::HashMap;
use std::sync::Arc;
use thiserror::Error;
@@ -94,6 +96,29 @@ impl GatewaysDetailsStore for InMemGatewaysDetails {
Ok(())
}
async fn upgrade_stored_remote_gateway_key(
&self,
gateway_id: PublicKey,
updated_key: &SharedSymmetricKey,
) -> Result<(), Self::StorageError> {
let mut guard = self.inner.write().await;
#[allow(clippy::unwrap_used)]
if let Some(target) = guard.gateways.get_mut(&gateway_id.to_string()) {
let GatewayDetails::Remote(details) = &mut target.details else {
return Ok(());
};
assert_eq!(Arc::strong_count(&details.shared_key), 1);
// eh. that's nasty, but it's only ever used for ephemeral clients so should be fine for now...
details.shared_key = Arc::new(SharedGatewayKey::Current(
SharedSymmetricKey::try_from_bytes(updated_key.as_bytes()).unwrap(),
))
}
Ok(())
}
async fn remove_gateway_details(&self, gateway_id: &str) -> Result<(), Self::StorageError> {
let mut guard = self.inner.write().await;
if let Some(active) = guard.active_gateway.as_ref() {
@@ -6,6 +6,7 @@
use async_trait::async_trait;
use nym_crypto::asymmetric::identity;
use nym_gateway_requests::SharedSymmetricKey;
use std::error::Error;
pub mod backend;
@@ -61,6 +62,12 @@ pub trait GatewaysDetailsStore {
details: &GatewayRegistration,
) -> Result<(), Self::StorageError>;
async fn upgrade_stored_remote_gateway_key(
&self,
gateway_id: identity::PublicKey,
updated_key: &SharedSymmetricKey,
) -> Result<(), Self::StorageError>;
/// Remove given gateway details from the underlying store.
async fn remove_gateway_details(&self, gateway_id: &str) -> Result<(), Self::StorageError>;
}
@@ -4,9 +4,10 @@
use crate::BadGateway;
use cosmrs::AccountId;
use nym_crypto::asymmetric::identity;
use nym_gateway_requests::shared_key::SharedSymmetricKey;
use nym_gateway_requests::shared_key::{LegacySharedKeys, SharedGatewayKey, SharedSymmetricKey};
use serde::{Deserialize, Serialize};
use std::fmt::{Display, Formatter};
use std::ops::Deref;
use std::str::FromStr;
use std::sync::Arc;
use time::OffsetDateTime;
@@ -64,7 +65,7 @@ impl From<GatewayDetails> for GatewayRegistration {
impl GatewayDetails {
pub fn new_remote(
gateway_id: identity::PublicKey,
shared_key: Arc<SharedSymmetricKey>,
shared_key: Arc<SharedGatewayKey>,
gateway_owner_address: Option<AccountId>,
gateway_listener: Url,
) -> Self {
@@ -87,7 +88,7 @@ impl GatewayDetails {
}
}
pub fn shared_key(&self) -> Option<&SharedSymmetricKey> {
pub fn shared_key(&self) -> Option<&SharedGatewayKey> {
match self {
GatewayDetails::Remote(details) => Some(&details.shared_key),
GatewayDetails::Custom(_) => None,
@@ -167,7 +168,8 @@ pub struct RegisteredGateway {
#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
pub struct RawRemoteGatewayDetails {
pub gateway_id_bs58: String,
pub derived_aes256_gcm_siv_key: Vec<u8>,
pub derived_aes128_ctr_blake3_hmac_keys_bs58: Option<String>,
pub derived_aes256_gcm_siv_key: Option<Vec<u8>>,
pub gateway_owner_address: Option<String>,
pub gateway_listener: String,
}
@@ -184,11 +186,35 @@ impl TryFrom<RawRemoteGatewayDetails> for RemoteGatewayDetails {
}
})?;
let shared_key = SharedSymmetricKey::try_from_bytes(&value.derived_aes256_gcm_siv_key)
.map_err(|source| BadGateway::MalformedSharedKeys {
gateway_id: value.gateway_id_bs58.clone(),
source,
})?;
let shared_key =
match (
&value.derived_aes256_gcm_siv_key,
&value.derived_aes128_ctr_blake3_hmac_keys_bs58,
) {
(None, None) => {
return Err(BadGateway::MissingSharedKey {
gateway_id: value.gateway_id_bs58.clone(),
})
}
(Some(aes256gcm_siv), _) => {
let current_key =
SharedSymmetricKey::try_from_bytes(aes256gcm_siv).map_err(|source| {
BadGateway::MalformedSharedKeys {
gateway_id: value.gateway_id_bs58.clone(),
source,
}
})?;
SharedGatewayKey::Current(current_key)
}
(None, Some(aes128ctr_hmac)) => {
let legacy_key = LegacySharedKeys::try_from_base58_string(aes128ctr_hmac)
.map_err(|source| BadGateway::MalformedSharedKeys {
gateway_id: value.gateway_id_bs58.clone(),
source,
})?;
SharedGatewayKey::Legacy(legacy_key)
}
};
let gateway_owner_address = value
.gateway_owner_address
@@ -221,11 +247,29 @@ impl TryFrom<RawRemoteGatewayDetails> for RemoteGatewayDetails {
}
}
impl<'a> From<&'a RemoteGatewayDetails> for RawRemoteGatewayDetails {
fn from(value: &'a RemoteGatewayDetails) -> Self {
let (derived_aes128_ctr_blake3_hmac_keys_bs58, derived_aes256_gcm_siv_key) =
match value.shared_key.deref() {
SharedGatewayKey::Current(key) => (None, Some(key.to_bytes())),
SharedGatewayKey::Legacy(key) => (Some(key.to_base58_string()), None),
};
RawRemoteGatewayDetails {
gateway_id_bs58: value.gateway_id.to_base58_string(),
derived_aes128_ctr_blake3_hmac_keys_bs58,
derived_aes256_gcm_siv_key,
gateway_owner_address: value.gateway_owner_address.as_ref().map(|o| o.to_string()),
gateway_listener: value.gateway_listener.to_string(),
}
}
}
#[derive(Debug, Clone)]
pub struct RemoteGatewayDetails {
pub gateway_id: identity::PublicKey,
pub shared_key: Arc<SharedSymmetricKey>,
pub shared_key: Arc<SharedGatewayKey>,
pub gateway_owner_address: Option<AccountId>,
@@ -120,7 +120,6 @@ where
&core.client.nym_api_urls,
user_agent,
core.debug.topology.minimum_gateway_performance,
core.debug.topology.ignore_ingress_epoch_role,
)
.await?
};
@@ -93,10 +93,6 @@ pub struct CommonClientInitArgs {
/// Sets the address to report statistics
#[cfg_attr(feature = "cli", clap(long, hide = true))]
pub stats_reporting_address: Option<Recipient>,
/// Sets the forget me flag
#[cfg_attr(feature = "cli", clap(long, hide = true, default_value_t = false))]
pub forget_me: bool,
}
pub struct InitResultsWithConfig<T> {
@@ -179,7 +175,6 @@ where
&core.client.nym_api_urls,
user_agent,
core.debug.topology.minimum_gateway_performance,
core.debug.topology.ignore_ingress_epoch_role,
)
.await?
};
@@ -61,8 +61,4 @@ pub struct CommonClientRunArgs {
/// Sets the address to report statistics
#[cfg_attr(feature = "cli", clap(long, hide = true))]
pub stats_reporting_address: Option<Recipient>,
/// Sets the forget me flag
#[cfg_attr(feature = "cli", clap(long, hide = true, default_value_t = false))]
pub forget_me: bool,
}
@@ -1,7 +1,6 @@
// Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use super::mix_traffic::ClientRequestSender;
use super::received_buffer::ReceivedBufferMessage;
use super::statistics_control::StatisticsControl;
use crate::client::base_client::storage::helpers::store_client_keys;
@@ -32,15 +31,13 @@ use crate::init::{
setup_gateway,
types::{GatewaySetup, InitialisationResult},
};
use crate::{config, spawn_future};
use crate::{config, spawn_future, ForgetMe};
use futures::channel::mpsc;
use log::*;
use nym_bandwidth_controller::BandwidthController;
use nym_client_core_config_types::ForgetMe;
use nym_client_core_gateways_storage::{GatewayDetails, GatewaysDetailsStore};
use nym_credential_storage::storage::Storage as CredentialStorage;
use nym_crypto::asymmetric::{encryption, identity};
use nym_crypto::hkdf::DerivationMaterial;
use nym_gateway_client::client::config::GatewayClientConfig;
use nym_gateway_client::{
AcknowledgementReceiver, GatewayClient, GatewayConfig, MixnetMessageReceiver, PacketRouter,
@@ -178,8 +175,8 @@ impl From<bool> for CredentialsToggle {
}
}
pub struct BaseClientBuilder<C, S: MixnetClientStorage> {
config: Config,
pub struct BaseClientBuilder<'a, C, S: MixnetClientStorage> {
config: &'a Config,
client_store: S,
dkg_query_client: Option<C>,
@@ -194,19 +191,19 @@ pub struct BaseClientBuilder<C, S: MixnetClientStorage> {
#[cfg(unix)]
connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
derivation_material: Option<DerivationMaterial>,
forget_me: ForgetMe,
}
impl<C, S> BaseClientBuilder<C, S>
impl<'a, C, S> BaseClientBuilder<'a, C, S>
where
S: MixnetClientStorage + 'static,
C: DkgQueryClient + Send + Sync + 'static,
{
pub fn new(
base_config: Config,
base_config: &'a Config,
client_store: S,
dkg_query_client: Option<C>,
) -> BaseClientBuilder<C, S> {
) -> BaseClientBuilder<'a, C, S> {
BaseClientBuilder {
config: base_config,
client_store,
@@ -219,22 +216,13 @@ where
setup_method: GatewaySetup::MustLoad { gateway_id: None },
#[cfg(unix)]
connection_fd_callback: None,
derivation_material: None,
forget_me: Default::default(),
}
}
#[must_use]
pub fn with_derivation_material(
mut self,
derivation_material: Option<DerivationMaterial>,
) -> Self {
self.derivation_material = derivation_material;
self
}
#[must_use]
pub fn with_forget_me(mut self, forget_me: &ForgetMe) -> Self {
self.config.debug.forget_me = *forget_me;
self.forget_me = forget_me.clone();
self
}
@@ -310,7 +298,7 @@ where
topology_accessor: TopologyAccessor,
mix_tx: BatchMixMessageSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
shutdown: TaskClient,
) {
info!("Starting loop cover traffic stream...");
@@ -323,10 +311,9 @@ where
debug_config.traffic,
debug_config.cover_traffic,
stats_tx,
task_client,
);
stream.start();
stream.start_with_shutdown(shutdown);
}
#[allow(clippy::too_many_arguments)]
@@ -341,7 +328,7 @@ where
reply_controller_receiver: ReplyControllerReceiver,
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
task_client: TaskClient,
shutdown: TaskClient,
packet_type: PacketType,
stats_tx: ClientStatsSender,
) {
@@ -359,9 +346,8 @@ where
lane_queue_lengths,
client_connection_rx,
stats_tx,
task_client,
)
.start(packet_type);
.start_with_shutdown(shutdown, packet_type);
}
// buffer controlling all messages fetched from provider
@@ -384,9 +370,8 @@ where
reply_key_storage,
reply_controller_sender,
metrics_reporter,
shutdown,
);
controller.start()
controller.start_with_shutdown(shutdown)
}
#[allow(clippy::too_many_arguments)]
@@ -394,6 +379,7 @@ where
config: &Config,
initialisation_result: InitialisationResult,
bandwidth_controller: Option<BandwidthController<C, S::CredentialStore>>,
details_store: &S::GatewaysDetailsStore,
packet_router: PacketRouter,
stats_reporter: ClientStatsSender,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
@@ -402,6 +388,7 @@ where
where
<S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
<S::CredentialStore as CredentialStorage>::StorageError: Send + Sync + 'static,
<S::GatewaysDetailsStore as GatewaysDetailsStore>::StorageError: Sync + Send,
{
let managed_keys = initialisation_result.client_keys;
let GatewayDetails::Remote(details) = initialisation_result.gateway_registration.details
@@ -456,13 +443,31 @@ where
// we need to:
// - perform handshake (reg or auth)
// - check for key upgrade
// - maybe perform another upgrade handshake
// - check for bandwidth
// - start background tasks
let _auth_res = gateway_client
let auth_res = gateway_client
.perform_initial_authentication()
.await
.map_err(gateway_failure)?;
if auth_res.requires_key_upgrade {
// drop the shared_key arc because we don't need it and we can't hold it for the purposes of upgrade
drop(auth_res);
let updated_key = gateway_client
.upgrade_key_authenticated()
.await
.map_err(gateway_failure)?;
details_store
.upgrade_stored_remote_gateway_key(gateway_client.gateway_identity(), &updated_key)
.await.map_err(|err| {
error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
ClientCoreError::GatewaysDetailsStoreError { source: Box::new(err) }
})?
}
gateway_client
.claim_initial_bandwidth()
.await
@@ -481,6 +486,7 @@ where
config: &Config,
initialisation_result: InitialisationResult,
bandwidth_controller: Option<BandwidthController<C, S::CredentialStore>>,
details_store: &S::GatewaysDetailsStore,
packet_router: PacketRouter,
stats_reporter: ClientStatsSender,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
@@ -489,6 +495,7 @@ where
where
<S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
<S::CredentialStore as CredentialStorage>::StorageError: Send + Sync + 'static,
<S::GatewaysDetailsStore as GatewaysDetailsStore>::StorageError: Sync + Send,
{
// if we have setup custom gateway sender and persisted details agree with it, return it
if let Some(mut custom_gateway_transceiver) = custom_gateway_transceiver {
@@ -511,6 +518,7 @@ where
config,
initialisation_result,
bandwidth_controller,
details_store,
packet_router,
stats_reporter,
#[cfg(unix)]
@@ -552,22 +560,15 @@ where
topology_accessor: TopologyAccessor,
local_gateway: NodeIdentity,
wait_for_gateway: bool,
mut task_client: TaskClient,
mut shutdown: TaskClient,
) -> Result<(), ClientCoreError> {
let topology_refresher_config =
TopologyRefresherConfig::new(topology_config.topology_refresh_rate);
if topology_config.disable_refreshing {
// if we're not spawning the refresher, don't cause shutdown immediately
info!("The background topology refesher is not going to be started");
task_client.disarm();
}
let mut topology_refresher = TopologyRefresher::new(
topology_refresher_config,
topology_accessor,
topology_provider,
task_client,
);
// before returning, block entire runtime to refresh the current network view so that any
// components depending on topology would see a non-empty view
@@ -608,11 +609,15 @@ where
}
}
if !topology_config.disable_refreshing {
if topology_config.disable_refreshing {
// if we're not spawning the refresher, don't cause shutdown immediately
info!("The topology refesher is not going to be started");
shutdown.disarm();
} else {
// don't spawn the refresher if we don't want to be refreshing the topology.
// only use the initial values obtained
info!("Starting topology refresher...");
topology_refresher.start();
topology_refresher.start_with_shutdown(shutdown);
}
Ok(())
@@ -623,29 +628,30 @@ where
user_agent: Option<UserAgent>,
client_stats_id: String,
input_sender: Sender<InputMessage>,
task_client: TaskClient,
shutdown: TaskClient,
) -> ClientStatsSender {
info!("Starting statistics control...");
StatisticsControl::create_and_start(
StatisticsControl::create_and_start_with_shutdown(
config.debug.stats_reporting,
user_agent
.map(|u| u.application)
.unwrap_or("unknown".to_string()),
client_stats_id,
input_sender.clone(),
task_client,
shutdown.with_suffix("controller"),
)
}
fn start_mix_traffic_controller(
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
shutdown: TaskClient,
) -> (BatchMixMessageSender, ClientRequestSender) {
forget_me: ForgetMe,
) -> BatchMixMessageSender {
info!("Starting mix traffic controller...");
let (mix_traffic_controller, mix_tx, client_tx) =
MixTrafficController::new(gateway_transceiver, shutdown);
mix_traffic_controller.start();
(mix_tx, client_tx)
let (mix_traffic_controller, mix_tx) =
MixTrafficController::new(gateway_transceiver, forget_me);
mix_traffic_controller.start_with_shutdown(shutdown);
mix_tx
}
// TODO: rename it as it implies the data is persistent whilst one can use InMemBackend
@@ -680,7 +686,6 @@ where
setup_method: GatewaySetup,
key_store: &S::KeyStore,
details_store: &S::GatewaysDetailsStore,
derivation_material: Option<DerivationMaterial>,
) -> Result<InitialisationResult, ClientCoreError>
where
<S::KeyStore as KeyStore>::StorageError: Sync + Send,
@@ -690,12 +695,7 @@ where
if key_store.load_keys().await.is_err() {
info!("could not find valid client keys - a new set will be generated");
let mut rng = OsRng;
let keys = if let Some(derivation_material) = derivation_material {
ClientKeys::from_master_key(&mut rng, &derivation_material)
.map_err(|_| ClientCoreError::HkdfDerivationError {})?
} else {
ClientKeys::generate_new(&mut rng)
};
let keys = ClientKeys::generate_new(&mut rng);
store_client_keys(keys, key_store).await?;
}
@@ -717,11 +717,11 @@ where
self.setup_method,
self.client_store.key_store(),
self.client_store.gateway_details_store(),
self.derivation_material,
)
.await?;
let (reply_storage_backend, credential_store, _) = self.client_store.into_runtime_stores();
let (reply_storage_backend, credential_store, details_store) =
self.client_store.into_runtime_stores();
// channels for inter-component communication
// TODO: make the channels be internally created by the relevant components
@@ -773,7 +773,7 @@ where
);
let stats_reporter = Self::start_statistics_control(
&self.config,
self.config,
self.user_agent.clone(),
generate_client_stats_id(*self_address.identity()),
input_sender.clone(),
@@ -799,9 +799,10 @@ where
let gateway_transceiver = Self::setup_gateway_transceiver(
self.custom_gateway_transceiver,
&self.config,
self.config,
init_res,
bandwidth_controller,
&details_store,
gateway_packet_router,
stats_reporter.clone(),
#[cfg(unix)]
@@ -832,9 +833,10 @@ where
// traffic stream.
// The MixTrafficController then sends the actual traffic
let (message_sender, client_request_sender) = Self::start_mix_traffic_controller(
let message_sender = Self::start_mix_traffic_controller(
gateway_transceiver,
shutdown.fork("mix_traffic_controller"),
self.forget_me,
);
// Channels that the websocket listener can use to signal downstream to the real traffic
@@ -909,8 +911,6 @@ where
},
stats_reporter,
task_handle: shutdown,
client_request_sender,
forget_me: self.config.debug.forget_me,
})
}
}
@@ -922,7 +922,6 @@ pub struct BaseClient {
pub client_output: ClientOutputStatus,
pub client_state: ClientState,
pub stats_reporter: ClientStatsSender,
pub client_request_sender: ClientRequestSender,
pub task_handle: TaskHandle,
pub forget_me: ForgetMe,
}
@@ -2,11 +2,24 @@
// SPDX-License-Identifier: Apache-2.0
pub mod v1_1_33 {
use crate::client::base_client::{
non_wasm_helpers::setup_fs_gateways_storage,
storage::helpers::{set_active_gateway, store_gateway_details},
};
use crate::config::disk_persistence::old_v1_1_33::CommonClientPathsV1_1_33;
use crate::config::disk_persistence::CommonClientPaths;
use crate::config::old_config_v1_1_33::OldGatewayEndpointConfigV1_1_33;
use crate::error::ClientCoreError;
use nym_client_core_gateways_storage::{
CustomGatewayDetails, GatewayDetails, GatewayRegistration, RemoteGatewayDetails,
};
use nym_gateway_requests::shared_key::LegacySharedKeys;
use serde::{Deserialize, Serialize};
use sha2::{digest::Digest, Sha256};
use std::ops::Deref;
use std::path::Path;
use std::sync::Arc;
use zeroize::Zeroizing;
mod base64 {
use base64::{engine::general_purpose::STANDARD, Engine as _};
@@ -44,18 +57,155 @@ pub mod v1_1_33 {
details: OldGatewayEndpointConfigV1_1_33,
}
impl PersistedGatewayConfig {
fn verify(&self, shared_key: &LegacySharedKeys) -> bool {
let key_bytes = Zeroizing::new(shared_key.to_bytes());
let mut key_hasher = Sha256::new();
key_hasher.update(&key_bytes);
let key_hash = key_hasher.finalize();
self.key_hash == key_hash.deref()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PersistedCustomGatewayDetails {
gateway_id: String,
}
fn load_shared_key<P: AsRef<Path>>(path: P) -> Result<LegacySharedKeys, ClientCoreError> {
// the shared key was a simple pem file
Ok(nym_pemstore::load_key(path)?)
}
fn gateway_details_from_raw(
gateway_id: String,
gateway_owner: String,
gateway_listener: String,
gateway_shared_key: LegacySharedKeys,
) -> Result<GatewayDetails, ClientCoreError> {
Ok(GatewayDetails::Remote(RemoteGatewayDetails {
gateway_id: gateway_id
.parse()
.map_err(|err| ClientCoreError::UpgradeFailure {
message: format!("the stored gateway id was malformed: {err}"),
})?,
shared_key: Arc::new(gateway_shared_key.into()),
gateway_owner_address: Some(gateway_owner.parse().map_err(|err| {
ClientCoreError::UpgradeFailure {
message: format!("the stored gateway owner address was malformed: {err}"),
}
})?),
gateway_listener: gateway_listener.parse().map_err(|err| {
ClientCoreError::UpgradeFailure {
message: format!("the stored gateway listener address was malformed: {err}"),
}
})?,
}))
}
// helper to extract shared key and gateway details into the new GatewayRegistration
fn extract_gateway_registration(
storage_paths: &CommonClientPathsV1_1_33,
) -> Result<GatewayRegistration, ClientCoreError> {
let details_file = std::fs::File::open(&storage_paths.gateway_details).map_err(|err| {
ClientCoreError::UpgradeFailure {
message: format!(
"failed to open gateway details file at {}: {err}",
storage_paths.gateway_details.display()
),
}
})?;
// in v1.1.33 of the clients, the gateway details struct was saved as json
let details: PersistedGatewayDetails =
serde_json::from_reader(details_file).map_err(|err| {
ClientCoreError::UpgradeFailure {
message: format!(
"failed to deserialize gateway details from {}: {err}",
storage_paths.gateway_details.display()
),
}
})?;
let details = match details {
PersistedGatewayDetails::Default(config) => {
let gateway_shared_key =
load_shared_key(&storage_paths.keys.gateway_shared_key_file)?;
if !config.verify(&gateway_shared_key) {
return Err(ClientCoreError::UpgradeFailure {
message: "failed to verify consistency of the existing gateway details"
.to_string(),
});
}
gateway_details_from_raw(
config.details.gateway_id,
config.details.gateway_owner,
config.details.gateway_listener,
gateway_shared_key,
)?
}
PersistedGatewayDetails::Custom(custom) => {
GatewayDetails::Custom(CustomGatewayDetails {
gateway_id: custom.gateway_id.parse().map_err(|err| {
ClientCoreError::UpgradeFailure {
message: format!("the stored gateway id was malformed: {err}"),
}
})?,
data: None,
})
}
};
Ok(details.into())
}
// it's responsibility of the caller to ensure this is called **after** new registration has already been saved
fn remove_old_gateway_details(storage_paths: &CommonClientPathsV1_1_33) -> std::io::Result<()> {
std::fs::remove_file(&storage_paths.gateway_details)?;
if storage_paths.keys.gateway_shared_key_file.exists() {
std::fs::remove_file(&storage_paths.keys.gateway_shared_key_file)?;
}
Ok(())
}
pub async fn migrate_gateway_details(
_old_storage_paths: &CommonClientPathsV1_1_33,
_new_storage_paths: &CommonClientPaths,
_preloaded_config: Option<OldGatewayEndpointConfigV1_1_33>,
old_storage_paths: &CommonClientPathsV1_1_33,
new_storage_paths: &CommonClientPaths,
preloaded_config: Option<OldGatewayEndpointConfigV1_1_33>,
) -> Result<(), ClientCoreError> {
Err(ClientCoreError::UnsupportedMigration(
"migration of legacy keys has been removed and is no longer supported".into(),
))
let gateway_registration = match preloaded_config {
Some(config) => {
let gateway_shared_key =
load_shared_key(&old_storage_paths.keys.gateway_shared_key_file)?;
gateway_details_from_raw(
config.gateway_id,
config.gateway_owner,
config.gateway_listener,
gateway_shared_key,
)?
.into()
}
None => extract_gateway_registration(old_storage_paths)?,
};
// since we're migrating to a brand new store, the store should be empty
// and thus set the 'new' gateway as the active one
let details_store =
setup_fs_gateways_storage(&new_storage_paths.gateway_registrations).await?;
store_gateway_details(&details_store, &gateway_registration).await?;
set_active_gateway(
&details_store,
&gateway_registration.details.gateway_id().to_base58_string(),
)
.await?;
remove_old_gateway_details(old_storage_paths).map_err(|err| {
ClientCoreError::UpgradeFailure {
message: format!("failed to remove old data: {err}"),
}
})
}
}
@@ -4,8 +4,6 @@
// TODO: combine those more closely. Perhaps into a single underlying store.
// Like for persistent, on-disk, storage, what's the point of having 3 different databases?
use rand::rngs::OsRng;
use crate::client::key_manager::persistence::{InMemEphemeralKeys, KeyStore};
use crate::client::replies::reply_storage;
use crate::client::replies::reply_storage::ReplyStorageBackend;
@@ -65,7 +63,7 @@ pub trait MixnetClientStorage {
fn gateway_details_store(&self) -> &Self::GatewaysDetailsStore;
}
#[derive(Clone)]
#[derive(Clone, Default)]
pub struct Ephemeral {
key_store: InMemEphemeralKeys,
reply_store: reply_storage::Empty,
@@ -73,14 +71,9 @@ pub struct Ephemeral {
gateway_details_store: InMemGatewaysDetails,
}
impl Default for Ephemeral {
fn default() -> Self {
Ephemeral {
key_store: InMemEphemeralKeys::new(&mut OsRng),
reply_store: Default::default(),
credential_store: Default::default(),
gateway_details_store: Default::default(),
}
impl Ephemeral {
pub fn new() -> Self {
Default::default()
}
}
@@ -13,7 +13,6 @@ use nym_sphinx::cover::generate_loop_cover_packet;
use nym_sphinx::params::{PacketSize, PacketType};
use nym_sphinx::utils::sample_poisson_duration;
use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender};
use nym_task::TaskClient;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::pin::Pin;
use std::sync::Arc;
@@ -65,8 +64,6 @@ where
packet_type: PacketType,
stats_tx: ClientStatsSender,
task_client: TaskClient,
}
impl<R> Stream for LoopCoverTrafficStream<R>
@@ -113,7 +110,6 @@ impl LoopCoverTrafficStream<OsRng> {
traffic_config: config::Traffic,
cover_config: config::CoverTraffic,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
let rng = OsRng;
@@ -132,7 +128,6 @@ impl LoopCoverTrafficStream<OsRng> {
secondary_packet_size: traffic_config.secondary_packet_size,
packet_type: traffic_config.packet_type,
stats_tx,
task_client,
}
}
@@ -180,7 +175,7 @@ impl LoopCoverTrafficStream<OsRng> {
}
};
let cover_message = match generate_loop_cover_packet(
let cover_message = generate_loop_cover_packet(
&mut self.rng,
topology_ref,
&self.ack_key,
@@ -189,15 +184,8 @@ impl LoopCoverTrafficStream<OsRng> {
self.cover_traffic.loop_cover_traffic_average_delay,
cover_traffic_packet_size,
self.packet_type,
) {
Ok(cover_message) => cover_message,
Err(err) => {
warn!(
"Somehow failed to generate a loop cover message with a valid topology: {err}"
);
return;
}
};
)
.expect("Somehow failed to generate a loop cover message with a valid topology");
if let Err(err) = self.mix_tx.try_send(vec![cover_message]) {
match err {
@@ -229,7 +217,7 @@ impl LoopCoverTrafficStream<OsRng> {
tokio::task::yield_now().await;
}
pub fn start(mut self) {
pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) {
if self.cover_traffic.disable_loop_cover_traffic_stream {
// we should have never got here in the first place - the task should have never been created to begin with
// so panic and review the code that lead to this branch
@@ -243,8 +231,6 @@ impl LoopCoverTrafficStream<OsRng> {
);
self.set_next_delay(sampled);
let mut shutdown = self.task_client.fork("select");
spawn_future(async move {
debug!("Started LoopCoverTrafficStream with graceful shutdown support");
@@ -2,18 +2,14 @@
// SPDX-License-Identifier: Apache-2.0
use crate::client::key_manager::persistence::KeyStore;
use nym_crypto::{
asymmetric::{encryption, identity},
hkdf::{DerivationMaterial, InvalidLength},
};
use nym_gateway_requests::shared_key::SharedSymmetricKey;
use nym_crypto::asymmetric::{encryption, identity};
use nym_gateway_requests::shared_key::{LegacySharedKeys, SharedGatewayKey, SharedSymmetricKey};
use nym_sphinx::acknowledgements::AckKey;
use rand::{CryptoRng, RngCore};
use std::sync::Arc;
use zeroize::ZeroizeOnDrop;
pub mod persistence;
mod test;
// Note: to support key rotation in the future, all keys will require adding an extra smart pointer,
// most likely an AtomicCell, or if it doesn't work as I think it does, a Mutex. Although I think
@@ -47,24 +43,6 @@ impl ClientKeys {
}
}
pub fn from_master_key<R>(
rng: &mut R,
derivation_material: &DerivationMaterial,
) -> Result<Self, InvalidLength>
where
R: RngCore + CryptoRng,
{
let secret = derivation_material.derive_secret()?;
Ok(ClientKeys {
identity_keypair: Arc::new(identity::KeyPair::from_secret(
secret,
derivation_material.index(),
)),
encryption_keypair: Arc::new(encryption::KeyPair::new(rng)),
ack_key: Arc::new(AckKey::new(rng)),
})
}
pub fn from_keys(
id_keypair: identity::KeyPair,
enc_keypair: encryption::KeyPair,
@@ -106,5 +84,7 @@ fn _assert_keys_zeroize_on_drop() {
_assert_zeroize_on_drop::<identity::KeyPair>();
_assert_zeroize_on_drop::<encryption::KeyPair>();
_assert_zeroize_on_drop::<AckKey>();
_assert_zeroize_on_drop::<LegacySharedKeys>();
_assert_zeroize_on_drop::<SharedSymmetricKey>();
_assert_zeroize_on_drop::<SharedGatewayKey>();
}
@@ -3,7 +3,6 @@
use crate::client::key_manager::ClientKeys;
use async_trait::async_trait;
use rand::{CryptoRng, RngCore};
use std::error::Error;
use std::sync::Arc;
use tokio::sync::Mutex;
@@ -196,20 +195,9 @@ impl KeyStore for OnDiskKeys {
}
}
#[derive(Clone)]
#[derive(Clone, Default)]
pub struct InMemEphemeralKeys {
keys: Arc<Mutex<ClientKeys>>,
}
impl InMemEphemeralKeys {
pub fn new<R>(rng: &mut R) -> Self
where
R: RngCore + CryptoRng,
{
InMemEphemeralKeys {
keys: Arc::new(Mutex::new(ClientKeys::generate_new(rng))),
}
}
keys: Arc<Mutex<Option<ClientKeys>>>,
}
#[derive(Debug, thiserror::Error)]
@@ -222,11 +210,11 @@ impl KeyStore for InMemEphemeralKeys {
type StorageError = EphemeralKeysError;
async fn load_keys(&self) -> Result<ClientKeys, Self::StorageError> {
Ok(self.keys.lock().await.clone())
self.keys.lock().await.clone().ok_or(EphemeralKeysError)
}
async fn store_keys(&self, keys: &ClientKeys) -> Result<(), Self::StorageError> {
*self.keys.lock().await = keys.clone();
*self.keys.lock().await = Some(keys.clone());
Ok(())
}
}
@@ -1,89 +0,0 @@
#[cfg(test)]
mod tests {
use crate::client::key_manager::ClientKeys;
use nym_crypto::hkdf::DerivationMaterial;
use rand::SeedableRng;
use rand_chacha::ChaCha20Rng;
#[test]
fn test_from_master_key_success() {
// Set up a deterministic RNG.
let seed = [33u8; 32];
let mut rng = ChaCha20Rng::from_seed(seed);
// Set up the derivation material.
let master_key = b"this is a secret master key";
let salt = b"unique-salt";
let derivation_material = DerivationMaterial::new(master_key, 0, salt);
// Generate ClientKeys from the master key.
let client_keys = ClientKeys::from_master_key(&mut rng, &derivation_material)
.expect("Failed to create client keys");
assert_eq!(
client_keys.identity_keypair().public_key().to_string(),
String::from("FX4Undr5LPPBA7zThWWpAKXKQTXSbW1C28PnxbCqUkU4")
);
assert_eq!(
client_keys.identity_keypair().private_key().to_string(),
String::from("6S3uMi2rU5SwyUUYCiMrF5qqdcYnEDMYLggBSvavVzEt")
);
}
#[test]
fn test_from_master_key_deterministic_identity() {
// Using identical derivation material should result in the exactly same identity keypair.
let seed = [1u8; 32];
let mut rng1 = ChaCha20Rng::from_seed(seed);
let mut rng2 = ChaCha20Rng::from_seed(seed);
let master_key = b"another secret master key";
let salt = b"deterministic-salt";
let index = 7u32;
let derivation_material = DerivationMaterial::new(master_key, index, salt);
let client_keys1 = ClientKeys::from_master_key(&mut rng1, &derivation_material)
.expect("Failed to create client keys (first instance)");
let client_keys2 = ClientKeys::from_master_key(&mut rng2, &derivation_material)
.expect("Failed to create client keys (second instance)");
assert_eq!(
client_keys1.identity_keypair().public_key().to_string(),
client_keys2.identity_keypair().public_key().to_string()
);
assert_eq!(
client_keys1.identity_keypair().private_key().to_string(),
client_keys2.identity_keypair().private_key().to_string()
);
}
#[test]
fn test_from_master_key_different_indices() {
// Changing the index should yield a different identity key.
let seed = [5u8; 32];
let mut rng = ChaCha20Rng::from_seed(seed);
let master_key = b"same secret key";
let salt = b"same-salt";
let derivation_material1 = DerivationMaterial::new(master_key, 1, salt);
let derivation_material2 = DerivationMaterial::new(master_key, 2, salt);
let client_keys1 = ClientKeys::from_master_key(&mut rng, &derivation_material1)
.expect("Failed to create client keys for index 1");
let client_keys2 = ClientKeys::from_master_key(&mut rng, &derivation_material2)
.expect("Failed to create client keys for index 2");
assert_ne!(
client_keys1.identity_keypair().public_key().to_string(),
client_keys2.identity_keypair().public_key().to_string()
);
assert_ne!(
client_keys1.identity_keypair().private_key().to_string(),
client_keys2.identity_keypair().private_key().to_string()
);
}
}
@@ -2,18 +2,13 @@
// SPDX-License-Identifier: Apache-2.0
use crate::client::mix_traffic::transceiver::GatewayTransceiver;
use crate::error::ClientCoreError;
use crate::spawn_future;
use crate::{spawn_future, ForgetMe};
use log::*;
use nym_gateway_requests::ClientRequest;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_task::TaskClient;
use transceiver::ErasedGatewayError;
pub type BatchMixMessageSender = tokio::sync::mpsc::Sender<Vec<MixPacket>>;
pub type BatchMixMessageReceiver = tokio::sync::mpsc::Receiver<Vec<MixPacket>>;
pub type ClientRequestReceiver = tokio::sync::mpsc::Receiver<ClientRequest>;
pub type ClientRequestSender = tokio::sync::mpsc::Sender<ClientRequest>;
pub mod transceiver;
@@ -28,73 +23,52 @@ pub struct MixTrafficController {
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
mix_rx: BatchMixMessageReceiver,
client_rx: ClientRequestReceiver,
// TODO: this is temporary work-around.
// in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
consecutive_gateway_failure_count: usize,
task_client: TaskClient,
forget_me: ForgetMe,
}
impl MixTrafficController {
pub fn new<T>(
gateway_transceiver: T,
task_client: TaskClient,
) -> (
MixTrafficController,
BatchMixMessageSender,
ClientRequestSender,
)
forget_me: ForgetMe,
) -> (MixTrafficController, BatchMixMessageSender)
where
T: GatewayTransceiver + Send + 'static,
{
let (message_sender, message_receiver) =
tokio::sync::mpsc::channel(MIX_MESSAGE_RECEIVER_BUFFER_SIZE);
let (client_sender, client_receiver) = tokio::sync::mpsc::channel(1);
(
MixTrafficController {
gateway_transceiver: Box::new(gateway_transceiver),
mix_rx: message_receiver,
client_rx: client_receiver,
consecutive_gateway_failure_count: 0,
task_client,
forget_me,
},
message_sender,
client_sender,
)
}
pub fn new_dynamic(
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
task_client: TaskClient,
) -> (
MixTrafficController,
BatchMixMessageSender,
ClientRequestSender,
) {
forget_me: ForgetMe,
) -> (MixTrafficController, BatchMixMessageSender) {
let (message_sender, message_receiver) =
tokio::sync::mpsc::channel(MIX_MESSAGE_RECEIVER_BUFFER_SIZE);
let (client_sender, client_receiver) = tokio::sync::mpsc::channel(1);
(
MixTrafficController {
gateway_transceiver,
mix_rx: message_receiver,
client_rx: client_receiver,
consecutive_gateway_failure_count: 0,
task_client,
forget_me,
},
message_sender,
client_sender,
)
}
async fn on_messages(
&mut self,
mut mix_packets: Vec<MixPacket>,
) -> Result<(), ErasedGatewayError> {
async fn on_messages(&mut self, mut mix_packets: Vec<MixPacket>) {
debug_assert!(!mix_packets.is_empty());
let result = if mix_packets.len() == 1 {
@@ -106,60 +80,64 @@ impl MixTrafficController {
.await
};
if result.is_err() {
self.consecutive_gateway_failure_count += 1;
} else {
trace!("We *might* have managed to forward sphinx packet(s) to the gateway!");
self.consecutive_gateway_failure_count = 0;
match result {
Err(err) => {
error!("Failed to send sphinx packet(s) to the gateway: {err}");
self.consecutive_gateway_failure_count += 1;
if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
// todo: in the future this should initiate a 'graceful' shutdown or try
// to reconnect?
panic!("failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead. Can't do anything about it yet :(")
}
}
Ok(_) => {
trace!("We *might* have managed to forward sphinx packet(s) to the gateway!");
self.consecutive_gateway_failure_count = 0;
}
}
result
}
pub fn start(mut self) {
pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) {
spawn_future(async move {
debug!("Started MixTrafficController with graceful shutdown support");
while !self.task_client.is_shutdown() {
loop {
tokio::select! {
mix_packets = self.mix_rx.recv() => match mix_packets {
Some(mix_packets) => {
if let Err(err) = self.on_messages(mix_packets).await {
error!("Failed to send sphinx packet(s) to the gateway: {err}");
if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
// Disconnect from the gateway. If we should try to re-connect
// is handled at a higher layer.
error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
// Do we need to handle the embedded mixnet client case
// separately?
self.task_client.send_we_stopped(Box::new(ClientCoreError::GatewayFailedToForwardMessages));
break;
}
}
self.on_messages(mix_packets).await;
},
None => {
log::trace!("MixTrafficController: Stopping since channel closed");
break;
}
},
client_request = self.client_rx.recv() => match client_request {
Some(client_request) => {
match self.gateway_transceiver.send_client_request(client_request).await {
Ok(_) => (),
Err(e) => error!("Failed to send client request: {}", e),
};
},
None => {
log::trace!("MixTrafficController, client request channel closed");
}
},
_ = self.task_client.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("MixTrafficController: Received shutdown");
break;
}
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
if self.forget_me.any() {
log::info!("Sending forget me request to the gateway");
match self
.gateway_transceiver
.send_client_request(ClientRequest::ForgetMe {
client: self.forget_me.client(),
stats: self.forget_me.stats(),
})
.await
{
Ok(_) => {
log::info!("Successfully sent forget me request to the gateway");
}
Err(err) => {
log::error!("Failed to send forget me request to the gateway: {err}");
}
}
}
log::debug!("MixTrafficController: Exiting");
});
@@ -86,9 +86,7 @@ impl<G: GatewayTransceiver + ?Sized + Send> GatewayTransceiver for Box<G> {
&mut self,
message: ClientRequest,
) -> Result<(), GatewayClientError> {
let _ = (**self).send_client_request(message.clone()).await?;
log::debug!("Sent client request: {:?}", message);
Ok(())
(**self).send_client_request(message).await
}
}
@@ -145,7 +143,14 @@ where
&mut self,
message: ClientRequest,
) -> Result<(), GatewayClientError> {
self.gateway_client.send_client_request(message).await
if let Some(shared_key) = self.gateway_client.shared_key() {
self.gateway_client
.send_websocket_message(message.encrypt(&*shared_key)?)
.await?;
Ok(())
} else {
Err(GatewayClientError::ConnectionInInvalidState)
}
}
}
@@ -11,7 +11,6 @@ use nym_sphinx::{
acknowledgements::{identifier::recover_identifier, AckKey},
chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID},
};
use nym_task::TaskClient;
use std::sync::Arc;
/// Module responsible for listening for any data resembling acknowledgements from the network
@@ -21,7 +20,6 @@ pub(super) struct AcknowledgementListener {
ack_receiver: AcknowledgementReceiver,
action_sender: AckActionSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
}
impl AcknowledgementListener {
@@ -30,14 +28,12 @@ impl AcknowledgementListener {
ack_receiver: AcknowledgementReceiver,
action_sender: AckActionSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
AcknowledgementListener {
ack_key,
ack_receiver,
action_sender,
stats_tx,
task_client,
}
}
@@ -68,14 +64,9 @@ impl AcknowledgementListener {
trace!("Received {} from the mix network", frag_id);
self.stats_tx
.report(PacketStatisticsEvent::RealAckReceived(ack_content.len()).into());
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::new_remove(frag_id))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send remove action to action controller: {err}");
}
}
.unwrap();
}
async fn handle_ack_receiver_item(&mut self, item: Vec<Vec<u8>>) {
@@ -85,10 +76,10 @@ impl AcknowledgementListener {
}
}
pub(super) async fn run(&mut self) {
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started AcknowledgementListener with graceful shutdown support");
while !self.task_client.is_shutdown() {
while !shutdown.is_shutdown() {
tokio::select! {
acks = self.ack_receiver.next() => match acks {
Some(acks) => self.handle_ack_receiver_item(acks).await,
@@ -97,12 +88,12 @@ impl AcknowledgementListener {
break;
}
},
_ = self.task_client.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("AcknowledgementListener: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
log::debug!("AcknowledgementListener: Exiting");
}
}
@@ -9,7 +9,6 @@ use log::*;
use nym_nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, QueueKey};
use nym_sphinx::chunking::fragment::FragmentIdentifier;
use nym_sphinx::Delay as SphinxDelay;
use nym_task::TaskClient;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
@@ -102,8 +101,6 @@ pub(super) struct ActionController {
/// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements.
retransmission_sender: RetransmissionRequestSender,
task_client: TaskClient,
}
impl ActionController {
@@ -111,7 +108,6 @@ impl ActionController {
config: Config,
retransmission_sender: RetransmissionRequestSender,
incoming_actions: AckActionReceiver,
task_client: TaskClient,
) -> Self {
ActionController {
config,
@@ -119,7 +115,6 @@ impl ActionController {
pending_acks_timers: NonExhaustiveDelayQueue::new(),
incoming_actions,
retransmission_sender,
task_client,
}
}
@@ -221,7 +216,11 @@ impl ActionController {
}
// note: when the entry expires it's automatically removed from pending_acks_timers
fn handle_expired_ack_timer(&mut self, expired_ack: Expired<FragmentIdentifier>) {
fn handle_expired_ack_timer(
&mut self,
expired_ack: Expired<FragmentIdentifier>,
task_client: &mut nym_task::TaskClient,
) {
// I'm honestly not sure how to handle it, because getting it means other things in our
// system are already misbehaving. If we ever see this panic, then I guess we should worry
// about it. Perhaps just reschedule it at later point?
@@ -239,13 +238,15 @@ impl ActionController {
// downgrading an arc and then upgrading vs cloning is difference of 30ns vs 15ns
// so it's literally a NO difference while it might prevent us from unnecessarily
// resending data (in maybe 1 in 1 million cases, but it's something)
if let Err(err) = self
if self
.retransmission_sender
.unbounded_send(Arc::downgrade(pending_ack_data))
.is_err()
{
if !self.task_client.is_shutdown_poll() {
log::error!("Failed to send pending ack for retransmission: {err}");
}
assert!(
task_client.is_shutdown_poll(),
"Failed to send pending ack for retransmission"
);
}
} else {
// this shouldn't cause any issues but shouldn't have happened to begin with!
@@ -264,10 +265,10 @@ impl ActionController {
}
}
pub(super) async fn run(&mut self) {
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started ActionController with graceful shutdown support");
while !self.task_client.is_shutdown() {
loop {
tokio::select! {
action = self.incoming_actions.next() => match action {
Some(action) => self.process_action(action),
@@ -279,19 +280,19 @@ impl ActionController {
}
},
expired_ack = self.pending_acks_timers.next() => match expired_ack {
Some(expired_ack) => self.handle_expired_ack_timer(expired_ack),
Some(expired_ack) => self.handle_expired_ack_timer(expired_ack, &mut shutdown),
None => {
log::trace!("ActionController: Stopping since ack channel closed");
break;
}
},
_ = self.task_client.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("ActionController: Received shutdown");
break;
}
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
log::debug!("ActionController: Exiting");
}
}
@@ -11,7 +11,6 @@ use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
/// Module responsible for dealing with the received messages: splitting them, creating acknowledgements,
@@ -24,7 +23,6 @@ where
input_receiver: InputMessageReceiver,
message_handler: MessageHandler<R>,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
}
impl<R> InputMessageListener<R>
@@ -38,13 +36,11 @@ where
input_receiver: InputMessageReceiver,
message_handler: MessageHandler<R>,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
) -> Self {
InputMessageListener {
input_receiver,
message_handler,
reply_controller_sender,
task_client,
}
}
@@ -67,14 +63,8 @@ where
lane: TransmissionLane,
) {
// offload reply handling to the dedicated task
if let Err(err) = self
.reply_controller_sender
self.reply_controller_sender
.send_reply(recipient_tag, data, lane)
{
if !self.task_client.is_shutdown_poll() {
error!("failed to send a reply - {err}");
}
}
}
async fn handle_plain_message(
@@ -174,10 +164,10 @@ where
};
}
pub(super) async fn run(&mut self) {
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started InputMessageListener with graceful shutdown support");
while !self.task_client.is_shutdown() {
while !shutdown.is_shutdown() {
tokio::select! {
input_msg = self.input_receiver.recv() => match input_msg {
Some(input_msg) => {
@@ -188,12 +178,12 @@ where
break;
}
},
_ = self.task_client.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("InputMessageListener: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
log::debug!("InputMessageListener: Exiting");
}
}
@@ -24,7 +24,6 @@ use nym_sphinx::{
Delay as SphinxDelay,
};
use nym_statistics_common::clients::ClientStatsSender;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::{
sync::{Arc, Weak},
@@ -67,7 +66,7 @@ pub(crate) enum PacketDestination {
/// Structure representing a data `Fragment` that is on-route to the specified `Recipient`
#[derive(Debug)]
pub struct PendingAcknowledgement {
pub(crate) struct PendingAcknowledgement {
message_chunk: Fragment,
delay: SphinxDelay,
destination: PacketDestination,
@@ -217,7 +216,6 @@ where
message_handler: MessageHandler<R>,
reply_controller_sender: ReplyControllerSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
@@ -227,7 +225,6 @@ where
action_config,
retransmission_tx,
connectors.ack_action_receiver,
task_client.fork("action_controller"),
);
// will listen for any acks coming from the network
@@ -236,7 +233,6 @@ where
connectors.ack_receiver,
connectors.ack_action_sender.clone(),
stats_tx,
task_client.fork("acknowledgement_listener"),
);
// will listen for any new messages from the client
@@ -244,7 +240,6 @@ where
connectors.input_receiver,
message_handler.clone(),
reply_controller_sender.clone(),
task_client.fork("input_message_listener"),
);
// will listen for any ack timeouts and trigger retransmission
@@ -254,16 +249,12 @@ where
message_handler,
retransmission_rx,
reply_controller_sender,
task_client.fork("retransmission_request_listener"),
);
// will listen for events indicating the packet was sent through the network so that
// the retransmission timer should be started.
let sent_notification_listener = SentNotificationListener::new(
connectors.sent_notifier,
connectors.ack_action_sender,
task_client.with_suffix("sent_notification_listener"),
);
let sent_notification_listener =
SentNotificationListener::new(connectors.sent_notifier, connectors.ack_action_sender);
AcknowledgementController {
acknowledgement_listener,
@@ -274,35 +265,53 @@ where
}
}
pub(super) fn start(self, packet_type: PacketType) {
pub(super) fn start_with_shutdown(
self,
shutdown: nym_task::TaskClient,
packet_type: PacketType,
) {
let mut acknowledgement_listener = self.acknowledgement_listener;
let mut input_message_listener = self.input_message_listener;
let mut retransmission_request_listener = self.retransmission_request_listener;
let mut sent_notification_listener = self.sent_notification_listener;
let mut action_controller = self.action_controller;
let shutdown_handle = shutdown.fork("acknowledgement_listener");
spawn_future(async move {
acknowledgement_listener.run().await;
acknowledgement_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The acknowledgement listener has finished execution!");
});
let shutdown_handle = shutdown.fork("input_message_listener");
spawn_future(async move {
input_message_listener.run().await;
input_message_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The input listener has finished execution!");
});
let shutdown_handle = shutdown.fork("retransmission_request_listener");
spawn_future(async move {
retransmission_request_listener.run(packet_type).await;
retransmission_request_listener
.run_with_shutdown(shutdown_handle, packet_type)
.await;
debug!("The retransmission request listener has finished execution!");
});
let shutdown_handle = shutdown.fork("sent_notification_listener");
spawn_future(async move {
sent_notification_listener.run().await;
sent_notification_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The sent notification listener has finished execution!");
});
spawn_future(async move {
action_controller.run().await;
action_controller
.run_with_shutdown(shutdown.with_suffix("action_controller"))
.await;
debug!("The controller has finished execution!");
});
}
@@ -14,7 +14,7 @@ use log::*;
use nym_sphinx::chunking::fragment::Fragment;
use nym_sphinx::preparer::PreparedFragment;
use nym_sphinx::{addressing::clients::Recipient, params::PacketType};
use nym_task::{connections::TransmissionLane, TaskClient};
use nym_task::connections::TransmissionLane;
use rand::{CryptoRng, Rng};
use std::sync::{Arc, Weak};
@@ -25,7 +25,6 @@ pub(super) struct RetransmissionRequestListener<R> {
message_handler: MessageHandler<R>,
request_receiver: RetransmissionRequestReceiver,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
}
impl<R> RetransmissionRequestListener<R>
@@ -38,7 +37,6 @@ where
message_handler: MessageHandler<R>,
request_receiver: RetransmissionRequestReceiver,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
) -> Self {
RetransmissionRequestListener {
maximum_retransmissions,
@@ -46,7 +44,6 @@ where
message_handler,
request_receiver,
reply_controller_sender,
task_client,
}
}
@@ -82,12 +79,9 @@ where
if let Some(limit) = self.maximum_retransmissions {
if timed_out_ack.retransmissions >= limit {
warn!("reached maximum number of allowed retransmissions for the packet");
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::new_remove(frag_id))
{
error!("Failed to send remove action to the controller: {err}");
}
.unwrap();
return;
}
}
@@ -99,16 +93,11 @@ where
} => {
// if this is retransmission for reply, offload it to the dedicated task
// that deals with all the surbs
if let Err(err) = self.reply_controller_sender.send_retransmission_data(
return self.reply_controller_sender.send_retransmission_data(
*recipient_tag,
weak_timed_out_ack,
*extra_surb_request,
) {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send retransmission data to the reply controller: {err}");
}
}
return;
);
}
PacketDestination::KnownRecipient(recipient) => {
self.prepare_normal_retransmission_chunk(
@@ -125,12 +114,9 @@ where
Err(err) => {
warn!("Could not retransmit the packet - {err}");
// we NEED to start timer here otherwise we will have this guy permanently stuck in memory
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::new_start_timer(frag_id))
{
error!("Failed to send start timer action to the controller: {err}");
}
.unwrap();
return;
}
};
@@ -155,14 +141,9 @@ where
// is sent to the `OutQueueControl` and has gone through its internal queue
// with the additional poisson delay.
// And since Actions are executed in order `UpdateTimer` will HAVE TO be executed before `StartTimer`
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::new_update_pending_ack(frag_id, new_delay))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send update pending ack action to the controller: {err}");
}
}
.unwrap();
// send to `OutQueueControl` to eventually send to the mix network
self.message_handler
@@ -176,10 +157,14 @@ where
.await
}
pub(super) async fn run(&mut self, packet_type: PacketType) {
pub(super) async fn run_with_shutdown(
&mut self,
mut shutdown: nym_task::TaskClient,
packet_type: PacketType,
) {
debug!("Started RetransmissionRequestListener with graceful shutdown support");
while !self.task_client.is_shutdown() {
while !shutdown.is_shutdown() {
tokio::select! {
timed_out_ack = self.request_receiver.next() => match timed_out_ack {
Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack, packet_type).await,
@@ -188,12 +173,12 @@ where
break;
}
},
_ = self.task_client.recv() => {
_ = shutdown.recv() => {
log::trace!("RetransmissionRequestListener: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
log::debug!("RetransmissionRequestListener: Exiting");
}
}
@@ -6,7 +6,6 @@ use super::SentPacketNotificationReceiver;
use futures::StreamExt;
use log::*;
use nym_sphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID};
use nym_task::TaskClient;
/// Module responsible for starting up retransmission timers.
/// It is required because when we send our packet to the `real traffic stream` controlled
@@ -15,19 +14,16 @@ use nym_task::TaskClient;
pub(super) struct SentNotificationListener {
sent_notifier: SentPacketNotificationReceiver,
action_sender: AckActionSender,
task_client: TaskClient,
}
impl SentNotificationListener {
pub(super) fn new(
sent_notifier: SentPacketNotificationReceiver,
action_sender: AckActionSender,
task_client: TaskClient,
) -> Self {
SentNotificationListener {
sent_notifier,
action_sender,
task_client,
}
}
@@ -36,20 +32,15 @@ impl SentNotificationListener {
trace!("sent off a cover message - no need to start retransmission timer!");
return;
}
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::new_start_timer(frag_id))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send start timer action to action controller: {err}");
}
}
.unwrap();
}
pub(super) async fn run(&mut self) {
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started SentNotificationListener with graceful shutdown support");
while !self.task_client.is_shutdown() {
loop {
tokio::select! {
frag_id = self.sent_notifier.next() => match frag_id {
Some(frag_id) => {
@@ -60,13 +51,13 @@ impl SentNotificationListener {
break;
}
},
_ = self.task_client.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("SentNotificationListener: Received shutdown");
break;
}
}
}
assert!(self.task_client.is_shutdown_poll());
assert!(shutdown.is_shutdown_poll());
log::debug!("SentNotificationListener: Exiting");
}
}
@@ -19,7 +19,6 @@ use nym_sphinx::params::{PacketSize, PacketType};
use nym_sphinx::preparer::{MessagePreparer, PreparedFragment};
use nym_sphinx::Delay;
use nym_task::connections::TransmissionLane;
use nym_task::TaskClient;
use nym_topology::{NymRouteProvider, NymTopologyError};
use rand::{CryptoRng, Rng};
use std::collections::HashMap;
@@ -150,14 +149,12 @@ pub(crate) struct MessageHandler<R> {
topology_access: TopologyAccessor,
reply_key_storage: SentReplyKeys,
tag_storage: UsedSenderTags,
task_client: TaskClient,
}
impl<R> MessageHandler<R>
where
R: CryptoRng + Rng,
{
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
config: Config,
rng: R,
@@ -166,7 +163,6 @@ where
topology_access: TopologyAccessor,
reply_key_storage: SentReplyKeys,
tag_storage: UsedSenderTags,
task_client: TaskClient,
) -> Self
where
R: Copy,
@@ -187,7 +183,6 @@ where
topology_access,
reply_key_storage,
tag_storage,
task_client,
}
}
@@ -614,25 +609,15 @@ where
}
pub(crate) fn update_ack_delay(&self, id: FragmentIdentifier, new_delay: Delay) {
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::UpdatePendingAck(id, new_delay))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send update action to the controller: {err}");
}
}
.expect("action control task has died")
}
pub(crate) fn insert_pending_acks(&self, pending_acks: Vec<PendingAcknowledgement>) {
if let Err(err) = self
.action_sender
self.action_sender
.unbounded_send(Action::new_insert(pending_acks))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send insert action to the controller: {err}");
}
}
.expect("action control task has died")
}
// tells real message sender (with the poisson timer) to send this to the mix network
@@ -641,14 +626,9 @@ where
messages: Vec<RealMessage>,
transmission_lane: TransmissionLane,
) {
if let Err(err) = self
.real_message_sender
self.real_message_sender
.send((messages, transmission_lane))
.await
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to forward messages to the real message sender: {err}");
}
}
.expect("real message receiver task (OutQueueControl) has died");
}
}
@@ -31,7 +31,6 @@ use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::PacketType;
use nym_statistics_common::clients::ClientStatsSender;
use nym_task::connections::{ConnectionCommandReceiver, LaneQueueLengths};
use nym_task::TaskClient;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::sync::Arc;
@@ -148,7 +147,6 @@ impl RealMessagesController<OsRng> {
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
let rng = OsRng;
@@ -179,7 +177,6 @@ impl RealMessagesController<OsRng> {
topology_access.clone(),
reply_storage.key_storage(),
reply_storage.tags_storage(),
task_client.fork("message_handler"),
);
let ack_control = AcknowledgementController::new(
@@ -189,7 +186,6 @@ impl RealMessagesController<OsRng> {
message_handler.clone(),
reply_controller_sender,
stats_tx.clone(),
task_client.fork("ack_control"),
);
let reply_control = ReplyController::new(
@@ -197,7 +193,6 @@ impl RealMessagesController<OsRng> {
message_handler,
reply_storage,
reply_controller_receiver,
task_client.fork("reply_controller"),
);
let out_queue_control = OutQueueControl::new(
@@ -210,7 +205,6 @@ impl RealMessagesController<OsRng> {
lane_queue_lengths,
client_connection_rx,
stats_tx,
task_client.with_suffix("out_queue_control"),
);
RealMessagesController {
@@ -220,20 +214,22 @@ impl RealMessagesController<OsRng> {
}
}
pub fn start(self, packet_type: PacketType) {
pub fn start_with_shutdown(self, shutdown: nym_task::TaskClient, packet_type: PacketType) {
let mut out_queue_control = self.out_queue_control;
let ack_control = self.ack_control;
let mut reply_control = self.reply_control;
let shutdown_handle = shutdown.fork("out_queue_control");
spawn_future(async move {
out_queue_control.run().await;
out_queue_control.run_with_shutdown(shutdown_handle).await;
debug!("The out queue controller has finished execution!");
});
let shutdown_handle = shutdown.fork("reply_control");
spawn_future(async move {
reply_control.run().await;
reply_control.run_with_shutdown(shutdown_handle).await;
debug!("The reply controller has finished execution!");
});
ack_control.start(packet_type);
ack_control.start_with_shutdown(shutdown.with_suffix("ack_control"), packet_type);
}
}
@@ -22,7 +22,6 @@ use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, C
use nym_task::connections::{
ConnectionCommand, ConnectionCommandReceiver, ConnectionId, LaneQueueLengths, TransmissionLane,
};
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::pin::Pin;
use std::sync::Arc;
@@ -118,8 +117,6 @@ where
/// Channel used for sending metrics events (specifically `PacketStatistics` events) to the metrics tracker.
stats_tx: ClientStatsSender,
task_client: TaskClient,
}
#[derive(Debug)]
@@ -179,7 +176,6 @@ where
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
OutQueueControl {
config,
@@ -194,7 +190,6 @@ where
client_connection_rx,
lane_queue_lengths,
stats_tx,
task_client,
}
}
@@ -203,9 +198,7 @@ where
// queues and client load rather than the required delay. So realistically we can treat
// whatever is about to happen as negligible additional delay.
trace!("{} is about to get sent to the mixnet", frag_id);
if let Err(err) = self.sent_notifier.unbounded_send(frag_id) {
error!("Failed to notify about sent message: {err}");
}
self.sent_notifier.unbounded_send(frag_id).unwrap();
}
fn loop_cover_message_size(&mut self) -> PacketSize {
@@ -278,9 +271,7 @@ where
};
if let Err(err) = self.mix_tx.send(vec![next_message]).await {
if !self.task_client.is_shutdown_poll() {
log::error!("Failed to send: {err}");
}
log::error!("Failed to send: {err}");
} else {
let event = if fragment_id.is_some() {
PacketStatisticsEvent::RealPacketSent(packet_size)
@@ -513,29 +504,21 @@ where
}
#[cfg(not(target_arch = "wasm32"))]
fn log_status(&self, shutdown: &mut TaskClient) {
fn log_status(&self, shutdown: &mut nym_task::TaskClient) {
use crate::error::ClientCoreStatusMessage;
let packets = self.transmission_buffer.total_size();
let lanes = self.transmission_buffer.lanes();
let backlog = self.transmission_buffer.total_size_in_bytes() as f64 / 1024.0;
let lanes = self.transmission_buffer.num_lanes();
let mult = self.sending_delay_controller.current_multiplier();
let delay = self.current_average_message_sending_delay().as_millis();
let lane_status = lanes
.iter()
.map(|lane_name| {
let lane_length = self.transmission_buffer.lane_length(lane_name).unwrap_or(0);
format!("{lane_name:?}: {lane_length}")
})
.collect::<Vec<String>>()
.join(", ");
let status_str = if self.config.traffic.disable_main_poisson_packet_distribution {
format!("Packet backlog: {lane_status}, no delay")
format!("Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, no delay")
} else {
format!("Packet backlog: {lane_status}, avg delay: {delay}ms ({mult})")
format!(
"Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, avg delay: {delay}ms ({mult})"
)
};
if packets > 1000 {
log::warn!("{status_str}");
} else if packets > 0 {
@@ -552,19 +535,17 @@ where
}
}
pub(super) async fn run(&mut self) {
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started OutQueueControl with graceful shutdown support");
let mut shutdown = self.task_client.fork("select");
#[cfg(not(target_arch = "wasm32"))]
{
let mut status_timer = tokio::time::interval(Duration::from_secs(5));
while !shutdown.is_shutdown() {
loop {
tokio::select! {
biased;
_ = shutdown.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("OutQueueControl: Received shutdown");
break;
}
@@ -20,7 +20,6 @@ use nym_sphinx::message::{NymMessage, PlainMessage};
use nym_sphinx::params::ReplySurbKeyDigestAlgorithm;
use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage};
use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender};
use nym_task::TaskClient;
use std::collections::HashSet;
use std::sync::Arc;
@@ -153,7 +152,6 @@ struct ReceivedMessagesBuffer<R: MessageReceiver> {
inner: Arc<Mutex<ReceivedMessagesBufferInner<R>>>,
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
}
impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
@@ -162,7 +160,6 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
ReceivedMessagesBuffer {
inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
@@ -175,7 +172,6 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
})),
reply_key_storage,
reply_controller_sender,
task_client,
}
}
@@ -261,15 +257,11 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
}
};
if let Err(err) = self.reply_controller_sender.send_additional_surbs(
self.reply_controller_sender.send_additional_surbs(
msg.sender_tag,
reply_surbs,
from_surb_request,
) {
if !self.task_client.is_shutdown_poll() {
error!("{err}");
}
}
)
}
reconstructed
}
@@ -284,14 +276,8 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
ReplyMessageContent::Data { message } => reconstructed.push(message.into()),
ReplyMessageContent::SurbRequest { recipient, amount } => {
debug!("received request for {amount} additional reply SURBs from {recipient}");
if let Err(err) = self
.reply_controller_sender
.send_additional_surbs_request(*recipient, amount)
{
if !self.task_client.is_shutdown_poll() {
error!("{err}");
}
}
self.reply_controller_sender
.send_additional_surbs_request(*recipient, amount);
}
}
}
@@ -413,19 +399,16 @@ pub enum ReceivedBufferMessage {
struct RequestReceiver<R: MessageReceiver> {
received_buffer: ReceivedMessagesBuffer<R>,
query_receiver: ReceivedBufferRequestReceiver,
task_client: TaskClient,
}
impl<R: MessageReceiver> RequestReceiver<R> {
fn new(
received_buffer: ReceivedMessagesBuffer<R>,
query_receiver: ReceivedBufferRequestReceiver,
task_client: TaskClient,
) -> Self {
RequestReceiver {
received_buffer,
query_receiver,
task_client,
}
}
@@ -440,12 +423,12 @@ impl<R: MessageReceiver> RequestReceiver<R> {
}
}
async fn run(&mut self) {
async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started RequestReceiver with graceful shutdown support");
while !self.task_client.is_shutdown() {
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = self.task_client.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("RequestReceiver: Received shutdown");
}
request = self.query_receiver.next() => {
@@ -458,7 +441,7 @@ impl<R: MessageReceiver> RequestReceiver<R> {
},
}
}
self.task_client.recv().await;
shutdown.recv_timeout().await;
log::debug!("RequestReceiver: Exiting");
}
}
@@ -466,25 +449,25 @@ impl<R: MessageReceiver> RequestReceiver<R> {
struct FragmentedMessageReceiver<R: MessageReceiver> {
received_buffer: ReceivedMessagesBuffer<R>,
mixnet_packet_receiver: MixnetMessageReceiver,
task_client: TaskClient,
}
impl<R: MessageReceiver> FragmentedMessageReceiver<R> {
fn new(
received_buffer: ReceivedMessagesBuffer<R>,
mixnet_packet_receiver: MixnetMessageReceiver,
task_client: TaskClient,
) -> Self {
FragmentedMessageReceiver {
received_buffer,
mixnet_packet_receiver,
task_client,
}
}
async fn run(&mut self) -> Result<(), MessageRecoveryError> {
async fn run_with_shutdown(
&mut self,
mut shutdown: nym_task::TaskClient,
) -> Result<(), MessageRecoveryError> {
debug!("Started FragmentedMessageReceiver with graceful shutdown support");
while !self.task_client.is_shutdown() {
while !shutdown.is_shutdown() {
tokio::select! {
new_messages = self.mixnet_packet_receiver.next() => {
if let Some(new_messages) = new_messages {
@@ -494,12 +477,12 @@ impl<R: MessageReceiver> FragmentedMessageReceiver<R> {
break;
}
},
_ = self.task_client.recv_with_delay() => {
_ = shutdown.recv_with_delay() => {
log::trace!("FragmentedMessageReceiver: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
log::debug!("FragmentedMessageReceiver: Exiting");
Ok(())
}
@@ -518,42 +501,41 @@ impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferControll
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
metrics_reporter: ClientStatsSender,
task_client: TaskClient,
) -> Self {
let received_buffer = ReceivedMessagesBuffer::new(
local_encryption_keypair,
reply_key_storage,
reply_controller_sender,
metrics_reporter,
task_client.fork("received_messages_buffer"),
);
ReceivedMessagesBufferController {
fragmented_message_receiver: FragmentedMessageReceiver::new(
received_buffer.clone(),
mixnet_packet_receiver,
task_client.fork("fragmented_message_receiver"),
),
request_receiver: RequestReceiver::new(
received_buffer,
query_receiver,
task_client.with_suffix("request_receiver"),
),
request_receiver: RequestReceiver::new(received_buffer, query_receiver),
}
}
pub fn start(self) {
pub fn start_with_shutdown(self, shutdown: nym_task::TaskClient) {
let mut fragmented_message_receiver = self.fragmented_message_receiver;
let mut request_receiver = self.request_receiver;
let shutdown_handle = shutdown.fork("fragmented_message_receiver");
spawn_future(async move {
match fragmented_message_receiver.run().await {
match fragmented_message_receiver
.run_with_shutdown(shutdown_handle)
.await
{
Ok(_) => {}
Err(e) => error!("{e}"),
}
});
spawn_future(async move {
request_receiver.run().await;
request_receiver
.run_with_shutdown(shutdown.with_suffix("request_receiver"))
.await;
});
}
}
@@ -12,7 +12,6 @@ use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::anonymous_replies::ReplySurb;
use nym_sphinx::chunking::fragment::{Fragment, FragmentIdentifier};
use nym_task::connections::{ConnectionId, TransmissionLane};
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::cmp::{max, min};
use std::collections::btree_map::Entry;
@@ -69,9 +68,6 @@ pub struct ReplyController<R> {
message_handler: MessageHandler<R>,
full_reply_storage: CombinedReplyStorage,
// Listen for shutdown signals
task_client: TaskClient,
}
impl<R> ReplyController<R>
@@ -83,7 +79,6 @@ where
message_handler: MessageHandler<R>,
full_reply_storage: CombinedReplyStorage,
request_receiver: ReplyControllerReceiver,
task_client: TaskClient,
) -> Self {
ReplyController {
config,
@@ -92,7 +87,6 @@ where
pending_retransmissions: HashMap::new(),
message_handler,
full_reply_storage,
task_client,
}
}
@@ -102,7 +96,6 @@ where
fragments: I,
lane: TransmissionLane,
) {
trace!("buffering pending replies for {recipient}");
self.pending_replies
.entry(*recipient)
.or_insert_with(TransmissionBuffer::new)
@@ -114,7 +107,6 @@ where
recipient: &AnonymousSenderTag,
fragments: Vec<(TransmissionLane, Fragment)>,
) {
trace!("re-inserting pending replies for {recipient}");
// the buffer should ALWAYS exist at this point, if it doesn't, it's a bug...
self.pending_replies
.entry(*recipient)
@@ -127,7 +119,6 @@ where
recipient: &AnonymousSenderTag,
data: Vec<Arc<PendingAcknowledgement>>,
) {
trace!("re-inserting pending retransmissions for {recipient}");
// the underlying entry MUST exist as we've just got data from there
let map_entry = self
.pending_retransmissions
@@ -145,7 +136,7 @@ where
}
fn should_request_more_surbs(&self, target: &AnonymousSenderTag) -> bool {
trace!("checking if we should request more surbs from {target}");
trace!("checking if we should request more surbs from {:?}", target);
let pending_queue_size = self
.pending_replies
@@ -161,6 +152,11 @@ where
let total_queue = pending_queue_size + retransmission_queue;
// simple as that - there's absolutely nothing to retransmit
if total_queue == 0 {
return false;
}
let available_surbs = self
.full_reply_storage
.surbs_storage_ref()
@@ -177,27 +173,11 @@ where
.full_reply_storage
.surbs_storage_ref()
.max_surb_threshold();
let min_surbs_threshold_buffer =
self.config.reply_surbs.minimum_reply_surb_threshold_buffer;
// After clearing the queue, we want to have at least `min_surbs_threshold` surbs available
// and reserved for requesting additional surbs, and in addition to that we also want to
// have `min_surbs_threshold_buffer` surbs available proactively.
let target_surbs_after_clearing_queue = min_surbs_threshold + min_surbs_threshold_buffer;
debug!("total queue size: {total_queue} = pending data {pending_queue_size} + pending retransmission {retransmission_queue}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..{max_surbs_threshold}");
// Check if we have enough surbs to handle the total queue and maintain minimum thresholds
let total_required_surbs = total_queue + target_surbs_after_clearing_queue;
let total_available_surbs = pending_surbs + available_surbs;
debug!("total queue size: {total_queue} = pending data {pending_queue_size} + pending retransmission {retransmission_queue}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..+{min_surbs_threshold_buffer}..{max_surbs_threshold}");
// We should request more surbs if:
// 1. We haven't hit the maximum surb threshold, and
// 2. We don't have enough surbs to handle the queue plus minimum thresholds
let is_below_max_threshold = total_available_surbs < max_surbs_threshold;
let is_below_required_surbs = total_available_surbs < total_required_surbs;
is_below_max_threshold && is_below_required_surbs
(pending_surbs + available_surbs) < max_surbs_threshold
&& (pending_surbs + available_surbs) < (total_queue + min_surbs_threshold)
}
async fn handle_send_reply(
@@ -258,10 +238,6 @@ where
&recipient_tag,
);
warn!("failed to send reply to {recipient_tag}: {err}");
info!(
"buffering {no_fragments} fragments for {recipient_tag}",
no_fragments = to_send.len()
);
self.insert_pending_replies(&recipient_tag, to_send, lane);
}
}
@@ -269,13 +245,6 @@ where
// if there's leftover data we didn't send because we didn't have enough (or any) surbs - buffer it
if !fragments.is_empty() {
// Ideally we should have enough surbs above the minimum threshold to handle sending
// new replies without having to first request more surbs. That's why I'd like to log
// these cases as they might indicate a problem with the surb management.
debug!(
"buffering {no_fragments} fragments for {recipient_tag}",
no_fragments = fragments.len()
);
self.insert_pending_replies(&recipient_tag, fragments, lane);
}
@@ -290,7 +259,6 @@ where
target: AnonymousSenderTag,
amount: u32,
) -> Result<(), PreparationError> {
debug!("requesting {amount} additional reply surbs for {target}");
let reply_surb = self
.full_reply_storage
.surbs_storage_ref()
@@ -712,7 +680,7 @@ where
// it should take into consideration the average latency, sending rate and queue size.
// it should request as many surbs as it takes to saturate its sending rate before next batch arrives
async fn request_reply_surbs_for_queue_clearing(&mut self, target: AnonymousSenderTag) {
trace!("requesting surbs for queue clearing");
trace!("requesting surbs for queues clearing");
let pending_queue_size = self
.pending_replies
@@ -726,18 +694,17 @@ where
.map(|pending_queue| pending_queue.len())
.unwrap_or_default();
let min_surbs_buffer = self.config.reply_surbs.minimum_reply_surb_threshold_buffer as u32;
let total_queue = (pending_queue_size + retransmission_queue) as u32;
// To proactively request additional surbs, we aim to have a buffer of extra surbs in our
// storage.
let total_queue_with_buffer = total_queue + min_surbs_buffer;
if total_queue == 0 {
trace!("the pending queues for {:?} are already empty", target);
return;
}
let request_size = min(
self.config.reply_surbs.maximum_reply_surb_request_size,
max(
total_queue_with_buffer,
total_queue,
self.config.reply_surbs.minimum_reply_surb_request_size,
),
);
@@ -879,11 +846,9 @@ where
// todo!()
// }
pub(crate) async fn run(&mut self) {
pub(crate) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
debug!("Started ReplyController with graceful shutdown support");
let mut shutdown = self.task_client.fork("select");
let polling_rate = Duration::from_secs(5);
let mut stale_inspection = new_interval_stream(polling_rate);
@@ -895,7 +860,7 @@ where
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
_ = shutdown.recv_with_delay() => {
log::trace!("ReplyController: Received shutdown");
},
req = self.request_receiver.next() => match req {
@@ -15,27 +15,6 @@ pub(crate) fn new_control_channels() -> (ReplyControllerSender, ReplyControllerR
(tx.into(), rx)
}
#[derive(Debug, thiserror::Error)]
pub enum ReplyControllerSenderError {
#[error("failed to send retransmission data to reply controller")]
SendRetransmissionData(#[source] mpsc::TrySendError<ReplyControllerMessage>),
#[error("failed to send reply to reply controller")]
SendReply(#[source] mpsc::TrySendError<ReplyControllerMessage>),
#[error("failed to send additional surbs to reply controller")]
AdditionalSurbs(#[source] mpsc::TrySendError<ReplyControllerMessage>),
#[error("failed to send additional surbs request to reply controller")]
AdditionalSurbsRequest(#[source] mpsc::TrySendError<ReplyControllerMessage>),
#[error("failed to request lane queue length from reply controller")]
LaneQueueLength(#[source] mpsc::TrySendError<ReplyControllerMessage>),
#[error("response channel was dropped before we could receive the response")]
ResponseChannelDropped(#[source] oneshot::Canceled),
}
#[derive(Debug, Clone)]
pub struct ReplyControllerSender(mpsc::UnboundedSender<ReplyControllerMessage>);
@@ -51,14 +30,14 @@ impl ReplyControllerSender {
recipient: AnonymousSenderTag,
timed_out_ack: Weak<PendingAcknowledgement>,
extra_surb_request: bool,
) -> Result<(), ReplyControllerSenderError> {
) {
self.0
.unbounded_send(ReplyControllerMessage::RetransmitReply {
recipient,
timed_out_ack,
extra_surb_request,
})
.map_err(ReplyControllerSenderError::SendRetransmissionData)
.expect("ReplyControllerReceiver has died!")
}
pub(crate) fn send_reply(
@@ -66,14 +45,14 @@ impl ReplyControllerSender {
recipient: AnonymousSenderTag,
message: Vec<u8>,
lane: TransmissionLane,
) -> Result<(), ReplyControllerSenderError> {
) {
self.0
.unbounded_send(ReplyControllerMessage::SendReply {
recipient,
message,
lane,
})
.map_err(ReplyControllerSenderError::SendReply)
.expect("ReplyControllerReceiver has died!")
}
pub(crate) fn send_additional_surbs(
@@ -81,47 +60,42 @@ impl ReplyControllerSender {
sender_tag: AnonymousSenderTag,
reply_surbs: Vec<ReplySurb>,
from_surb_request: bool,
) -> Result<(), ReplyControllerSenderError> {
) {
self.0
.unbounded_send(ReplyControllerMessage::AdditionalSurbs {
sender_tag,
reply_surbs,
from_surb_request,
})
.map_err(ReplyControllerSenderError::AdditionalSurbs)
.expect("ReplyControllerReceiver has died!")
}
pub(crate) fn send_additional_surbs_request(
&self,
recipient: Recipient,
amount: u32,
) -> Result<(), ReplyControllerSenderError> {
pub(crate) fn send_additional_surbs_request(&self, recipient: Recipient, amount: u32) {
self.0
.unbounded_send(ReplyControllerMessage::AdditionalSurbsRequest {
recipient: Box::new(recipient),
amount,
})
.map_err(ReplyControllerSenderError::AdditionalSurbsRequest)
.expect("ReplyControllerReceiver has died!")
}
pub async fn get_lane_queue_length(
&self,
connection_id: ConnectionId,
) -> Result<usize, ReplyControllerSenderError> {
pub async fn get_lane_queue_length(&self, connection_id: ConnectionId) -> usize {
let (response_tx, response_rx) = oneshot::channel();
if let Err(err) = self
.0
self.0
.unbounded_send(ReplyControllerMessage::LaneQueueLength {
connection_id,
response_channel: response_tx,
})
{
return Err(ReplyControllerSenderError::LaneQueueLength(err));
}
.expect("ReplyControllerReceiver has died!");
response_rx
.await
.map_err(ReplyControllerSenderError::ResponseChannelDropped)
match response_rx.await {
Ok(length) => length,
Err(_) => {
error!("The reply controller has dropped our response channel!");
// TODO: should we panic here instead? this message implies something weird and unrecoverable has happened
0
}
}
}
}
@@ -136,10 +110,7 @@ impl ReplyQueueLengths {
}
}
pub async fn get_lane_queue_length(
&self,
connection_id: ConnectionId,
) -> Result<usize, ReplyControllerSenderError> {
pub async fn get_lane_queue_length(&self, connection_id: ConnectionId) -> usize {
self.reply_controller_sender
.get_lane_queue_length(connection_id)
.await
@@ -149,7 +120,7 @@ impl ReplyQueueLengths {
pub(crate) type ReplyControllerReceiver = mpsc::UnboundedReceiver<ReplyControllerMessage>;
#[derive(Debug)]
pub enum ReplyControllerMessage {
pub(crate) enum ReplyControllerMessage {
RetransmitReply {
recipient: AnonymousSenderTag,
timed_out_ack: Weak<PendingAcknowledgement>,
@@ -22,7 +22,7 @@ use nym_sphinx::addressing::Recipient;
use nym_statistics_common::clients::{
ClientStatsController, ClientStatsReceiver, ClientStatsSender,
};
use nym_task::{connections::TransmissionLane, TaskClient};
use nym_task::connections::TransmissionLane;
use std::time::Duration;
use crate::{
@@ -51,9 +51,6 @@ pub(crate) struct StatisticsControl {
/// Config for stats reporting (enabled, address, interval)
reporting_config: StatsReporting,
/// Task client for listening for shutdown
task_client: TaskClient,
}
impl StatisticsControl {
@@ -62,24 +59,19 @@ impl StatisticsControl {
client_type: String,
client_stats_id: String,
report_tx: InputMessageSender,
task_client: TaskClient,
) -> (Self, ClientStatsSender) {
let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel();
let stats = ClientStatsController::new(client_stats_id, client_type);
let mut task_client_stats_sender = task_client.fork("stats_sender");
task_client_stats_sender.disarm();
(
StatisticsControl {
stats,
stats_rx,
report_tx,
reporting_config,
task_client,
},
ClientStatsSender::new(Some(stats_tx), task_client_stats_sender),
ClientStatsSender::new(Some(stats_tx)),
)
}
@@ -99,7 +91,7 @@ impl StatisticsControl {
}
}
async fn run(&mut self) {
async fn run_with_shutdown(&mut self, mut task_client: nym_task::TaskClient) {
log::debug!("Started StatisticsControl with graceful shutdown support");
#[cfg(not(target_arch = "wasm32"))]
@@ -129,13 +121,8 @@ impl StatisticsControl {
let mut snapshot_interval =
gloo_timers::future::IntervalStream::new(SNAPSHOT_INTERVAL.as_millis() as u32);
while !self.task_client.is_shutdown() {
loop {
tokio::select! {
biased;
_ = self.task_client.recv() => {
log::trace!("StatisticsControl: Received shutdown");
break;
},
stats_event = self.stats_rx.recv() => match stats_event {
Some(stats_event) => self.stats.handle_event(stats_event),
None => {
@@ -157,34 +144,34 @@ impl StatisticsControl {
}
_ = local_report_interval.next() => {
self.stats.local_report(&mut self.task_client);
self.stats.local_report(&mut task_client);
}
_ = task_client.recv_with_delay() => {
log::trace!("StatisticsControl: Received shutdown");
break;
},
}
}
task_client.recv_timeout().await;
log::debug!("StatisticsControl: Exiting");
}
pub(crate) fn start(mut self) {
pub(crate) fn start_with_shutdown(mut self, task_client: nym_task::TaskClient) {
spawn_future(async move {
self.run().await;
self.run_with_shutdown(task_client).await;
})
}
pub(crate) fn create_and_start(
pub(crate) fn create_and_start_with_shutdown(
reporting_config: StatsReporting,
client_type: String,
client_stats_id: String,
report_tx: InputMessageSender,
task_client: TaskClient,
task_client: nym_task::TaskClient,
) -> ClientStatsSender {
let (controller, sender) = Self::create(
reporting_config,
client_type,
client_stats_id,
report_tx,
task_client,
);
controller.start();
let (controller, sender) =
Self::create(reporting_config, client_type, client_stats_id, report_tx);
controller.start_with_shutdown(task_client);
sender
}
}
@@ -6,7 +6,6 @@ pub(crate) use accessor::{TopologyAccessor, TopologyReadPermit};
use futures::StreamExt;
use log::*;
use nym_sphinx::addressing::nodes::NodeIdentity;
use nym_task::TaskClient;
use nym_topology::NymTopologyError;
use std::time::Duration;
@@ -44,8 +43,6 @@ pub struct TopologyRefresher {
refresh_rate: Duration,
consecutive_failure_count: usize,
task_client: TaskClient,
}
impl TopologyRefresher {
@@ -53,14 +50,12 @@ impl TopologyRefresher {
cfg: TopologyRefresherConfig,
topology_accessor: TopologyAccessor,
topology_provider: Box<dyn TopologyProvider + Send + Sync>,
task_client: TaskClient,
) -> Self {
TopologyRefresher {
topology_provider,
topology_accessor,
refresh_rate: cfg.refresh_rate,
consecutive_failure_count: 0,
task_client,
}
}
@@ -147,7 +142,7 @@ impl TopologyRefresher {
}
}
pub fn start(mut self) {
pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) {
spawn_future(async move {
debug!("Started TopologyRefresher with graceful shutdown support");
@@ -160,17 +155,17 @@ impl TopologyRefresher {
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
while !self.task_client.is_shutdown() {
while !shutdown.is_shutdown() {
tokio::select! {
_ = interval.next() => {
self.try_refresh().await;
},
_ = self.task_client.recv() => {
_ = shutdown.recv() => {
log::trace!("TopologyRefresher: Received shutdown");
},
}
}
self.task_client.recv_timeout().await;
shutdown.recv_timeout().await;
log::debug!("TopologyRefresher: Exiting");
})
}
@@ -58,8 +58,8 @@ impl<T> TransmissionBuffer<T> {
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn lanes(&self) -> Vec<TransmissionLane> {
self.buffer.keys().cloned().collect()
pub(crate) fn num_lanes(&self) -> usize {
self.buffer.keys().count()
}
pub(crate) fn lane_length(&self, lane: &TransmissionLane) -> Option<usize> {
@@ -83,7 +83,6 @@ impl<T> TransmissionBuffer<T> {
}
#[cfg(not(target_arch = "wasm32"))]
#[allow(unused)]
pub(crate) fn total_size_in_bytes(&self) -> usize
where
T: SizedData,
-16
View File
@@ -12,9 +12,6 @@ use std::path::PathBuf;
#[derive(thiserror::Error, Debug)]
pub enum ClientCoreError {
#[error("could not perform the state migration: {0}")]
UnsupportedMigration(String),
#[error("I/O error: {0}")]
IoError(#[from] std::io::Error),
@@ -39,13 +36,6 @@ pub enum ClientCoreError {
#[error("no gateway with id: {0}")]
NoGatewayWithId(String),
#[error("Invalid URL: {0}")]
InvalidUrl(String),
#[cfg(not(target_arch = "wasm32"))]
#[error("resolution failed: {0}")]
ResolutionFailed(#[from] nym_http_api_client::HickoryDnsError),
#[error("no gateways on network")]
NoGatewaysOnNetwork,
@@ -106,9 +96,6 @@ pub enum ClientCoreError {
#[error("timed out while trying to establish gateway connection")]
GatewayConnectionTimeout,
#[error("failed to forward mix messages to gateway")]
GatewayFailedToForwardMessages,
#[error("no ping measurements for the gateway ({identity}) performed")]
NoGatewayMeasurements { identity: String },
@@ -225,9 +212,6 @@ pub enum ClientCoreError {
"fresh registration with gateway {gateway_id} somehow requires an additional key upgrade!"
)]
UnexpectedKeyUpgrade { gateway_id: String },
#[error("failed to derive keys from master key")]
HkdfDerivationError {},
}
/// Set of messages that the client can send to listeners via the task manager
+12 -6
View File
@@ -15,9 +15,6 @@ use std::{sync::Arc, time::Duration};
use tungstenite::Message;
use url::Url;
#[cfg(not(target_arch = "wasm32"))]
use crate::init::websockets::connect_async;
use nym_topology::NodeId;
#[cfg(not(target_arch = "wasm32"))]
use tokio::net::TcpStream;
@@ -26,6 +23,8 @@ use tokio::time::sleep;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::Instant;
#[cfg(not(target_arch = "wasm32"))]
use tokio_tungstenite::connect_async;
#[cfg(not(target_arch = "wasm32"))]
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
#[cfg(target_arch = "wasm32")]
use wasm_utils::websocket::JSWebsocket;
@@ -92,7 +91,6 @@ pub async fn gateways_for_init<R: Rng>(
nym_apis: &[Url],
user_agent: Option<UserAgent>,
minimum_performance: u8,
ignore_epoch_roles: bool,
) -> Result<Vec<RoutingNode>, ClientCoreError> {
let nym_api = nym_apis
.choose(rng)
@@ -114,7 +112,7 @@ pub async fn gateways_for_init<R: Rng>(
// (we don't want instability)
let valid_gateways = gateways
.iter()
.filter(|g| ignore_epoch_roles || !g.supported_roles.mixnode)
.filter(|g| !g.supported_roles.mixnode)
.filter(|g| g.performance.round_to_integer() >= minimum_performance)
.filter_map(|gateway| gateway.try_into().ok())
.collect::<Vec<_>>();
@@ -133,7 +131,7 @@ pub async fn gateways_for_init<R: Rng>(
async fn connect(endpoint: &str) -> Result<WsConn, ClientCoreError> {
match tokio::time::timeout(CONN_TIMEOUT, connect_async(endpoint)).await {
Err(_elapsed) => Err(ClientCoreError::GatewayConnectionTimeout),
Ok(Err(conn_failure)) => Err(conn_failure),
Ok(Err(conn_failure)) => Err(conn_failure.into()),
Ok(Ok((stream, _))) => Ok(stream),
}
}
@@ -335,6 +333,14 @@ pub(super) async fn register_with_gateway(
}
})?;
// this should NEVER happen, if it did, it means the function was misused,
// because for any fresh **registration**, the derived key is always up to date
if auth_response.requires_key_upgrade {
return Err(ClientCoreError::UnexpectedKeyUpgrade {
gateway_id: gateway_id.to_base58_string(),
});
}
Ok(RegistrationResult {
shared_keys: auth_response.initial_shared_key,
authenticated_ephemeral_client: gateway_client,
-2
View File
@@ -26,8 +26,6 @@ use serde::Serialize;
pub mod helpers;
pub mod types;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) mod websockets;
// helpers for error wrapping
+2 -2
View File
@@ -11,7 +11,7 @@ use nym_client_core_gateways_storage::{
};
use nym_crypto::asymmetric::identity;
use nym_gateway_client::client::InitGatewayClient;
use nym_gateway_requests::shared_key::SharedSymmetricKey;
use nym_gateway_requests::shared_key::SharedGatewayKey;
use nym_sphinx::addressing::clients::Recipient;
use nym_topology::node::RoutingNode;
use nym_validator_client::client::IdentityKey;
@@ -96,7 +96,7 @@ impl SelectedGateway {
/// - shared keys derived between ourselves and the node
/// - an authenticated handle of an ephemeral handle created for the purposes of registration
pub struct RegistrationResult {
pub shared_keys: Arc<SharedSymmetricKey>,
pub shared_keys: Arc<SharedGatewayKey>,
pub authenticated_ephemeral_client: InitGatewayClient,
}
-44
View File
@@ -1,44 +0,0 @@
use crate::error::ClientCoreError;
use nym_http_api_client::HickoryDnsResolver;
use tokio::net::TcpStream;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
use tungstenite::handshake::client::Response;
use url::{Host, Url};
use std::net::SocketAddr;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn connect_async(
endpoint: &str,
) -> Result<(WebSocketStream<MaybeTlsStream<TcpStream>>, Response), ClientCoreError> {
let resolver = HickoryDnsResolver::default();
let uri = Url::parse(endpoint).map_err(|_| ClientCoreError::InvalidUrl(endpoint.to_owned()))?;
let port: u16 = uri.port_or_known_default().unwrap_or(443);
let host = uri
.host()
.ok_or(ClientCoreError::InvalidUrl(endpoint.to_owned()))?;
// Get address for tcp connection, if a domain is provided use our preferred resolver rather than
// the default std resolve
let sock_addrs: Vec<SocketAddr> = match host {
Host::Ipv4(addr) => vec![SocketAddr::new(addr.into(), port)],
Host::Ipv6(addr) => vec![SocketAddr::new(addr.into(), port)],
Host::Domain(domain) => {
// Do a DNS lookup for the domain using our custom DNS resolver
resolver
.resolve_str(domain)
.await?
.into_iter()
.map(|a| SocketAddr::new(a, port))
.collect()
}
};
let stream = TcpStream::connect(&sock_addrs[..]).await?;
tokio_tungstenite::client_async_tls(endpoint, stream)
.await
.map_err(Into::into)
}
+45
View File
@@ -33,3 +33,48 @@ where
{
tokio::spawn(future);
}
#[derive(Clone, Default, Debug)]
pub struct ForgetMe {
client: bool,
stats: bool,
}
impl ForgetMe {
pub fn new_all() -> Self {
Self {
client: true,
stats: true,
}
}
pub fn new_client() -> Self {
Self {
client: true,
stats: false,
}
}
pub fn new_stats() -> Self {
Self {
client: false,
stats: true,
}
}
pub fn new(client: bool, stats: bool) -> Self {
Self { client, stats }
}
pub fn any(&self) -> bool {
self.client || self.stats
}
pub fn client(&self) -> bool {
self.client
}
pub fn stats(&self) -> bool {
self.stats
}
}
@@ -27,7 +27,6 @@ nym-credential-storage = { path = "../../credential-storage" }
nym-credentials-interface = { path = "../../credentials-interface" }
nym-crypto = { path = "../../crypto" }
nym-gateway-requests = { path = "../../gateway-requests" }
nym-http-api-client = { path = "../../http-api-client" }
nym-network-defaults = { path = "../../network-defaults" }
nym-sphinx = { path = "../../nymsphinx" }
nym-statistics-common = { path = "../../statistics" }
@@ -20,8 +20,9 @@ use nym_credentials_interface::TicketType;
use nym_crypto::asymmetric::identity;
use nym_gateway_requests::registration::handshake::client_handshake;
use nym_gateway_requests::{
BinaryRequest, ClientControlRequest, ClientRequest, GatewayProtocolVersionExt, ServerResponse,
SharedSymmetricKey, CREDENTIAL_UPDATE_V2_PROTOCOL_VERSION, CURRENT_PROTOCOL_VERSION,
BinaryRequest, ClientControlRequest, ClientRequest, SensitiveServerResponse, ServerResponse,
SharedGatewayKey, SharedSymmetricKey, AES_GCM_SIV_PROTOCOL_VERSION,
CREDENTIAL_UPDATE_V2_PROTOCOL_VERSION, CURRENT_PROTOCOL_VERSION,
};
use nym_sphinx::forwarding::packet::MixPacket;
use nym_statistics_common::clients::connection::ConnectionStatsEvent;
@@ -39,6 +40,8 @@ use url::Url;
use std::os::fd::RawFd;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep;
#[cfg(not(target_arch = "wasm32"))]
use tokio_tungstenite::connect_async;
#[cfg(not(unix))]
use std::os::raw::c_int as RawFd;
@@ -46,14 +49,10 @@ use std::os::raw::c_int as RawFd;
use wasm_utils::websocket::JSWebsocket;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::sleep;
use zeroize::Zeroizing;
pub mod config;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) mod websockets;
#[cfg(not(target_arch = "wasm32"))]
use websockets::connect_async;
pub struct GatewayConfig {
pub gateway_identity: identity::PublicKey,
@@ -80,7 +79,8 @@ impl GatewayConfig {
#[must_use]
#[derive(Debug)]
pub struct AuthenticationResponse {
pub initial_shared_key: Arc<SharedSymmetricKey>,
pub initial_shared_key: Arc<SharedGatewayKey>,
pub requires_key_upgrade: bool,
}
// TODO: this should be refactored into a state machine that keeps track of its authentication state
@@ -92,7 +92,7 @@ pub struct GatewayClient<C, St = EphemeralCredentialStorage> {
gateway_address: String,
gateway_identity: identity::PublicKey,
local_identity: Arc<identity::KeyPair>,
shared_key: Option<Arc<SharedSymmetricKey>>,
shared_key: Option<Arc<SharedGatewayKey>>,
connection: SocketState,
packet_router: PacketRouter,
bandwidth_controller: Option<BandwidthController<C, St>>,
@@ -116,7 +116,7 @@ impl<C, St> GatewayClient<C, St> {
gateway_config: GatewayConfig,
local_identity: Arc<identity::KeyPair>,
// TODO: make it mandatory. if you don't want to pass it, use `new_init`
shared_key: Option<Arc<SharedSymmetricKey>>,
shared_key: Option<Arc<SharedGatewayKey>>,
packet_router: PacketRouter,
bandwidth_controller: Option<BandwidthController<C, St>>,
stats_reporter: ClientStatsSender,
@@ -146,7 +146,7 @@ impl<C, St> GatewayClient<C, St> {
self.gateway_identity
}
pub fn shared_key(&self) -> Option<Arc<SharedSymmetricKey>> {
pub fn shared_key(&self) -> Option<Arc<SharedGatewayKey>> {
self.shared_key.clone()
}
@@ -201,15 +201,23 @@ impl<C, St> GatewayClient<C, St> {
"Attemting to establish connection to gateway at: {}",
self.gateway_address
);
let (ws_stream, _) = connect_async(
&self.gateway_address,
#[cfg(unix)]
self.connection_fd_callback.clone(),
)
.await?;
let ws_stream = match connect_async(&self.gateway_address).await {
Ok((ws_stream, _)) => ws_stream,
Err(error) => {
return Err(GatewayClientError::NetworkConnectionFailed {
address: self.gateway_address.clone(),
source: error,
})
}
};
self.connection = SocketState::Available(Box::new(ws_stream));
#[cfg(unix)]
if let (Some(callback), Some(fd)) = (self.connection_fd_callback.as_ref(), self.ws_fd()) {
callback.as_ref()(fd);
}
Ok(())
}
@@ -263,19 +271,6 @@ impl<C, St> GatewayClient<C, St> {
}
}
pub async fn send_client_request(
&mut self,
message: ClientRequest,
) -> Result<(), GatewayClientError> {
if let Some(shared_key) = self.shared_key() {
let encrypted = message.encrypt(&shared_key)?;
Box::pin(self.send_websocket_message(encrypted)).await?;
Ok(())
} else {
Err(GatewayClientError::ConnectionInInvalidState)
}
}
async fn read_control_response(&mut self) -> Result<ServerResponse, GatewayClientError> {
// we use the fact that all request responses are Message::Text and only pushed
// sphinx packets are Message::Binary
@@ -407,39 +402,49 @@ impl<C, St> GatewayClient<C, St> {
}
}
fn check_gateway_protocol(&self, gateway_protocol: u8) -> Result<(), GatewayClientError> {
fn check_gateway_protocol(
&self,
gateway_protocol: Option<u8>,
) -> Result<(), GatewayClientError> {
debug!("gateway protocol: {gateway_protocol:?}, ours: {CURRENT_PROTOCOL_VERSION}");
// client should reject any gateways that do not indicate they support auth v2 or aes256gcm-siv
if !gateway_protocol.supports_authenticate_v2()
|| !gateway_protocol.supports_aes256_gcm_siv()
{
return Err(GatewayClientError::IncompatibleProtocol {
gateway: gateway_protocol,
current: CURRENT_PROTOCOL_VERSION,
});
}
// right now there are no failure cases here, but this might change in the future
match gateway_protocol {
None => {
warn!("the gateway we're connected to has not specified its protocol version. It's probably running version < 1.1.X, but that's still fine for now. It will become a hard error in 1.2.0");
// note: in +1.2.0 we will have to return a hard error here
Ok(())
}
Some(v) if v > CURRENT_PROTOCOL_VERSION => {
let err = GatewayClientError::IncompatibleProtocol {
gateway: Some(v),
current: CURRENT_PROTOCOL_VERSION,
};
error!("{err}");
Err(err)
}
// we can't handle gateways with higher protocol than ours
if gateway_protocol <= CURRENT_PROTOCOL_VERSION {
debug!("the gateway is using exactly the same (or older) protocol version as we are. We're good to continue!");
Ok(())
} else {
let err = GatewayClientError::IncompatibleProtocol {
gateway: gateway_protocol,
current: CURRENT_PROTOCOL_VERSION,
};
error!("{err}");
Err(err)
Some(_) => {
debug!("the gateway is using exactly the same (or older) protocol version as we are. We're good to continue!");
Ok(())
}
}
}
async fn register(&mut self) -> Result<(), GatewayClientError> {
async fn register(
&mut self,
derive_aes256_gcm_siv_key: bool,
) -> Result<(), GatewayClientError> {
if !self.connection.is_established() {
return Err(GatewayClientError::ConnectionNotEstablished);
}
debug_assert!(self.connection.is_available());
log::debug!(
"registering with gateway. using legacy key derivation: {}",
!derive_aes256_gcm_siv_key
);
// it's fine to instantiate it here as it's only used once (during authentication or registration)
// and putting it into the GatewayClient struct would be a hassle
let mut rng = OsRng;
@@ -451,6 +456,7 @@ impl<C, St> GatewayClient<C, St> {
self.local_identity.as_ref(),
self.gateway_identity,
self.cfg.bandwidth.require_tickets,
derive_aes256_gcm_siv_key,
#[cfg(not(target_arch = "wasm32"))]
self.task_client.clone(),
)
@@ -478,15 +484,99 @@ impl<C, St> GatewayClient<C, St> {
}
// populate the negotiated protocol for future uses
self.negotiated_protocol = Some(gateway_protocol);
self.negotiated_protocol = gateway_protocol;
Ok(())
}
async fn send_authenticate_request_and_handle_response(
pub async fn upgrade_key_authenticated(
&mut self,
msg: ClientControlRequest,
) -> Result<(), GatewayClientError> {
) -> Result<Zeroizing<SharedSymmetricKey>, GatewayClientError> {
info!("*** STARTING AES128CTR-HMAC KEY UPGRADE INTO AES256GCM-SIV***");
if !self.connection.is_established() {
return Err(GatewayClientError::ConnectionNotEstablished);
}
if !self.authenticated {
return Err(GatewayClientError::NotAuthenticated);
}
let Some(shared_key) = self.shared_key.as_ref() else {
return Err(GatewayClientError::NoSharedKeyAvailable);
};
if !shared_key.is_legacy() {
return Err(GatewayClientError::KeyAlreadyUpgraded);
}
// make sure we have the only reference, so we could safely swap it
if Arc::strong_count(shared_key) != 1 {
return Err(GatewayClientError::KeyAlreadyInUse);
}
assert!(shared_key.is_legacy());
let legacy_key = shared_key.unwrap_legacy();
let (updated_key, hkdf_salt) = legacy_key.upgrade();
let derived_key_digest = updated_key.digest();
let upgrade_request = ClientRequest::UpgradeKey {
hkdf_salt,
derived_key_digest,
}
.encrypt(legacy_key)?;
info!("sending upgrade request and awaiting the acknowledgement back");
let (ciphertext, nonce) = match self.send_websocket_message(upgrade_request).await? {
ServerResponse::EncryptedResponse { ciphertext, nonce } => (ciphertext, nonce),
ServerResponse::Error { message } => {
return Err(GatewayClientError::GatewayError(message))
}
other => return Err(GatewayClientError::UnexpectedResponse { name: other.name() }),
};
// attempt to decrypt it using NEW key
let Ok(response) = SensitiveServerResponse::decrypt(&ciphertext, &nonce, &updated_key)
else {
return Err(GatewayClientError::FatalKeyUpgradeFailure);
};
match response {
SensitiveServerResponse::KeyUpgradeAck { .. } => {
info!("received key upgrade acknowledgement")
}
_ => return Err(GatewayClientError::FatalKeyUpgradeFailure),
}
// perform in memory swap and make a copy for updating storage
let zeroizing_updated_key = updated_key.zeroizing_clone();
self.shared_key = Some(Arc::new(updated_key.into()));
Ok(zeroizing_updated_key)
}
async fn authenticate(&mut self) -> Result<(), GatewayClientError> {
let Some(shared_key) = self.shared_key.as_ref() else {
return Err(GatewayClientError::NoSharedKeyAvailable);
};
if !self.connection.is_established() {
return Err(GatewayClientError::ConnectionNotEstablished);
}
debug!("authenticating with gateway");
let self_address = self
.local_identity
.as_ref()
.public_key()
.derive_destination_address();
let msg = ClientControlRequest::new_authenticate(
self_address,
shared_key,
self.cfg.bandwidth.require_tickets,
)?;
match self.send_websocket_message(msg).await? {
ServerResponse::Authenticate {
protocol_version,
@@ -497,7 +587,7 @@ impl<C, St> GatewayClient<C, St> {
self.authenticated = status;
self.bandwidth.update_and_maybe_log(bandwidth_remaining);
self.negotiated_protocol = Some(protocol_version);
self.negotiated_protocol = protocol_version;
log::debug!("authenticated: {status}, bandwidth remaining: {bandwidth_remaining}");
self.task_client.send_status_msg(Box::new(
@@ -510,26 +600,6 @@ impl<C, St> GatewayClient<C, St> {
}
}
async fn authenticate_v2(&mut self) -> Result<(), GatewayClientError> {
debug!("using v2 authentication");
let Some(shared_key) = self.shared_key.as_ref() else {
return Err(GatewayClientError::NoSharedKeyAvailable);
};
let msg = ClientControlRequest::new_authenticate_v2(shared_key, &self.local_identity)?;
self.send_authenticate_request_and_handle_response(msg)
.await
}
async fn authenticate(&mut self) -> Result<(), GatewayClientError> {
if !self.connection.is_established() {
return Err(GatewayClientError::ConnectionNotEstablished);
}
debug!("authenticating with gateway");
self.authenticate_v2().await
}
/// Helper method to either call register or authenticate based on self.shared_key value
#[instrument(skip_all,
fields(
@@ -545,26 +615,18 @@ impl<C, St> GatewayClient<C, St> {
}
// 1. check gateway's protocol version
// if we failed to get this request resolved, it means the gateway is on an old version
// that definitely does not support auth v2 or aes256gcm, so we bail
let gw_protocol = self.get_gateway_protocol().await?;
let supports_aes_gcm_siv = gw_protocol.supports_aes256_gcm_siv();
let supports_auth_v2 = gw_protocol.supports_authenticate_v2();
let supports_aes_gcm_siv = match self.get_gateway_protocol().await {
Ok(protocol) => protocol >= AES_GCM_SIV_PROTOCOL_VERSION,
Err(_) => {
// if we failed to send the request, it means the gateway is running the old binary,
// so it has reset our connection - we have to reconnect
self.establish_connection().await?;
false
}
};
if !supports_aes_gcm_siv {
error!("this gateway is on an old version that doesn't support AES256-GCM-SIV");
}
if !supports_aes_gcm_siv {
error!("this gateway is on an old version that doesn't support authentication v2");
}
if !supports_auth_v2 || !supports_aes_gcm_siv {
// we can't continue
return Err(GatewayClientError::IncompatibleProtocol {
gateway: gw_protocol,
current: CURRENT_PROTOCOL_VERSION,
});
warn!("this gateway is on an old version that doesn't support AES256-GCM-SIV");
}
if self.authenticated {
@@ -572,6 +634,7 @@ impl<C, St> GatewayClient<C, St> {
return if let Some(shared_key) = &self.shared_key {
Ok(AuthenticationResponse {
initial_shared_key: Arc::clone(shared_key),
requires_key_upgrade: shared_key.is_legacy() && supports_aes_gcm_siv,
})
} else {
Err(GatewayClientError::AuthenticationFailureWithPreexistingSharedKey)
@@ -585,14 +648,17 @@ impl<C, St> GatewayClient<C, St> {
// if we are authenticated it means we MUST have an associated shared_key
let shared_key = self.shared_key.as_ref().unwrap();
let requires_key_upgrade = shared_key.is_legacy() && supports_aes_gcm_siv;
Ok(AuthenticationResponse {
initial_shared_key: Arc::clone(shared_key),
requires_key_upgrade,
})
} else {
Err(GatewayClientError::AuthenticationFailure)
}
} else {
self.register().await?;
self.register(supports_aes_gcm_siv).await?;
// if registration didn't return an error, we MUST have an associated shared key
let shared_key = self.shared_key.as_ref().unwrap();
@@ -601,6 +667,7 @@ impl<C, St> GatewayClient<C, St> {
// so no upgrades are required
Ok(AuthenticationResponse {
initial_shared_key: Arc::clone(shared_key),
requires_key_upgrade: false,
})
}
}
@@ -978,7 +1045,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
connection: SocketState::NotConnected,
packet_router,
bandwidth_controller: None,
stats_reporter: ClientStatsSender::new(None, task_client.clone()),
stats_reporter: ClientStatsSender::new(None),
negotiated_protocol: None,
#[cfg(unix)]
connection_fd_callback: None,
@@ -1,88 +0,0 @@
use crate::error::GatewayClientError;
use nym_http_api_client::HickoryDnsResolver;
#[cfg(unix)]
use std::{
os::fd::{AsRawFd, RawFd},
sync::Arc,
};
use tokio::net::TcpStream;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
use tungstenite::handshake::client::Response;
use url::{Host, Url};
use std::net::SocketAddr;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn connect_async(
endpoint: &str,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
) -> Result<(WebSocketStream<MaybeTlsStream<TcpStream>>, Response), GatewayClientError> {
use tokio::net::TcpSocket;
let resolver = HickoryDnsResolver::default();
let uri =
Url::parse(endpoint).map_err(|_| GatewayClientError::InvalidUrl(endpoint.to_owned()))?;
let port: u16 = uri.port_or_known_default().unwrap_or(443);
let host = uri
.host()
.ok_or(GatewayClientError::InvalidUrl(endpoint.to_owned()))?;
// Get address for tcp connection, if a domain is provided use our preferred resolver rather than
// the default std resolve
let sock_addrs: Vec<SocketAddr> = match host {
Host::Ipv4(addr) => vec![SocketAddr::new(addr.into(), port)],
Host::Ipv6(addr) => vec![SocketAddr::new(addr.into(), port)],
Host::Domain(domain) => {
// Do a DNS lookup for the domain using our custom DNS resolver
resolver
.resolve_str(domain)
.await?
.into_iter()
.map(|a| SocketAddr::new(a, port))
.collect()
}
};
let mut stream = Err(GatewayClientError::NoEndpointForConnection {
address: endpoint.to_owned(),
});
for sock_addr in sock_addrs {
let socket = if sock_addr.is_ipv4() {
TcpSocket::new_v4()
} else {
TcpSocket::new_v6()
}
.map_err(|err| GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: err.into(),
})?;
#[cfg(unix)]
if let Some(callback) = connection_fd_callback.as_ref() {
callback.as_ref()(socket.as_raw_fd());
}
match socket.connect(sock_addr).await {
Ok(s) => {
stream = Ok(s);
break;
}
Err(err) => {
stream = Err(GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: err.into(),
});
continue;
}
}
}
tokio_tungstenite::client_async_tls(endpoint, stream?)
.await
.map_err(|error| GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: error,
})
}
@@ -43,15 +43,8 @@ pub enum GatewayClientError {
#[error("connection failed: {address}: {source}")]
NetworkConnectionFailed { address: String, source: WsError },
#[error("no socket address for endpoint: {address}")]
NoEndpointForConnection { address: String },
#[error("Invalid URL: {0}")]
InvalidUrl(String),
#[cfg(not(target_arch = "wasm32"))]
#[error("resolution failed: {0}")]
ResolutionFailed(#[from] nym_http_api_client::HickoryDnsError),
InvalidURL(String),
#[error("No shared key was provided or obtained")]
NoSharedKeyAvailable,
@@ -114,7 +107,7 @@ pub enum GatewayClientError {
MixnetMsgSenderFailedToSend,
#[error("Attempted to negotiate connection with gateway using incompatible protocol version. Ours is {current} and the gateway reports {gateway:?}")]
IncompatibleProtocol { gateway: u8, current: u8 },
IncompatibleProtocol { gateway: Option<u8>, current: u8 },
#[error(
"The packet router hasn't been set - are you sure you started up the client correctly?"
+4 -2
View File
@@ -7,7 +7,9 @@ use tracing::{error, warn};
use tungstenite::{protocol::Message, Error as WsError};
pub use client::{config::GatewayClientConfig, GatewayClient, GatewayConfig};
pub use nym_gateway_requests::shared_key::SharedSymmetricKey;
pub use nym_gateway_requests::shared_key::{
LegacySharedKeys, SharedGatewayKey, SharedSymmetricKey,
};
pub use packet_router::{
AcknowledgementReceiver, AcknowledgementSender, MixnetMessageReceiver, MixnetMessageSender,
PacketRouter,
@@ -45,7 +47,7 @@ pub(crate) fn cleanup_socket_messages(
pub(crate) fn try_decrypt_binary_message(
bin_msg: Vec<u8>,
shared_keys: &SharedSymmetricKey,
shared_keys: &SharedGatewayKey,
) -> Option<Vec<u8>> {
match BinaryResponse::try_from_encrypted_tagged_bytes(bin_msg, shared_keys) {
Ok(bin_response) => match bin_response {
@@ -9,7 +9,7 @@ use crate::{cleanup_socket_messages, try_decrypt_binary_message};
use futures::channel::oneshot;
use futures::stream::{SplitSink, SplitStream};
use futures::{SinkExt, StreamExt};
use nym_gateway_requests::shared_key::SharedSymmetricKey;
use nym_gateway_requests::shared_key::SharedGatewayKey;
use nym_gateway_requests::{ServerResponse, SimpleGatewayRequestsError};
use nym_task::TaskClient;
use si_scale::helpers::bibytes2;
@@ -63,7 +63,7 @@ pub(crate) struct PartiallyDelegatedHandle {
struct PartiallyDelegatedRouter {
packet_router: PacketRouter,
shared_key: Arc<SharedSymmetricKey>,
shared_key: Arc<SharedGatewayKey>,
client_bandwidth: ClientBandwidth,
stream_return: SplitStreamSender,
@@ -73,7 +73,7 @@ struct PartiallyDelegatedRouter {
impl PartiallyDelegatedRouter {
fn new(
packet_router: PacketRouter,
shared_key: Arc<SharedSymmetricKey>,
shared_key: Arc<SharedGatewayKey>,
client_bandwidth: ClientBandwidth,
stream_return: SplitStreamSender,
stream_return_requester: oneshot::Receiver<()>,
@@ -253,7 +253,7 @@ impl PartiallyDelegatedHandle {
pub(crate) fn split_and_listen_for_mixnet_messages(
conn: WsConn,
packet_router: PacketRouter,
shared_key: Arc<SharedSymmetricKey>,
shared_key: Arc<SharedGatewayKey>,
client_bandwidth: ClientBandwidth,
shutdown: TaskClient,
) -> Self {
@@ -11,7 +11,8 @@ use crate::{
use nym_api_requests::ecash::models::{
AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
BatchRedeemTicketsBody, EcashBatchTicketRedemptionResponse, EcashTicketVerificationResponse,
IssuedTicketbooksChallengeResponse, IssuedTicketbooksForResponse, VerifyEcashTicketBody,
IssuedTicketbooksChallengeResponse, IssuedTicketbooksForResponse, SpentCredentialsResponse,
VerifyEcashTicketBody,
};
use nym_api_requests::ecash::{
BlindSignRequestBody, BlindedSignatureResponse, PartialCoinIndicesSignatureResponse,
@@ -23,12 +24,11 @@ use nym_api_requests::models::{
NymNodeDescription, RewardEstimationResponse, StakeSaturationResponse,
};
use nym_api_requests::models::{LegacyDescribedGateway, MixNodeBondAnnotated};
use nym_api_requests::nym_nodes::{NodesByAddressesResponse, SkimmedNode};
use nym_api_requests::nym_nodes::SkimmedNode;
use nym_coconut_dkg_common::types::EpochId;
use nym_ecash_contract_common::deposit::DepositId;
use nym_http_api_client::UserAgent;
use nym_network_defaults::NymNetworkDetails;
use std::net::IpAddr;
use time::Date;
use url::Url;
@@ -647,6 +647,13 @@ impl NymApiClient {
.await?)
}
#[deprecated]
pub async fn spent_credentials_filter(
&self,
) -> Result<SpentCredentialsResponse, ValidatorClientError> {
Ok(self.nym_api.double_spending_filter_v1().await?)
}
pub async fn partial_expiration_date_signatures(
&self,
expiration_date: Option<Date>,
@@ -711,11 +718,4 @@ impl NymApiClient {
.issued_ticketbooks_challenge(expiration_date, deposits)
.await?)
}
pub async fn nodes_by_addresses(
&self,
addresses: Vec<IpAddr>,
) -> Result<NodesByAddressesResponse, ValidatorClientError> {
Ok(self.nym_api.nodes_by_addresses(addresses).await?)
}
}
@@ -65,12 +65,6 @@ pub enum EcashApiError {
#[from]
source: cosmrs::ErrorReport,
},
#[error("nym api error")]
NymApi {
#[from]
source: crate::ValidatorClientError,
},
}
impl TryFrom<ContractVKShare> for EcashApiClient {
@@ -15,9 +15,7 @@ use nym_api_requests::models::{
AnnotationResponse, ApiHealthResponse, LegacyDescribedMixNode, NodePerformanceResponse,
NodeRefreshBody, NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
};
use nym_api_requests::nym_nodes::{
NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponse,
};
use nym_api_requests::nym_nodes::PaginatedCachedNodesResponse;
use nym_api_requests::pagination::PaginatedResponse;
pub use nym_api_requests::{
ecash::{
@@ -33,7 +31,6 @@ pub use nym_api_requests::{
StakeSaturationResponse, UptimeResponse,
},
nym_nodes::{CachedNodesResponse, SkimmedNode},
NymNetworkDetailsResponse,
};
pub use nym_coconut_dkg_common::types::EpochId;
use nym_contracts_common::IdentityKey;
@@ -42,7 +39,6 @@ pub use nym_http_api_client::Client;
use nym_http_api_client::{ApiClient, NO_PARAMS};
use nym_mixnet_contract_common::mixnode::MixNodeDetails;
use nym_mixnet_contract_common::{GatewayBond, IdentityKeyRef, NodeId, NymNodeDetails};
use std::net::IpAddr;
use time::format_description::BorrowedFormatItem;
use time::Date;
use tracing::instrument;
@@ -853,6 +849,20 @@ pub trait NymApiClientExt: ApiClient {
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn double_spending_filter_v1(&self) -> Result<SpentCredentialsResponse, NymAPIError> {
self.get_json(
&[
routes::API_VERSION,
routes::ECASH_ROUTES,
routes::DOUBLE_SPENDING_FILTER_V1,
],
NO_PARAMS,
)
.await
}
#[instrument(level = "debug", skip(self))]
async fn partial_expiration_date_signatures(
&self,
@@ -1017,32 +1027,6 @@ pub trait NymApiClientExt: ApiClient {
)
.await
}
async fn nodes_by_addresses(
&self,
addresses: Vec<IpAddr>,
) -> Result<NodesByAddressesResponse, NymAPIError> {
self.post_json(
&[
routes::API_VERSION,
"unstable",
routes::NYM_NODES_ROUTES,
routes::nym_nodes::BY_ADDRESSES,
],
NO_PARAMS,
&NodesByAddressesRequestBody { addresses },
)
.await
}
#[instrument(level = "debug", skip(self))]
async fn get_network_details(&self) -> Result<NymNetworkDetailsResponse, NymAPIError> {
self.get_json(
&[routes::API_VERSION, routes::NETWORK, routes::DETAILS],
NO_PARAMS,
)
.await
}
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@@ -13,6 +13,8 @@ pub const DETAILED: &str = "detailed";
pub const DETAILED_UNFILTERED: &str = "detailed-unfiltered";
pub const ACTIVE: &str = "active";
pub const REWARDED: &str = "rewarded";
pub const DOUBLE_SPENDING_FILTER_V1: &str = "double-spending-filter-v1";
pub const ECASH_ROUTES: &str = "ecash";
pub use ecash::*;
@@ -43,7 +45,6 @@ pub mod nym_nodes {
pub const NYM_NODES_BONDED: &str = "bonded";
pub const NYM_NODES_REWARDED_SET: &str = "rewarded-set";
pub const NYM_NODES_REFRESH_DESCRIBED: &str = "refresh-described";
pub const BY_ADDRESSES: &str = "by-addresses";
}
pub const STATUS_ROUTES: &str = "status";
@@ -66,8 +67,6 @@ pub const STAKE_SATURATION: &str = "stake-saturation";
pub const INCLUSION_CHANCE: &str = "inclusion-probability";
pub const SUBMIT_GATEWAY: &str = "submit-gateway-monitoring-results";
pub const SUBMIT_NODE: &str = "submit-node-monitoring-results";
pub const PERFORMANCE: &str = "performance";
pub const SERVICE_PROVIDERS: &str = "services";
pub const DETAILS: &str = "details";
pub const NETWORK: &str = "network";
@@ -28,6 +28,7 @@ use nym_network_defaults::{ChainDetails, NymNetworkDetails};
use serde::{de::DeserializeOwned, Serialize};
use std::fmt::Debug;
use std::time::SystemTime;
use tendermint_rpc::endpoint::block::Response as BlockResponse;
use tendermint_rpc::endpoint::*;
use tendermint_rpc::{Error as TendermintRpcError, Order};
use url::Url;
@@ -62,7 +63,6 @@ pub use cw3;
pub use cw4;
pub use cw_controllers;
pub use fee::{gas_price::GasPrice, GasAdjustable, GasAdjustment};
pub use tendermint_rpc::endpoint::block::Response as BlockResponse;
pub use tendermint_rpc::{
endpoint::{tx::Response as TxResponse, validators::Response as ValidatorResponse},
query::Query,
+9
View File
@@ -25,6 +25,15 @@ pub fn in6addr_any_init() -> IpAddr {
IpAddr::V6(Ipv6Addr::UNSPECIFIED)
}
/// Helper for providing binding warnings if node tries to bind to any of those
pub const SPECIAL_ADDRESSES: &[IpAddr] = &[
IpAddr::V4(Ipv4Addr::LOCALHOST),
IpAddr::V4(Ipv4Addr::UNSPECIFIED),
IpAddr::V4(Ipv4Addr::BROADCAST),
IpAddr::V6(Ipv6Addr::LOCALHOST),
IpAddr::V6(Ipv6Addr::UNSPECIFIED),
];
// TODO: is it really part of 'Config'?
pub trait OptionalSet {
/// If the value is available (i.e. `Some`), the provided closure is applied.
@@ -8,81 +8,81 @@ use thiserror::Error;
#[derive(Error, Debug, PartialEq)]
pub enum VestingContractError {
#[error("VESTING ({l}): {0}", l = line!())]
#[error("VESTING ({}): {0}", line!())]
Std(#[from] StdError),
#[error("VESTING: {0}")]
OverflowError(#[from] OverflowError),
#[error("VESTING ({l}): Account does not exist - {0}", l = line!())]
#[error("VESTING ({}): Account does not exist - {0}", line!())]
NoAccountForAddress(String),
#[error("VESTING ({l}): Only admin can perform this action, {0} is not admin", l = line!())]
#[error("VESTING ({}): Only admin can perform this action, {0} is not admin", line!())]
NotAdmin(String),
#[error("VESTING ({l}): Balance not found for existing account ({0}), this is a bug", l = line!())]
#[error("VESTING ({}): Balance not found for existing account ({0}), this is a bug", line!())]
NoBalanceForAddress(String),
#[error("VESTING ({l}): Insufficient balance for address {0} -> {1}", l = line!())]
#[error("VESTING ({}): Insufficient balance for address {0} -> {1}", line!())]
InsufficientBalance(String, u128),
#[error("VESTING ({l}): Insufficient spendable balance for address {0} -> {1}", l = line!())]
#[error("VESTING ({}): Insufficient spendable balance for address {0} -> {1}", line!())]
InsufficientSpendable(String, u128),
#[error(
"VESTING ({l}):Only delegation owner can perform delegation actions, {0} is not the delegation owner"
, l = line!())]
"VESTING ({}):Only delegation owner can perform delegation actions, {0} is not the delegation owner"
, line!())]
NotDelegate(String),
#[error("VESTING ({l}): Total vesting amount is inprobably low -> {0}, this is likely an error", l = line!())]
#[error("VESTING ({}): Total vesting amount is inprobably low -> {0}, this is likely an error", line!())]
ImprobableVestingAmount(u128),
#[error("VESTING ({l}): Address {0} has already bonded a node", l = line!())]
#[error("VESTING ({}): Address {0} has already bonded a node", line!())]
AlreadyBonded(String),
#[error("VESTING ({l}): Received empty funds vector", l = line!())]
#[error("VESTING ({}): Received empty funds vector", line!())]
EmptyFunds,
#[error("VESTING ({l}): Received wrong denom: {0}, expected {1}", l = line!())]
#[error("VESTING ({}): Received wrong denom: {0}, expected {1}", line!())]
WrongDenom(String, String),
#[error("VESTING ({l}): Received multiple denoms, expected 1", l = line!())]
#[error("VESTING ({}): Received multiple denoms, expected 1", line!())]
MultipleDenoms,
#[error("VESTING ({l}): No delegations found for account {0}, mix_identity {1}", l = line!())]
#[error("VESTING ({}): No delegations found for account {0}, mix_identity {1}", line!())]
NoSuchDelegation(Addr, NodeId),
#[error("VESTING ({l}): Only mixnet contract can perform this operation, got {0}", l = line!())]
#[error("VESTING ({}): Only mixnet contract can perform this operation, got {0}", line!())]
NotMixnetContract(Addr),
#[error("VESTING ({l}): Calculation underflowed", l = line!())]
#[error("VESTING ({}): Calculation underflowed", line!())]
Underflow,
#[error("VESTING ({l}): No bond found for account {0}", l = line!())]
#[error("VESTING ({}): No bond found for account {0}", line!())]
NoBondFound(String),
#[error("VESTING: Attempted to reduce mixnode bond pledge below zero! The current pledge is {current} and we attempted to reduce it by {decrease_by}.")]
InvalidBondPledgeReduction { current: Coin, decrease_by: Coin },
#[error("VESTING ({l}): Action can only be executed by account owner -> {0}", l = line!())]
#[error("VESTING ({}): Action can only be executed by account owner -> {0}", line!())]
NotOwner(String),
#[error("VESTING ({l}): Invalid address: {0}", l = line!())]
#[error("VESTING ({}): Invalid address: {0}", line!())]
InvalidAddress(String),
#[error("VESTING ({l}): Account already exists: {0}", l = line!())]
#[error("VESTING ({}): Account already exists: {0}", line!())]
AccountAlreadyExists(String),
#[error("VESTING ({l}): Staking account already exists: {0}", l = line!())]
#[error("VESTING ({}): Staking account already exists: {0}", line!())]
StakingAccountAlreadyExists(String),
#[error("VESTING ({l}): Too few coins sent for vesting account creation, sent {sent}, need at least {need}", l = line!())]
#[error("VESTING ({}): Too few coins sent for vesting account creation, sent {sent}, need at least {need}", line!())]
MinVestingFunds { sent: u128, need: u128 },
#[error("VESTING ({l}): Maximum amount of locked coins has already been pledged: {current}, cap is {cap}", l = line!())]
#[error("VESTING ({}): Maximum amount of locked coins has already been pledged: {current}, cap is {cap}", line!())]
LockedPledgeCapReached { current: Uint128, cap: Uint128 },
#[error("VESTING: ({l}: Account owned by {owner} has unpopulated vesting periods!", l = line!())]
#[error("VESTING: ({}: Account owned by {owner} has unpopulated vesting periods!", line!())]
UnpopulatedVestingPeriods { owner: Addr },
#[error("VESTING: Vesting account associated with {0} already exists, only addresses with not existing vesting accounts can be added as staking addresses")]
@@ -26,6 +26,7 @@ nym-api-requests = { path = "../../nym-api/nym-api-requests" }
nym-credentials = { path = "../credentials" }
nym-credentials-interface = { path = "../credentials-interface" }
nym-ecash-contract-common = { path = "../cosmwasm-smart-contracts/ecash-contract" }
nym-ecash-double-spending = { path = "../ecash-double-spending" }
nym-gateway-requests = { path = "../gateway-requests" }
nym-gateway-storage = { path = "../gateway-storage" }
nym-task = { path = "../task" }
@@ -13,7 +13,6 @@ use nym_api_requests::constants::MIN_BATCH_REDEMPTION_DELAY;
use nym_api_requests::ecash::models::{BatchRedeemTicketsBody, VerifyEcashTicketBody};
use nym_credentials_interface::Bandwidth;
use nym_credentials_interface::{ClientTicket, TicketType};
use nym_validator_client::coconut::EcashApiError;
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::nyxd::contract_traits::{
EcashSigningClient, MultisigQueryClient, MultisigSigningClient, PagedMultisigQueryClient,
@@ -353,9 +352,7 @@ impl CredentialHandler {
}
Err(err) => {
error!("failed to send ticket {ticket_id} for verification to ecash signer '{client}': {err}. if we don't reach quorum, we'll retry later");
Err(EcashTicketError::ApiFailure(EcashApiError::NymApi {
source: err,
}))
Ok(false)
}
}
}
+3 -6
View File
@@ -24,7 +24,6 @@ ed25519-dalek = { workspace = true, features = ["rand_core"], optional = true }
rand = { workspace = true, optional = true }
serde_bytes = { workspace = true, optional = true }
serde = { workspace = true, features = ["derive"], optional = true }
sha2 = { workspace = true, optional = true }
subtle-encoding = { workspace = true, features = ["bech32-preview"] }
thiserror = { workspace = true }
zeroize = { workspace = true, optional = true, features = ["zeroize_derive"] }
@@ -37,13 +36,11 @@ nym-pemstore = { path = "../../common/pemstore", version = "0.3.0" }
rand_chacha = { workspace = true }
[features]
default = []
default = ["sphinx"]
aead = ["dep:aead", "aead/std", "aes-gcm-siv", "generic-array"]
serde = ["dep:serde", "serde_bytes", "ed25519-dalek/serde", "x25519-dalek/serde"]
asymmetric = ["x25519-dalek", "ed25519-dalek", "zeroize"]
hashing = ["blake3", "digest", "hkdf", "hmac", "generic-array", "sha2"]
hashing = ["blake3", "digest", "hkdf", "hmac", "generic-array"]
stream_cipher = ["aes", "ctr", "cipher", "generic-array"]
sphinx = ["nym-sphinx-types/sphinx"]
[lints]
workspace = true
outfox = ["nym-sphinx-types/outfox"]
+100 -18
View File
@@ -202,18 +202,6 @@ impl PemStorableKey for PublicKey {
}
}
impl From<x25519_dalek::PublicKey> for PublicKey {
fn from(public_key: x25519_dalek::PublicKey) -> Self {
PublicKey(public_key)
}
}
impl From<PublicKey> for x25519_dalek::PublicKey {
fn from(public_key: PublicKey) -> Self {
public_key.0
}
}
#[derive(Zeroize, ZeroizeOnDrop)]
pub struct PrivateKey(x25519_dalek::StaticSecret);
@@ -320,15 +308,109 @@ impl PemStorableKey for PrivateKey {
}
}
impl From<x25519_dalek::StaticSecret> for PrivateKey {
fn from(secret: x25519_dalek::StaticSecret) -> Self {
PrivateKey(secret)
// compatibility with sphinx keys:
#[cfg(feature = "sphinx")]
impl From<PublicKey> for nym_sphinx_types::PublicKey {
fn from(key: PublicKey) -> Self {
nym_sphinx_types::PublicKey::from(key.to_bytes())
}
}
impl AsRef<x25519_dalek::StaticSecret> for PrivateKey {
fn as_ref(&self) -> &x25519_dalek::StaticSecret {
&self.0
#[cfg(feature = "sphinx")]
impl<'a> From<&'a PublicKey> for nym_sphinx_types::PublicKey {
fn from(key: &'a PublicKey) -> Self {
nym_sphinx_types::PublicKey::from((*key).to_bytes())
}
}
#[cfg(feature = "sphinx")]
impl From<nym_sphinx_types::PublicKey> for PublicKey {
fn from(pub_key: nym_sphinx_types::PublicKey) -> Self {
Self(x25519_dalek::PublicKey::from(*pub_key.as_bytes()))
}
}
#[cfg(feature = "sphinx")]
impl From<PrivateKey> for nym_sphinx_types::PrivateKey {
fn from(key: PrivateKey) -> Self {
nym_sphinx_types::PrivateKey::from(key.to_bytes())
}
}
#[cfg(feature = "sphinx")]
impl<'a> From<&'a PrivateKey> for nym_sphinx_types::PrivateKey {
fn from(key: &'a PrivateKey) -> Self {
nym_sphinx_types::PrivateKey::from(key.to_bytes())
}
}
#[cfg(feature = "sphinx")]
impl From<nym_sphinx_types::PrivateKey> for PrivateKey {
fn from(private_key: nym_sphinx_types::PrivateKey) -> Self {
let private_key_bytes = private_key.to_bytes();
assert_eq!(private_key_bytes.len(), PRIVATE_KEY_SIZE);
Self::from_bytes(&private_key_bytes).unwrap()
}
}
#[cfg(test)]
mod sphinx_key_conversion {
use super::*;
use rand_chacha::rand_core::SeedableRng;
use rand_chacha::ChaCha20Rng;
pub(super) fn test_rng() -> ChaCha20Rng {
let dummy_seed = [42u8; 32];
ChaCha20Rng::from_seed(dummy_seed)
}
const NUM_ITERATIONS: usize = 100;
#[test]
fn works_for_forward_conversion() {
let mut rng = test_rng();
for _ in 0..NUM_ITERATIONS {
let keys = KeyPair::new(&mut rng);
let private = &keys.private_key;
let public = &keys.public_key;
let dummy_remote = KeyPair::new(&mut rng);
let dh1 = private.diffie_hellman(&dummy_remote.public_key);
let public_bytes = public.to_bytes();
let sphinx_private: nym_sphinx_types::PrivateKey = private.into();
let recovered_private = PrivateKey::from(sphinx_private);
let dh2 = recovered_private.diffie_hellman(&dummy_remote.public_key);
let sphinx_public: nym_sphinx_types::PublicKey = public.into();
let recovered_public = PublicKey::from(sphinx_public);
assert_eq!(public_bytes, recovered_public.to_bytes());
// even though the byte representation of the private key changed, the resultant DH is the same
// which is what matters
assert_eq!(dh1, dh2);
}
}
#[test]
fn works_for_backward_conversion() {
for _ in 0..NUM_ITERATIONS {
let (sphinx_private, sphinx_public) = nym_sphinx_types::crypto::keygen();
let private_bytes = sphinx_private.to_bytes();
let public_bytes = sphinx_public.as_bytes();
let private: PrivateKey = sphinx_private.into();
let recovered_sphinx_private: nym_sphinx_types::PrivateKey = private.into();
let public: PublicKey = sphinx_public.into();
let recovered_sphinx_public: nym_sphinx_types::PublicKey = public.into();
assert_eq!(private_bytes, recovered_sphinx_private.to_bytes());
assert_eq!(public_bytes, recovered_sphinx_public.as_bytes());
}
}
}
+3 -37
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
pub use ed25519_dalek::SignatureError;
use ed25519_dalek::{SecretKey, Signer, SigningKey};
use ed25519_dalek::{Signer, SigningKey};
pub use ed25519_dalek::{Verifier, PUBLIC_KEY_LENGTH, SECRET_KEY_LENGTH, SIGNATURE_LENGTH};
use nym_pemstore::traits::{PemStorableKey, PemStorableKeyPair};
use std::fmt::{self, Debug, Display, Formatter};
@@ -18,7 +18,7 @@ pub mod serde_helpers;
use nym_sphinx_types::{DestinationAddressBytes, DESTINATION_ADDRESS_LENGTH};
#[cfg(feature = "rand")]
use rand::{CryptoRng, Rng, RngCore};
use rand::{CryptoRng, RngCore};
#[cfg(feature = "serde")]
use serde::de::Error as SerdeError;
#[cfg(feature = "serde")]
@@ -62,33 +62,16 @@ pub struct KeyPair {
// nothing secret about public key
#[zeroize(skip)]
public_key: PublicKey,
#[zeroize(skip)]
index: u32,
}
/// All keys will always have an index field populated this is to prevent anyone from figuring out if
/// the keys are derived or random, and alter their behaviour based on that.
impl KeyPair {
#[cfg(feature = "rand")]
pub fn new<R: RngCore + CryptoRng>(rng: &mut R) -> Self {
let index = rng.gen();
let ed25519_signing_key = ed25519_dalek::SigningKey::generate(rng);
KeyPair {
private_key: PrivateKey(ed25519_signing_key.to_bytes()),
public_key: PublicKey(ed25519_signing_key.verifying_key()),
index,
}
}
pub fn from_secret(secret: SecretKey, index: u32) -> Self {
let ed25519_signing_key = SigningKey::from(secret);
KeyPair {
private_key: PrivateKey(ed25519_signing_key.to_bytes()),
public_key: PublicKey(ed25519_signing_key.verifying_key()),
index,
}
}
@@ -104,31 +87,15 @@ impl KeyPair {
Ok(KeyPair {
private_key: PrivateKey::from_bytes(priv_bytes)?,
public_key: PublicKey::from_bytes(pub_bytes)?,
index: fake_index(pub_bytes),
})
}
}
/// Reduces a byte slice into a u32 value by XOR-ing all its bytes into a 4-byte accumulator.
/// The process iterates over every byte in the input slice, XOR-ing each one into a slot based on its index modulo 4.
/// If the input slice contains fewer than 4 bytes, the remaining positions in the accumulator remain zero.
/// Finally, the accumulator is interpreted in big-endian order to produce the resulting u32.
/// Index is used to verify deterministic identity key, master key and salt are also requried for verification.
fn fake_index(input: &[u8]) -> u32 {
let mut accumulator = [0u8; 4];
for (i, &byte) in input.iter().enumerate() {
accumulator[i % 4] ^= byte;
}
u32::from_be_bytes(accumulator)
}
impl From<PrivateKey> for KeyPair {
fn from(private_key: PrivateKey) -> Self {
let public_key = (&private_key).into();
KeyPair {
public_key,
public_key: (&private_key).into(),
private_key,
index: fake_index(public_key.to_bytes().as_ref()),
}
}
}
@@ -148,7 +115,6 @@ impl PemStorableKeyPair for KeyPair {
KeyPair {
private_key,
public_key,
index: fake_index(public_key.to_bytes().as_ref()),
}
}
}
-81
View File
@@ -8,10 +8,6 @@ use hkdf::{
},
Hkdf,
};
use sha2::{Sha256, Sha512};
pub use hkdf::InvalidLength;
use zeroize::ZeroizeOnDrop;
/// Perform HKDF `extract` then `expand` as a single step.
pub fn extract_then_expand<D>(
@@ -32,80 +28,3 @@ where
Ok(okm)
}
/// `DerivationMaterial` encapsulates parameters for deterministic key derivation using
/// HKDF (SHA-512).
///
/// It consists of:
/// - A master key (`master_key`): the base secret.
/// - An index (`index`): ensures unique derivations.
/// - A salt (`salt`): adds additional uniqueness, should be application specific.
///
/// Use the `derive_secret()` method to generate a 32-byte secret. To prepare for a new derivation,
/// call the `next()` method, which increments the index. **It is the caller's responsibility to
/// track and persist the derivation index if keys need to be rederived.**
///
/// # Example
///
/// ```rust
/// use nym_crypto::hkdf::DerivationMaterial;
///
/// let master_key = [0u8; 32]; // your secret master key
/// let salt = b"unique-salt-value";
/// let material = DerivationMaterial::new(master_key, 0, salt);
///
/// // Derive a secret
/// let secret = material.derive_secret().expect("Failed to derive secret");
///
/// // Prepare for the next derivation
/// let next_material = material.next();
/// ```
#[derive(ZeroizeOnDrop)]
pub struct DerivationMaterial {
master_key: [u8; 32],
index: u32,
salt: Vec<u8>,
}
impl DerivationMaterial {
pub fn index(&self) -> u32 {
self.index
}
pub fn salt(&self) -> &[u8] {
&self.salt
}
/// Derives a 32-byte seed from a master seed and an index using HKDF (with SHA-512).
///
/// The `salt` and the use of the index (as info) bind this derivation to an application/client.
pub fn derive_secret(&self) -> Result<[u8; 32], hkdf::InvalidLength> {
let salt = &self.salt;
let info = self.index.to_be_bytes(); // Use the index as info
let hk = Hkdf::<Sha512>::new(Some(salt), &self.master_key);
let mut okm = [0u8; 32];
hk.expand(&info, &mut okm)?;
Ok(okm)
}
pub fn new<T: AsRef<[u8]>>(master_key: T, index: u32, salt: &[u8]) -> Self {
// Coerce master_key to [u8; 32]
let mut hasher = Sha256::new();
hasher.update(master_key.as_ref());
let master_key = hasher.finalize().into();
Self {
master_key,
index,
salt: salt.to_vec(),
}
}
pub fn next(&self) -> Self {
Self {
master_key: self.master_key,
index: self.index + 1,
salt: self.salt.clone(),
}
}
}
+4 -10
View File
@@ -16,11 +16,8 @@ pub fn compute_keyed_hmac<D>(key: &[u8], data: &[u8]) -> HmacOutput<D>
where
D: Digest + BlockSizeUser,
{
// SAFETY: hmac is fine with keys of any size; if they're smaller than the block size of the underlying
// digest, they're padded with 0. if they're larger they're hashed and padded
// the reason for `Result` return type is due to the trait definition
#[allow(clippy::unwrap_used)]
let mut hmac = SimpleHmac::<D>::new_from_slice(key).unwrap();
let mut hmac = SimpleHmac::<D>::new_from_slice(key)
.expect("HMAC was instantiated with a key of an invalid size!");
hmac.update(data);
hmac.finalize()
}
@@ -30,11 +27,8 @@ pub fn recompute_keyed_hmac_and_verify_tag<D>(key: &[u8], data: &[u8], tag: &[u8
where
D: Digest + BlockSizeUser,
{
// SAFETY: hmac is fine with keys of any size; if they're smaller than the block size of the underlying
// digest, they're padded with 0. if they're larger they're hashed and padded
// the reason for `Result` return type is due to the trait definition
#[allow(clippy::unwrap_used)]
let mut hmac = SimpleHmac::<D>::new_from_slice(key).unwrap();
let mut hmac = SimpleHmac::<D>::new_from_slice(key)
.expect("HMAC was instantiated with a key of an invalid size!");
hmac.update(data);
let tag_arr = Output::<D>::from_slice(tag);
+5 -14
View File
@@ -27,16 +27,12 @@ where
// after performing diffie-hellman we don't care about the private component anymore
let dh_result = ephemeral_keypair.private_key().diffie_hellman(remote_key);
// SAFETY: while this is a relatively weak assumption, it's unlikely that any stream cipher has `C::key_size()`
// larger than 255 * chunk_size of the digest (so for example keys larger than 8160 bytes if sh256 is used)
#[allow(clippy::expect_used)]
// there is no reason for this to fail as our okm is expected to be only C::KeySize bytes
let okm = hkdf::extract_then_expand::<D>(None, &dh_result, None, C::key_size())
.expect("somehow too long okm was provided");
// SAFETY: the generated okm has exactly `C::key_size()` elements,
// so this call is safe
#[allow(clippy::unwrap_used)]
let derived_shared_key = Key::<C>::from_exact_iter(okm).unwrap();
let derived_shared_key =
Key::<C>::from_exact_iter(okm).expect("okm was expanded to incorrect length!");
(ephemeral_keypair, derived_shared_key)
}
@@ -52,14 +48,9 @@ where
{
let dh_result = local_key.diffie_hellman(remote_key);
// SAFETY: while this is a relatively weak assumption, it's unlikely that any stream cipher has `C::key_size()`
// larger than 255 * chunk_size of the digest (so for example keys larger than 8160 bytes if sh256 is used)
#[allow(clippy::expect_used)]
// there is no reason for this to fail as our okm is expected to be only C::KeySize bytes
let okm = hkdf::extract_then_expand::<D>(None, &dh_result, None, C::key_size())
.expect("somehow too long okm was provided");
// SAFETY: the generated okm has exactly `C::key_size()` elements,
// so this call is safe
#[allow(clippy::unwrap_used)]
Key::<C>::from_exact_iter(okm).unwrap()
Key::<C>::from_exact_iter(okm).expect("okm was expanded to incorrect length!")
}
+9 -4
View File
@@ -60,15 +60,20 @@ where
Iv::<C>::default()
}
pub fn try_iv_from_slice<C>(b: &[u8]) -> Option<&IV<C>>
pub fn iv_from_slice<C>(b: &[u8]) -> &IV<C>
where
C: IvSizeUser,
{
if b.len() != C::iv_size() {
None
} else {
Some(IV::<C>::from_slice(b))
// `from_slice` would have caused a panic about this issue anyway.
// Now we at least have slightly more information
panic!(
"Tried to convert {} bytes to IV. Expected {}",
b.len(),
C::iv_size()
)
}
IV::<C>::from_slice(b)
}
// TODO: there's really no way to use more parts of the keystream if it was required at some point.
+14
View File
@@ -0,0 +1,14 @@
[package]
name = "nym-ecash-double-spending"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
[dependencies]
bit-vec = { workspace = true }
bloomfilter = { workspace = true }
nym-network-defaults = { path = "../network-defaults" }
+136
View File
@@ -0,0 +1,136 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use bit_vec::BitVec;
use bloomfilter::Bloom;
use nym_network_defaults::{BloomfilterParameters, ECASH_DS_BLOOMFILTER_PARAMS};
pub struct DoubleSpendingFilter {
params: BloomfilterParameters,
inner: Bloom<Vec<u8>>,
}
impl Default for DoubleSpendingFilter {
fn default() -> Self {
DoubleSpendingFilter::new_empty_ecash()
}
}
pub fn bloom_from_params<T>(params: &BloomfilterParameters, bitvec: BitVec) -> Bloom<Vec<T>> {
assert_eq!(params.bitmap_size, bitvec.len() as u64);
Bloom::from_bit_vec(
bitvec,
params.bitmap_size,
params.num_hashes,
params.sip_keys,
)
}
impl DoubleSpendingFilter {
pub fn new_empty(params: BloomfilterParameters) -> Self {
let bitvec = BitVec::from_elem(params.bitmap_size as usize, false);
DoubleSpendingFilter {
inner: bloom_from_params(&params, bitvec),
params,
}
}
pub fn params(&self) -> BloomfilterParameters {
self.params
}
pub fn rebuild(&self) -> DoubleSpendingFilterBuilder {
DoubleSpendingFilterBuilder::new(self.params)
}
pub fn reset(&mut self) {
self.inner.clear()
}
pub fn new_empty_ecash() -> Self {
DoubleSpendingFilter::new_empty(ECASH_DS_BLOOMFILTER_PARAMS)
}
pub fn builder(params: BloomfilterParameters) -> DoubleSpendingFilterBuilder {
DoubleSpendingFilterBuilder::new(params)
}
pub fn from_bytes(params: BloomfilterParameters, bitmap: &[u8]) -> Self {
DoubleSpendingFilter {
inner: bloom_from_params(&params, BitVec::from_bytes(bitmap)),
params,
}
}
pub fn replace_bitvec(&mut self, new: BitVec) {
self.inner = bloom_from_params(&self.params, new)
}
pub fn dump_bitmap(&self) -> Vec<u8> {
self.inner.bitmap()
}
pub fn set(&mut self, b: &Vec<u8>) {
self.inner.set(b);
}
pub fn check(&self, b: &Vec<u8>) -> bool {
self.inner.check(b)
}
}
pub struct DoubleSpendingFilterBuilder {
params: BloomfilterParameters,
bit_vec_builder: Option<BitVecBuilder>,
}
impl DoubleSpendingFilterBuilder {
pub fn new(params: BloomfilterParameters) -> Self {
DoubleSpendingFilterBuilder {
params,
bit_vec_builder: None,
}
}
pub fn add_bytes(&mut self, b: &[u8]) -> bool {
match &mut self.bit_vec_builder {
None => {
self.bit_vec_builder = Some(BitVecBuilder::new(b));
true
}
Some(builder) => builder.add_bytes(b),
}
}
pub fn build(self) -> DoubleSpendingFilter {
match self.bit_vec_builder {
None => DoubleSpendingFilter::new_empty(self.params),
Some(builder) => DoubleSpendingFilter {
inner: bloom_from_params(&self.params, builder.finish()),
params: self.params,
},
}
}
}
pub struct BitVecBuilder(BitVec);
impl BitVecBuilder {
pub fn new(initial_bitmap: &[u8]) -> Self {
BitVecBuilder(BitVec::from_bytes(initial_bitmap))
}
pub fn add_bytes(&mut self, b: &[u8]) -> bool {
let add = BitVec::from_bytes(b);
if self.0.len() != add.len() {
return false;
}
self.0.or(&add);
true
}
pub fn finish(self) -> BitVec {
self.0
}
}
-3
View File
@@ -20,14 +20,11 @@ serde_json = { workspace = true }
strum = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true, features = ["log"] }
time = { workspace = true }
subtle = { workspace = true }
zeroize = { workspace = true }
nym-crypto = { path = "../crypto", features = ["aead", "hashing"] }
nym-pemstore = { path = "../pemstore" }
nym-sphinx = { path = "../nymsphinx" }
nym-serde-helpers = { path = "../serde-helpers", features = ["base64"] }
nym-task = { path = "../task" }
nym-credentials = { path = "../credentials" }
@@ -0,0 +1,67 @@
// Copyright 2020-2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::shared_key::{SharedGatewayKey, SharedKeyUsageError};
use nym_sphinx::DestinationAddressBytes;
use thiserror::Error;
/// Replacement for what used to be an `AuthToken`.
///
/// Replacement for what used to be an `AuthToken`. We used to be generating an `AuthToken` based on
/// local secret and remote address in order to allow for authentication. Due to changes in registration
/// and the fact we are deriving a shared key, we are encrypting remote's address with the previously
/// derived shared key. If the value is as expected, then authentication is successful.
#[derive(Debug, PartialEq, Eq, Hash, Clone)]
// this is no longer constant size due to the differences in ciphertext between aes128ctr and aes256gcm-siv (inclusion of tag)
pub struct EncryptedAddressBytes(Vec<u8>);
#[derive(Debug, Error)]
pub enum EncryptedAddressConversionError {
#[error("Failed to decode the encrypted address - {0}")]
DecodeError(#[from] bs58::decode::Error),
}
impl EncryptedAddressBytes {
pub fn new(
address: &DestinationAddressBytes,
key: &SharedGatewayKey,
nonce: &[u8],
) -> Result<Self, SharedKeyUsageError> {
let ciphertext = key.encrypt_naive(address.as_bytes_ref(), Some(nonce))?;
Ok(EncryptedAddressBytes(ciphertext))
}
pub fn verify(
&self,
address: &DestinationAddressBytes,
key: &SharedGatewayKey,
nonce: &[u8],
) -> bool {
let Ok(reconstructed) = Self::new(address, key, nonce) else {
return false;
};
self == &reconstructed
}
pub fn as_bytes(&self) -> &[u8] {
&self.0
}
pub fn try_from_base58_string<S: Into<String>>(
val: S,
) -> Result<Self, EncryptedAddressConversionError> {
let decoded = bs58::decode(val.into()).into_vec()?;
Ok(EncryptedAddressBytes(decoded))
}
pub fn to_base58_string(self) -> String {
bs58::encode(self.0).into_string()
}
}
impl From<EncryptedAddressBytes> for String {
fn from(val: EncryptedAddressBytes) -> Self {
val.to_base58_string()
}
}
@@ -0,0 +1,4 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod encrypted_address;
+12 -18
View File
@@ -2,17 +2,24 @@
// SPDX-License-Identifier: Apache-2.0
pub use nym_crypto::generic_array;
use nym_crypto::OutputSizeUser;
use nym_sphinx::params::GatewayIntegrityHmacAlgorithm;
pub use types::*;
pub mod authentication;
pub mod models;
pub mod registration;
pub mod shared_key;
pub mod types;
pub use shared_key::{SharedKeyConversionError, SharedKeyUsageError, SharedSymmetricKey};
pub use shared_key::helpers::SymmetricKey;
pub use shared_key::legacy::{LegacySharedKeySize, LegacySharedKeys};
pub use shared_key::{
SharedGatewayKey, SharedKeyConversionError, SharedKeyUsageError, SharedSymmetricKey,
};
pub const CURRENT_PROTOCOL_VERSION: u8 = AUTHENTICATE_V2_PROTOCOL_VERSION;
pub const CURRENT_PROTOCOL_VERSION: u8 = AES_GCM_SIV_PROTOCOL_VERSION;
/// Defines the current version of the communication protocol between gateway and clients.
/// It has to be incremented for any breaking change.
@@ -20,23 +27,10 @@ pub const CURRENT_PROTOCOL_VERSION: u8 = AUTHENTICATE_V2_PROTOCOL_VERSION;
// 1 - initial release
// 2 - changes to client credentials structure
// 3 - change to AES-GCM-SIV and non-zero IVs
// 4 - introduction of v2 authentication protocol to prevent reply attacks
pub const INITIAL_PROTOCOL_VERSION: u8 = 1;
pub const CREDENTIAL_UPDATE_V2_PROTOCOL_VERSION: u8 = 2;
pub const AES_GCM_SIV_PROTOCOL_VERSION: u8 = 3;
pub const AUTHENTICATE_V2_PROTOCOL_VERSION: u8 = 4;
pub trait GatewayProtocolVersionExt {
fn supports_aes256_gcm_siv(&self) -> bool;
fn supports_authenticate_v2(&self) -> bool;
}
impl GatewayProtocolVersionExt for u8 {
fn supports_aes256_gcm_siv(&self) -> bool {
*self >= AES_GCM_SIV_PROTOCOL_VERSION
}
fn supports_authenticate_v2(&self) -> bool {
*self >= AUTHENTICATE_V2_PROTOCOL_VERSION
}
}
// TODO: could using `Mac` trait here for OutputSize backfire?
// Should hmac itself be exposed, imported and used instead?
pub type LegacyGatewayMacSize = <GatewayIntegrityHmacAlgorithm as OutputSizeUser>::OutputSize;
@@ -3,7 +3,7 @@
use crate::registration::handshake::messages::{Finalization, GatewayMaterialExchange};
use crate::registration::handshake::state::State;
use crate::registration::handshake::SharedSymmetricKey;
use crate::registration::handshake::SharedGatewayKey;
use crate::registration::handshake::{error::HandshakeError, WsItem};
use futures::{Sink, Stream};
use rand::{CryptoRng, RngCore};
@@ -15,12 +15,12 @@ impl<S, R> State<'_, S, R> {
S: Stream<Item = WsItem> + Sink<WsMessage> + Unpin,
R: CryptoRng + RngCore,
{
// 1. generate initiator salt for kdf
let hkdf_salt = self.generate_initiator_salt();
// 1. if we're using non-legacy, i.e. aes256gcm-siv derivation, generate initiator salt for kdf
let maybe_hkdf_salt = self.maybe_generate_initiator_salt();
// 1. send ed25519 pubkey alongside ephemeral x25519 pubkey and a hkdf salt
// 1. send ed25519 pubkey alongside ephemeral x25519 pubkey and a hkdf salt if we're using non-legacy client
// LOCAL_ID_PUBKEY || EPHEMERAL_KEY || MAYBE_SALT
let init_message = self.init_message(hkdf_salt.clone());
let init_message = self.init_message(maybe_hkdf_salt.clone());
self.send_handshake_data(init_message).await?;
// 2. wait for response with remote x25519 pubkey as well as encrypted signature
@@ -31,7 +31,7 @@ impl<S, R> State<'_, S, R> {
// 3. derive shared keys locally
// hkdf::<blake3>::(g^xy)
self.derive_shared_key(&mid_res.ephemeral_dh, &hkdf_salt);
self.derive_shared_key(&mid_res.ephemeral_dh, maybe_hkdf_salt.as_deref());
// 4. verify the received signature using the locally derived keys
self.verify_remote_key_material(&mid_res.materials, &mid_res.ephemeral_dh)?;
@@ -49,7 +49,7 @@ impl<S, R> State<'_, S, R> {
pub(crate) async fn perform_client_handshake(
mut self,
) -> Result<SharedSymmetricKey, HandshakeError>
) -> Result<SharedGatewayKey, HandshakeError>
where
S: Stream<Item = WsItem> + Sink<WsMessage> + Unpin,
R: CryptoRng + RngCore,
@@ -5,7 +5,7 @@ use crate::registration::handshake::messages::{
HandshakeMessage, Initialisation, MaterialExchange,
};
use crate::registration::handshake::state::State;
use crate::registration::handshake::SharedSymmetricKey;
use crate::registration::handshake::SharedGatewayKey;
use crate::registration::handshake::{error::HandshakeError, WsItem};
use futures::{Sink, Stream};
use tungstenite::Message as WsMessage;
@@ -18,14 +18,18 @@ impl<S, R> State<'_, S, R> {
where
S: Stream<Item = WsItem> + Sink<WsMessage> + Unpin,
{
// 1. receive remote ed25519 pubkey alongside ephemeral x25519 pubkey and initiator salt
// LOCAL_ID_PUBKEY || EPHEMERAL_KEY || INITIATOR_SALT
// 1. receive remote ed25519 pubkey alongside ephemeral x25519 pubkey and maybe a flag indicating non-legacy client
// LOCAL_ID_PUBKEY || EPHEMERAL_KEY || MAYBE_NON_LEGACY
let init_message = Initialisation::try_from_bytes(&raw_init_message)?;
self.update_remote_identity(init_message.identity);
self.set_aes256_gcm_siv_key_derivation(!init_message.is_legacy());
// 2. derive shared keys locally
// hkdf::<blake3>::(g^xy)
self.derive_shared_key(&init_message.ephemeral_dh, &init_message.initiator_salt);
self.derive_shared_key(
&init_message.ephemeral_dh,
init_message.initiator_salt.as_deref(),
);
// 3. send ephemeral x25519 pubkey alongside the encrypted signature
// g^y || AES(k, sig(gate_priv, (g^y || g^x))
@@ -50,7 +54,7 @@ impl<S, R> State<'_, S, R> {
pub(crate) async fn perform_gateway_handshake(
mut self,
raw_init_message: Vec<u8>,
) -> Result<SharedSymmetricKey, HandshakeError>
) -> Result<SharedGatewayKey, HandshakeError>
where
S: Stream<Item = WsItem> + Sink<WsMessage> + Unpin,
{
@@ -4,7 +4,7 @@
use crate::registration::handshake::error::HandshakeError;
use crate::registration::handshake::KDF_SALT_LENGTH;
use nym_crypto::asymmetric::{ed25519, x25519};
use nym_crypto::symmetric::aead::{nonce_size, tag_size, Nonce};
use nym_crypto::symmetric::aead::{nonce_size, tag_size};
use nym_sphinx::params::GatewayEncryptionAlgorithm;
// it is vital nobody changes the serialisation implementation unless you have an EXTREMELY good reason,
@@ -21,13 +21,20 @@ pub trait HandshakeMessage {
pub struct Initialisation {
pub identity: ed25519::PublicKey,
pub ephemeral_dh: x25519::PublicKey,
pub initiator_salt: Vec<u8>,
pub initiator_salt: Option<Vec<u8>>,
}
impl Initialisation {
#[cfg(not(target_arch = "wasm32"))]
pub fn is_legacy(&self) -> bool {
self.initiator_salt.is_none()
}
}
#[derive(Debug)]
pub struct MaterialExchange {
pub signature_ciphertext: Vec<u8>,
pub nonce: Nonce<GatewayEncryptionAlgorithm>,
pub nonce: Option<Vec<u8>>,
}
impl MaterialExchange {
@@ -65,12 +72,17 @@ impl HandshakeMessage for Initialisation {
// Eventually the ID_PUBKEY prefix will get removed and recipient will know
// initializer's identity from another source.
fn into_bytes(self) -> Vec<u8> {
self.identity
let bytes = self
.identity
.to_bytes()
.into_iter()
.chain(self.ephemeral_dh.to_bytes())
.chain(self.initiator_salt)
.collect()
.chain(self.ephemeral_dh.to_bytes());
if let Some(salt) = self.initiator_salt {
bytes.chain(salt).collect()
} else {
bytes.collect()
}
}
// this will need to be adjusted when REMOTE_ID_PUBKEY is removed
@@ -78,8 +90,9 @@ impl HandshakeMessage for Initialisation {
where
Self: Sized,
{
let current_len = ed25519::PUBLIC_KEY_LENGTH + x25519::PUBLIC_KEY_SIZE + KDF_SALT_LENGTH;
if bytes.len() != current_len {
let legacy_len = ed25519::PUBLIC_KEY_LENGTH + x25519::PUBLIC_KEY_SIZE;
let current_len = legacy_len + KDF_SALT_LENGTH;
if bytes.len() != legacy_len && bytes.len() != current_len {
return Err(HandshakeError::MalformedRequest);
}
@@ -88,13 +101,14 @@ impl HandshakeMessage for Initialisation {
// this can only fail if the provided bytes have len different from encryption::PUBLIC_KEY_SIZE
// which is impossible
let ephemeral_dh = x25519::PublicKey::from_bytes(
&bytes
[ed25519::PUBLIC_KEY_LENGTH..ed25519::PUBLIC_KEY_LENGTH + x25519::PUBLIC_KEY_SIZE],
)
.unwrap();
let ephemeral_dh =
x25519::PublicKey::from_bytes(&bytes[ed25519::PUBLIC_KEY_LENGTH..legacy_len]).unwrap();
let initiator_salt = bytes[ed25519::PUBLIC_KEY_LENGTH + x25519::PUBLIC_KEY_SIZE..].to_vec();
let initiator_salt = if bytes.len() == legacy_len {
None
} else {
Some(bytes[legacy_len..].to_vec())
};
Ok(Initialisation {
identity,
@@ -107,31 +121,43 @@ impl HandshakeMessage for Initialisation {
impl HandshakeMessage for MaterialExchange {
// AES(k, SIG(PRIV_GATE, G^y || G^x))
fn into_bytes(self) -> Vec<u8> {
self.signature_ciphertext
.iter()
.cloned()
.chain(self.nonce)
.collect()
if let Some(nonce) = self.nonce {
self.signature_ciphertext
.iter()
.cloned()
.chain(nonce)
.collect()
} else {
self.signature_ciphertext.to_vec()
}
}
fn try_from_bytes(bytes: &[u8]) -> Result<Self, HandshakeError>
where
Self: Sized,
{
// we expect to receive either:
// LEGACY: ed25519 signature ciphertext (64 bytes)
// CURRENT: ed25519 signature ciphertext (+ tag) + AES256-GCM-SIV nonce (76 bytes)
let current_len = ed25519::SIGNATURE_LENGTH
let legacy_len = ed25519::SIGNATURE_LENGTH;
let current_len = legacy_len
+ tag_size::<GatewayEncryptionAlgorithm>()
+ nonce_size::<GatewayEncryptionAlgorithm>();
if bytes.len() != current_len {
if bytes.len() != legacy_len && bytes.len() != current_len {
return Err(HandshakeError::MalformedResponse);
}
let ciphertext_len = ed25519::SIGNATURE_LENGTH + tag_size::<GatewayEncryptionAlgorithm>();
let signature_ciphertext = bytes[..ciphertext_len].to_vec();
// SAFETY: we know the bytes have correct length
let nonce = Nonce::<GatewayEncryptionAlgorithm>::clone_from_slice(&bytes[ciphertext_len..]);
let (signature_ciphertext, nonce) = if bytes.len() == current_len {
let ciphertext_len =
ed25519::SIGNATURE_LENGTH + tag_size::<GatewayEncryptionAlgorithm>();
(
bytes[..ciphertext_len].to_vec(),
Some(bytes[ciphertext_len..].to_vec()),
)
} else {
(bytes.to_vec(), None)
};
Ok(MaterialExchange {
signature_ciphertext,

Some files were not shown because too many files have changed in this diff Show More