Compare commits

..

4 Commits

Author SHA1 Message Date
Tommy Verrall d39ec4a048 more white space 2025-07-15 16:06:46 +02:00
Tommy Verrall c03c580b1f indentation 2025-07-15 16:05:51 +02:00
Tommy Verrall e0cc094ac8 revert-revert-revert 2025-07-15 15:54:12 +02:00
Tommy Verrall f1fcaa309b try changing the workflow 2025-07-15 15:39:21 +02:00
429 changed files with 9919 additions and 21209 deletions
@@ -38,14 +38,15 @@ jobs:
rm -rf ci-builds || true
mkdir -p $OUTPUT_DIR
echo $OUTPUT_DIR
- name: Install Dependencies (Linux)
run: sudo apt-get update && sudo apt-get -y install libudev-dev
- name: Sets env vars for tokio if set in manual dispatch inputs
if: github.event_name == 'workflow_dispatch' && inputs.add_tokio_unstable == true
run: |
echo "RUSTFLAGS=--cfg tokio_unstable" >> $GITHUB_ENV
echo "CARGO_FEATURES=--features tokio-console" >> $GITHUB_ENV
echo 'RUSTFLAGS="--cfg tokio_unstable"' >> $GITHUB_ENV
if: github.event_name == 'workflow_dispatch' && inputs.add_tokio_unstable == true
- name: Install Rust stable
uses: actions-rs/toolchain@v1
with:
@@ -102,6 +103,7 @@ jobs:
if [ ${{ github.event_name == 'workflow_dispatch' && inputs.enable_deb == true }} = true ]; then
cp target/debian/*.deb $OUTPUT_DIR
fi
- name: Deploy branch to CI www
continue-on-error: true
uses: easingthemes/ssh-deploy@main
+7 -7
View File
@@ -38,7 +38,7 @@ jobs:
strategy:
fail-fast: false
matrix:
os: [ arc-linux-latest, custom-windows-11, custom-macos-15 ]
os: [ arc-ubuntu-22.04, custom-windows-11, custom-macos-15 ]
runs-on: ${{ matrix.os }}
env:
CARGO_TERM_COLOR: always
@@ -46,9 +46,9 @@ jobs:
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 cmake
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
continue-on-error: true
if: contains(matrix.os, 'linux')
if: contains(matrix.os, 'ubuntu')
- name: Check out repository code
uses: actions/checkout@v4
@@ -63,7 +63,7 @@ jobs:
# To avoid running out of disk space, skip generating debug symbols
- name: Set debug to false (unix)
if: contains(matrix.os, 'linux') || contains(matrix.os, 'mac')
if: contains(matrix.os, 'ubuntu') || contains(matrix.os, 'mac')
run: |
sed -i.bak 's/\[profile.dev\]/\[profile.dev\]\ndebug = false/' Cargo.toml
git diff
@@ -93,14 +93,14 @@ jobs:
command: build
- name: Build all examples
if: contains(matrix.os, 'linux')
if: contains(matrix.os, 'ubuntu')
uses: actions-rs/cargo@v1
with:
command: build
args: --workspace --examples
- name: Run all tests
if: contains(matrix.os, 'linux')
if: contains(matrix.os, 'ubuntu')
uses: actions-rs/cargo@v1
env:
NYM_API: https://sandbox-nym-api1.nymtech.net/api
@@ -109,7 +109,7 @@ jobs:
args: --workspace
- name: Run expensive tests
if: (github.ref == 'refs/heads/develop' || github.event.pull_request.base.ref == 'develop' || github.event.pull_request.base.ref == 'master') && contains(matrix.os, 'linux')
if: (github.ref == 'refs/heads/develop' || github.event.pull_request.base.ref == 'develop' || github.event.pull_request.base.ref == 'master') && contains(matrix.os, 'ubuntu')
uses: actions-rs/cargo@v1
with:
command: test
+12 -6
View File
@@ -49,6 +49,8 @@ jobs:
run: |
curl -L0 https://www.ssl.com/download/codesigntool-for-linux-and-macos/ -o codesigntool.zip
unzip codesigntool.zip
chmod +x CodeSignTool.sh
- name: Get EV certificate credential id
working-directory: nym-wallet/src-tauri
if: ${{ inputs.sign }}
@@ -56,6 +58,7 @@ jobs:
shell: bash
run: |
echo "SSL_COM_CREDENTIAL_ID=$(./CodeSignTool.sh get_credential_ids -username=${{ secrets.SSL_COM_USERNAME }} -password=${{ secrets.SSL_COM_PASSWORD }} | sed -n '1!p' | sed 's/- //')" >> "$GITHUB_OUTPUT"
- name: Add custom sign command to tauri.conf.json
working-directory: nym-wallet/src-tauri
if: ${{ inputs.sign }}
@@ -79,6 +82,7 @@ jobs:
]
}
}' tauri.conf.json
- name: Install project dependencies
shell: bash
run: cd .. && yarn --network-timeout 100000
@@ -93,12 +97,14 @@ jobs:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
TAURI_SIGNING_PRIVATE_KEY: ${{ secrets.TAURI_PRIVATE_KEY }}
TAURI_SIGNING_PRIVATE_KEY_PASSWORD: ${{ secrets.TAURI_KEY_PASSWORD }}
SSL_COM_USERNAME: ${{ inputs.sign && secrets.SSL_COM_USERNAME }}
SSL_COM_PASSWORD: ${{ inputs.sign && secrets.SSL_COM_PASSWORD }}
SSL_COM_CREDENTIAL_ID: ${{ inputs.sign && steps.get_credential_ids.outputs.SSL_COM_CREDENTIAL_ID }}
SSL_COM_TOTP_SECRET: ${{ inputs.sign && secrets.SSL_COM_TOTP_SECRET }}
SSL_COM_USERNAME: ${{ inputs.sign && secrets.SSL_COM_USERNAME || '' }}
SSL_COM_PASSWORD: ${{ inputs.sign && secrets.SSL_COM_PASSWORD || '' }}
SSL_COM_CREDENTIAL_ID: ${{ inputs.sign && steps.get_credential_ids.outputs.SSL_COM_CREDENTIAL_ID || '' }}
SSL_COM_TOTP_SECRET: ${{ inputs.sign && secrets.SSL_COM_TOTP_SECRET || '' }}
CODE_SIGN_TOOL_PATH: ${{ inputs.sign && 'C:\\actions-runner\\_work\\nym\\nym\\nym-wallet\\src-tauri\\' || '' }}
run: |
echo "Starting build process..."
echo "Signing enabled: ${{ inputs.sign }}"
yarn build
- name: Check bundle directory
@@ -147,7 +153,7 @@ jobs:
nym-wallet/${{ env.BUNDLE_PATH }}/msi/*.msi.zip*
nym-wallet/${{ env.BUNDLE_PATH }}/*/nym-wallet*.msi
nym-wallet/src-tauri/target/release/bundle/msi/*.msi
- name: Find MSI path for deployment
id: find-msi
shell: bash
@@ -167,4 +173,4 @@ jobs:
needs: publish-tauri
with:
release_tag: ${{ needs.publish-tauri.outputs.release_tag || github.ref_name }}
secrets: inherit
secrets: inherit
@@ -38,10 +38,10 @@ jobs:
git config --global user.name "Lawrence Stalder"
- name: Get version from cargo.toml
uses: mikefarah/yq@v4.45.4
id: get_version
run: |
VERSION=$(yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml)
echo "result=$VERSION" >> $GITHUB_OUTPUT
with:
cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml
- name: cleanup-gateway-probe-ref
id: cleanup_gateway_probe_ref
@@ -53,16 +53,13 @@ jobs:
- name: Set GIT_TAG variable
run: echo "GIT_TAG=${{ env.WORKING_DIRECTORY }}-${{ steps.get_version.outputs.result }}-${{ steps.cleanup_gateway_probe_ref.outputs.git_ref }}" >> $GITHUB_ENV
- name: Initialize RELEASE_TAG
run: echo "RELEASE_TAG=" >> $GITHUB_ENV
- name: Set RELEASE_TAG for release
- name: Set RELEASE_TAG variable
if: github.event.inputs.release_image == 'true'
run: echo "RELEASE_TAG=golden-" >> $GITHUB_ENV
- name: Set IMAGE_NAME_AND_TAGS variable
run: echo "IMAGE_NAME_AND_TAGS=${{ env.CONTAINER_NAME }}:${{ env.RELEASE_TAG }}${{ steps.get_version.outputs.result }}-${{ steps.cleanup_gateway_probe_ref.outputs.git_ref }}" >> $GITHUB_ENV
- name: New env vars
run: echo "RELEASE_TAG='$RELEASE_TAG' GIT_TAG='$GIT_TAG' IMAGE_NAME_AND_TAGS='$IMAGE_NAME_AND_TAGS'"
+7 -10
View File
@@ -32,24 +32,21 @@ jobs:
git config --global user.name "Lawrence Stalder"
- name: Get version from cargo.toml
uses: mikefarah/yq@v4.45.4
id: get_version
run: |
VERSION=$(yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml)
echo "result=$VERSION" >> $GITHUB_OUTPUT
with:
cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml
- name: Set GIT_TAG variable
run: echo "GIT_TAG=${{ env.WORKING_DIRECTORY }}-${{ steps.get_version.outputs.result }}" >> $GITHUB_ENV
- name: Initialise RELEASE_TAG
run: echo "RELEASE_TAG=" >> $GITHUB_ENV
- name: Set RELEASE_TAG for release
- name: Set RELEASE_TAG variable
if: github.event.inputs.release_image == 'true'
run: echo "RELEASE_TAG=golden-" >> $GITHUB_ENV
- name: Set IMAGE_NAME_AND_TAGS variable
run: echo "IMAGE_NAME_AND_TAGS=${{ env.CONTAINER_NAME }}:${{ env.RELEASE_TAG }}${{ steps.get_version.outputs.result }}" >> $GITHUB_ENV
- name: New env vars
run: echo "RELEASE_TAG='$RELEASE_TAG' GIT_TAG='$GIT_TAG' IMAGE_NAME_AND_TAGS='$IMAGE_NAME_AND_TAGS'"
@@ -69,6 +66,6 @@ jobs:
- name: BuildAndPushImageOnHarbor
run: |
docker build -f ${{ env.WORKING_DIRECTORY }}/Dockerfile-sqlite . -t harbor.nymte.ch/nym/${{ env.IMAGE_NAME_AND_TAGS }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:latest
docker build -f ${{ env.WORKING_DIRECTORY }}/Dockerfile . -t harbor.nymte.ch/nym/${{ env.IMAGE_NAME_AND_TAGS }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:latest
docker push harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }} --all-tags
-84
View File
@@ -4,90 +4,6 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
## [Unreleased]
## [2025.15-gruyere] (2025-08-20)
- Migrate strum to 0.27.2 ([#5960])
- WG exit policy scripts update ([#5921])
- Make DNS Resolver fallback optional ([#5920])
- nym-node debug command to reset providers db ([#5914])
- basic zulip client for sending messages ([#5913])
- chore: allow compatibility with 'CDLA-Permissive-2.0' ([#5910])
- feat: ecash liveness check ([#5890])
- Remove old free credential handle ([#5864])
[#5960]: https://github.com/nymtech/nym/pull/5960
[#5921]: https://github.com/nymtech/nym/pull/5921
[#5920]: https://github.com/nymtech/nym/pull/5920
[#5914]: https://github.com/nymtech/nym/pull/5914
[#5913]: https://github.com/nymtech/nym/pull/5913
[#5910]: https://github.com/nymtech/nym/pull/5910
[#5890]: https://github.com/nymtech/nym/pull/5890
[#5864]: https://github.com/nymtech/nym/pull/5864
## [2025.14-feta] (2025-08-05)
- chore: nym node tokio console ([#5909])
- Feature/dkg snapshot epoch ([#5900])
- Feature/dkg epoch dealers query ([#5899])
- sqlx-pool-guard: allocate more memory on windows ([#5896])
- Support mnemonic in the NS agent ([#5883])
- Allow PG database backend ([#5880])
[#5909]: https://github.com/nymtech/nym/pull/5909
[#5900]: https://github.com/nymtech/nym/pull/5900
[#5899]: https://github.com/nymtech/nym/pull/5899
[#5896]: https://github.com/nymtech/nym/pull/5896
[#5883]: https://github.com/nymtech/nym/pull/5883
[#5880]: https://github.com/nymtech/nym/pull/5880
## [2025.13-emmental] (2025-07-22)
- fix: don't allow mixnode running in exit mode ([#5898])
- fix contract build process in Makefile ([#5892])
- bugfix: ignore 'Send' responses when claiming bandwidth ([#5884])
- Update push-node-status-agent.yaml ([#5882])
- listen for shutdown signals during nym-node startup ([#5879])
- feat: forbid running mixnode + entry on the same node ([#5878])
- chore: 1.88 clippy ([#5877])
- Batch SQL writes for packet stats ([#5874])
- fix the broken link ([#5873])
- Set busy_timeout in sqlx ([#5872])
- feat: basic performance contract integration [within Nym API] ([#5871])
- scraper bugfix: ignore precommits from missing validators ([#5867])
- Return true remaining ([#5866])
- Make Mix hops optional for Mixnet Client SURBs ([#5861])
- Check gateway supported versions ([#5860])
- Add build info endpoints ([#5857])
- Clear out screaming logs ([#5856])
- fix removal of qa env ([#5855])
- Use display when printing paths ([#5853])
- feat: initial performance contract ([#5833])
- Security patches for the `dkg` crate ([#5828])
- HTTP Discovery objects & network defaults ([#5814])
[#5898]: https://github.com/nymtech/nym/pull/5898
[#5892]: https://github.com/nymtech/nym/pull/5892
[#5884]: https://github.com/nymtech/nym/pull/5884
[#5882]: https://github.com/nymtech/nym/pull/5882
[#5879]: https://github.com/nymtech/nym/pull/5879
[#5878]: https://github.com/nymtech/nym/pull/5878
[#5877]: https://github.com/nymtech/nym/pull/5877
[#5874]: https://github.com/nymtech/nym/pull/5874
[#5873]: https://github.com/nymtech/nym/pull/5873
[#5872]: https://github.com/nymtech/nym/pull/5872
[#5871]: https://github.com/nymtech/nym/pull/5871
[#5867]: https://github.com/nymtech/nym/pull/5867
[#5866]: https://github.com/nymtech/nym/pull/5866
[#5861]: https://github.com/nymtech/nym/pull/5861
[#5860]: https://github.com/nymtech/nym/pull/5860
[#5857]: https://github.com/nymtech/nym/pull/5857
[#5856]: https://github.com/nymtech/nym/pull/5856
[#5855]: https://github.com/nymtech/nym/pull/5855
[#5853]: https://github.com/nymtech/nym/pull/5853
[#5833]: https://github.com/nymtech/nym/pull/5833
[#5828]: https://github.com/nymtech/nym/pull/5828
[#5814]: https://github.com/nymtech/nym/pull/5814
## [2025.12-dolcelatte] (2025-07-07)
- bugfix: key-rotation + reply SURBs ([#5876])
-686
View File
@@ -1,686 +0,0 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Project Overview
Nym is a privacy platform that uses mixnet technology to protect against metadata surveillance. The platform consists of several key components:
- Mixnet nodes (mixnodes) for packet mixing
- Gateways (entry/exit points for the network)
- Clients for interacting with the network
- Network monitoring tools
- Validators for network consensus
- Various service providers and integrations
## Build Commands
### Rust Components
```bash
# Default build (debug)
cargo build
# Release build
cargo build --release
# Build a specific package
cargo build -p <package-name>
# Build main components
make build
# Build release versions of main binaries and contracts
make build-release
# Build specific binaries
make build-nym-cli
cargo build -p nym-node --release
cargo build -p nym-api --release
```
### Testing
```bash
# Run clippy, unit tests, and formatting
make test
# Run all tests including slow tests
make test-all
# Run clippy on all workspaces
make clippy
# Run unit tests for a specific package
cargo test -p <package-name>
# Run only expensive/ignored tests
cargo test --workspace -- --ignored
# Run API tests
dotenv -f envs/sandbox.env -- cargo test --test public-api-tests
# Run tests with specific log level
RUST_LOG=debug cargo test -p <package-name>
# Run specific test scripts
./nym-node/tests/test_apis.sh
./scripts/wireguard-exit-policy/exit-policy-tests.sh
```
### Linting and Formatting
```bash
# Run rustfmt on all code
make fmt
# Check formatting without modifying
cargo fmt --all -- --check
# Run clippy with all targets
cargo clippy --workspace --all-targets -- -D warnings
# TypeScript linting
yarn lint
yarn lint:fix
yarn types:lint:fix
# Check dependencies for security/licensing issues
cargo deny check
```
### WASM Components
```bash
# Build all WASM components
make sdk-wasm-build
# Build TypeScript SDK
yarn build:sdk
npx lerna run --scope @nymproject/sdk build --stream
# Build and test WASM components
make sdk-wasm
# Build specific WASM packages
cd wasm/client && make
cd wasm/mix-fetch && make
cd wasm/node-tester && make
```
### Contract Development
```bash
# Build all contracts
make contracts
# Build contracts in release mode
make build-release-contracts
# Generate contract schemas
make contract-schema
# Run wasm-opt on contracts
make wasm-opt-contracts
# Check contracts with cosmwasm-check
make cosmwasm-check-contracts
```
### Running Components
```bash
# Run nym-node as a mixnode
cargo run -p nym-node -- run --mode mixnode
# Run nym-node as a gateway
cargo run -p nym-node -- run --mode gateway
# Run the network monitor
cargo run -p nym-network-monitor
# Run the API server
cargo run -p nym-api
# Run with specific environment
dotenv -f envs/sandbox.env -- cargo run -p nym-api
# Start a local network
./scripts/localnet_start.sh
```
## Architecture
The Nym platform consists of various components organized as a monorepo:
1. **Core Mixnet Infrastructure**:
- `nym-node`: Core binary supporting mixnode and gateway modes
- `common/nymsphinx`: Implementation of the Sphinx packet format
- `common/topology`: Network topology management
- `common/types`: Shared data types across components
2. **Network Monitoring**:
- `nym-network-monitor`: Monitors the network's reliability and performance
- `nym-api`: API server for network stats and monitoring data
- Metrics tracking for nodes, routes, and overall network health
3. **Client Implementations**:
- `clients/native`: Native Rust client implementation
- `clients/socks5`: SOCKS5 proxy client for standard applications
- `wasm`: WebAssembly client implementations (for browsers)
- `nym-connect`: Desktop and mobile clients
4. **Blockchain & Smart Contracts**:
- `common/cosmwasm-smart-contracts`: Smart contract implementations
- `contracts`: CosmWasm contracts for the Nym network
- `common/ledger`: Blockchain integration
5. **Utilities & Tools**:
- `tools`: Various CLI tools and utilities
- `sdk`: SDKs for different languages and platforms
- `documentation`: Documentation generation and management
## Packet System
Nym uses a modified Sphinx packet format for its mixnet:
1. **Message Chunking**:
- Messages are divided into "sets" and "fragments"
- Each fragment fits in a single Sphinx packet
- The `common/nymsphinx/chunking` module handles message fragmentation
2. **Routing**:
- Packets traverse through 3 layers of mixnodes
- Routing information is encrypted in layers (onion routing)
- The final gateway receives and processes the messages
3. **Monitoring**:
- Monitoring system tracks packet delivery through the network
- Routes are analyzed for reliability statistics
- Node performance metrics are collected
## Network Protocol
Nym implements the Loopix mixnet design with several key privacy features:
1. **Continuous-time Mixing**:
- Each mixnode delays messages independently with an exponential distribution
- This creates random reordering of packets, destroying timing correlations
- Offers better anonymity properties than batch mixing approaches
2. **Cover Traffic**:
- Clients and nodes generate dummy "loop" packets that circulate through the network
- These packets are indistinguishable from real traffic
- Creates a baseline level of traffic that hides actual communication patterns
- Provides unobservability (hiding when and how much real traffic is being sent)
3. **Stratified Network Architecture**:
- Traffic flows through Entry Gateway → 3 Mixnode Layers → Exit Gateway
- Path selection is independent per-message (unlike Tor)
- Each node connects only to adjacent layers
4. **Anonymous Replies**:
- Single-Use Reply Blocks (SURBs) allow receiving messages without revealing identity
- Enables bidirectional communication while maintaining privacy
## Network Monitoring Architecture
The network monitoring system is a core component that measures mixnet reliability:
1. The `nym-network-monitor` sends test packets through the network
2. These packets follow predefined routes through multiple mixnodes
3. Metrics are collected about:
- Successful and failed packet deliveries
- Node reliability (percentage of successful packet handling)
- Route reliability (which specific route combinations work best)
4. Results are stored in the database and used by `nym-api` to:
- Present node performance statistics
- Determine network rewards
- Provide route selection guidance to clients
In the current branch, metrics collection is being enhanced with a fanout approach to submit to multiple API endpoints.
## Development Environment
### Required Dependencies
- Rust toolchain (stable, 1.80+)
- Node.js (v20+) and yarn for TypeScript components
- SQLite for local database development
- PostgreSQL for API database (optional, for full API functionality)
- CosmWasm tools for contract development
- For building contracts: `wasm-opt` tool from `binaryen`
- Python 3.8+ for some scripts
- Docker (optional, for containerized development)
- protoc (Protocol Buffers compiler) for some components
### Environment Configurations
The `envs/` directory contains pre-configured environments:
#### Available Environments
- **`local.env`**: Local development environment
- Points to local services (localhost)
- Uses test mnemonics and keys
- Ideal for testing without external dependencies
- **`sandbox.env`**: Sandbox test network
- Public test network with real nodes
- Test tokens available from faucet
- Contract addresses for sandbox deployment
- API: https://sandbox-nym-api1.nymtech.net
- **`mainnet.env`**: Production mainnet
- Real network with real tokens
- Production contract addresses
- API: https://validator.nymtech.net
- Use with caution!
- **`canary.env`**: Canary deployment
- Pre-release testing environment
- Tests new features before mainnet
- **`mainnet-local-api.env`**: Hybrid environment
- Uses mainnet contracts but local API
- Useful for API development against mainnet data
#### Key Environment Variables
```bash
# Network configuration
NETWORK_NAME=sandbox # Network identifier
BECH32_PREFIX=n # Address prefix (n for sandbox, n for mainnet)
NYM_API=https://sandbox-nym-api1.nymtech.net/api
NYXD=https://rpc.sandbox.nymtech.net
NYM_API_NETWORK=sandbox
# Contract addresses (network-specific)
MIXNET_CONTRACT_ADDRESS=n1xr3rq8yvd7qplsw5yx90ftsr2zdhg4e9z60h5duusgxpv72hud3sjkxkav
VESTING_CONTRACT_ADDRESS=n1unyuj8qnmygvzuex3dwmg9yzt9alhvyeat0uu0jedg2wj33efl5qackslz
# ... other contract addresses
# Mnemonic for testing (NEVER use in production)
MNEMONIC="clutch captain shoe salt awake harvest setup primary inmate ugly among become"
# API Keys and tokens
IPINFO_API_TOKEN=your_token_here
AUTHENTICATOR_PASSWORD=password_here
# Logging
RUST_LOG=info # Options: error, warn, info, debug, trace
RUST_BACKTRACE=1 # Enable backtraces
# Database
DATABASE_URL=postgresql://user:pass@localhost/nym_api
```
#### Using Environment Files
```bash
# Load environment and run command
dotenv -f envs/sandbox.env -- cargo run -p nym-api
# Export to shell
source envs/sandbox.env
# Use with make targets
dotenv -f envs/sandbox.env -- make run-api-tests
```
## Initial Setup
### First Time Setup
1. **Install Prerequisites**
```bash
# Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
# Install Node.js and yarn
# Via nvm (recommended):
curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh | bash
nvm install 20
npm install -g yarn
# Install build tools
# Ubuntu/Debian:
sudo apt-get install build-essential pkg-config libssl-dev protobuf-compiler libpq-dev
# macOS:
brew install protobuf postgresql
# Install wasm-opt for contract builds
npm install -g wasm-opt
# Add wasm target for Rust
rustup target add wasm32-unknown-unknown
```
2. **Clone and Setup Repository**
```bash
git clone https://github.com/nymtech/nym.git
cd nym/nym
# Install JavaScript dependencies
yarn install
# Build the project
make build
```
3. **Database Setup (Optional, for API development)**
```bash
# Install PostgreSQL
# Create database
createdb nym_api
# Run migrations (from nym-api directory)
cd nym-api
sqlx migrate run
```
### Quick Start
```bash
# Run a mixnode locally
dotenv -f envs/sandbox.env -- cargo run -p nym-node -- run --mode mixnode --id my-mixnode
# Run a gateway locally
dotenv -f envs/sandbox.env -- cargo run -p nym-node -- run --mode gateway --id my-gateway
# Run the API server
dotenv -f envs/sandbox.env -- cargo run -p nym-api
# Run a client
cargo run -p nym-client -- init --id my-client
cargo run -p nym-client -- run --id my-client
```
## CI/CD Pipeline
The project uses GitHub Actions for CI/CD with several key workflows:
1. **Build and Test**:
- `ci-build.yml`: Main build workflow for Rust components
- Tests are run on multiple platforms (Linux, Windows, macOS)
- Includes formatting check (rustfmt) and linting (clippy)
2. **Release Process**:
- Binary artifacts are published on release tags
- Multiple platform builds are created
3. **Documentation**:
- Documentation is automatically built and deployed
## Database Structure
The system uses SQLite databases with tables like:
- `mixnode_status`: Status information about mixnodes
- `gateway_status`: Status information about gateways
- `routes`: Route performance information (success/failure of specific paths)
- `monitor_run`: Information about monitoring test runs
## Development Workflows
### Running a Node
To run the mixnode or gateway:
```bash
# Run nym-node as a mixnode with specified identity
cargo run -p nym-node -- run --mode mixnode --id my-mixnode
# Run nym-node as a gateway
cargo run -p nym-node -- run --mode gateway --id my-gateway
```
### Configuration
Nodes can be configured with files in various locations:
- Command-line arguments
- Environment variables
- `.env` files specified with `--config-env-file`
### Monitoring
To monitor the health of your node:
- View logs for real-time information
- Use the node's HTTP API for status information
- Check the explorer for public node statistics
## Common Libraries
- `common/types`: Shared data types across all components
- `common/crypto`: Cryptographic primitives and wrappers
- `common/client-core`: Core client functionality
- `common/gateway-client`: Client-gateway communication
- `common/task`: Task management and concurrency utilities
- `common/nymsphinx`: Sphinx packet implementation for mixnet
- `common/topology`: Network topology management
- `common/credentials`: Credential system for privacy-preserving authentication
- `common/bandwidth-controller`: Bandwidth management and accounting
## Code Conventions
- Error handling: Use anyhow/thiserror for structured error handling
- Logging: Use the tracing framework for logging and diagnostics
- State management: Generally use Tokio/futures for async code
- Configuration: Use the config crate and env vars with defaults
- Database: Use sqlx for type-safe database queries
- Follow clippy recommendations and rustfmt formatting
- Use semantic commit messages: feat, fix, docs, refactor, test, chore
## When Making Changes
- Run `make test` before submitting PRs
- Follow Rust naming conventions
- Use `clippy` to check for common issues
- Update SQLx query caches when modifying DB queries: `cargo sqlx prepare`
- Consider backward compatibility for protocol changes
- Use lefthook pre-commit hooks for TypeScript formatting
- Run `cargo deny check` to verify dependency compliance
- Test against both sandbox and local environments when possible
- Update relevant documentation and CHANGELOG.md
## Development Tools
### Useful Cargo Commands
```bash
# Check for outdated dependencies
cargo outdated
# Analyze binary size
cargo bloat --release -p nym-node
# Generate dependency graph
cargo tree -p nym-api
# Run with instrumentation
cargo run --features profiling -p nym-node
# Check for security advisories
cargo audit
```
### Database Tools
```bash
# SQLx CLI for migrations
cargo install sqlx-cli
# Create new migration
cd nym-api && sqlx migrate add <migration_name>
# Prepare query metadata for offline compilation
cargo sqlx prepare --workspace
# View database schema
./nym-api/enter_db.sh
```
### Development Scripts
- `scripts/build_topology.py`: Generate network topology files
- `scripts/node_api_check.py`: Verify node API endpoints
- `scripts/network_tunnel_manager.sh`: Manage network tunnels
- `scripts/localnet_start.sh`: Start a local test network
- Various deployment scripts in `deployment/` for different environments
## Debugging
- Enable more verbose logging with the RUST_LOG environment variable:
```
RUST_LOG=debug,nym_node=trace cargo run -p nym-node -- run --mode mixnode
```
- Use the HTTP API endpoints for status information
- Check monitoring data in the database for network performance metrics
- For complex issues, use tracing tools to follow packet flow
- Enable backtraces: `RUST_BACKTRACE=full`
- For WASM debugging: Use browser developer tools with source maps
## Deployment and Advanced Configurations
### Deployment Structure
The `deployment/` directory contains Ansible playbooks and configurations for various deployment scenarios:
- **`aws/`**: AWS-specific deployment configurations
- **`mixnode/`**: Mixnode deployment playbooks
- **`gateway/`**: Gateway deployment playbooks
- **`validator/`**: Validator node deployment
- **`sandbox-v2/`**: Complete sandbox environment setup
- **`big-dipper-2/`**: Block explorer deployment
### Sandbox V2 Deployment
The sandbox-v2 deployment (`deployment/sandbox-v2/`) provides a complete test environment:
```bash
# Key playbooks:
- deploy.yaml # Main deployment orchestrator
- deploy-mixnodes.yaml # Deploy mixnodes
- deploy-gateways.yaml # Deploy gateways
- deploy-validators.yaml # Deploy validator nodes
- deploy-nym-api.yaml # Deploy API services
```
### Custom Environment Setup
To create a custom environment:
1. Copy an existing env file: `cp envs/sandbox.env envs/custom.env`
2. Modify the network endpoints and contract addresses
3. Update the `NETWORK_NAME` to your identifier
4. Set appropriate mnemonics and keys (use fresh ones for production!)
### Contract Addresses
Contract addresses are network-specific and defined in environment files:
- Mixnet contract: Manages mixnode/gateway registry
- Vesting contract: Handles token vesting schedules
- Coconut contracts: Privacy-preserving credentials
- Name service: Human-readable address mapping
- Ecash contract: Electronic cash functionality
### Local Network Setup
For a completely local network:
```bash
# Start local chain
./scripts/localnet_start.sh
# Deploy contracts
cd contracts
make deploy-local
# Start nodes with local config
dotenv -f envs/local.env -- cargo run -p nym-node -- run --mode mixnode
```
## Common Issues and Troubleshooting
### Database Issues
- When modifying database queries, you must update SQLx query caches:
```bash
cargo sqlx prepare
```
- If you see SQLx errors about missing query files, this is likely the cause
- For "database is locked" errors with SQLite, ensure only one process accesses the DB
- For PostgreSQL connection issues, verify DATABASE_URL and that the server is running
### API Connection Issues
- Check the environment variables pointing to the APIs (NYM_API, NYXD)
- Verify network connectivity and API health endpoints
- For authentication issues, check node keys and credentials
- Common endpoints to verify:
- API health: `$NYM_API/health`
- Chain status: `$NYXD/status`
- Contract info: `$NYXD/cosmwasm/wasm/v1/contract/$CONTRACT_ADDRESS`
### Build Problems
- Clean dependencies with `cargo clean` for a fresh build
- Check for compatible Rust version (1.80+ recommended)
- For smart contract builds, ensure wasm-opt is installed: `npm install -g wasm-opt`
- For cross-compilation issues, check target-specific dependencies
- WASM build issues: Ensure wasm32-unknown-unknown target is installed:
```bash
rustup target add wasm32-unknown-unknown
```
- For "cannot find -lpq" errors, install PostgreSQL development files:
```bash
# Ubuntu/Debian
sudo apt-get install libpq-dev
# macOS
brew install postgresql
```
### Environment Issues
- Contract address mismatches: Ensure you're using the correct environment file
- "Account sequence mismatch": The account nonce is out of sync, wait and retry
- Token decimal issues: Sandbox uses different decimal places than mainnet
- API version mismatches: Ensure your local API version matches the network
- "Insufficient funds": Get test tokens from faucet (sandbox) or check balance
- Gateway/mixnode bonding issues: Verify minimum stake requirements
## Working with Routes and Monitoring
1. Route monitoring metrics are stored in a `routes` table with:
- Layer node IDs (layer1, layer2, layer3, gw)
- Success flag (boolean)
- Timestamp
2. To analyze routes:
- Check `NetworkAccount` and `AccountingRoute` in `nym-network-monitor/src/accounting.rs`
- View monitoring logic in `common/nymsphinx/chunking/monitoring.rs`
- Observe how routes are submitted to the database in the `submit_accounting_routes_to_db` function
## Performance Optimization
### Profiling and Benchmarking
```bash
# Run benchmarks
cargo bench -p nym-node
# Profile with perf (Linux)
cargo build --release --features profiling
perf record --call-graph=dwarf ./target/release/nym-node run --mode mixnode
perf report
# Generate flamegraph
cargo install flamegraph
cargo flamegraph --bin nym-node -- run --mode mixnode
```
### Common Performance Considerations
- Use bounded channels for backpressure
- Batch database operations where possible
- Monitor memory usage with `RUST_LOG=nym_node::metrics=debug`
- Use connection pooling for database connections
- Consider using `jemalloc` for better memory allocation performance
Generated
+2286 -1569
View File
File diff suppressed because it is too large Load Diff
+5 -13
View File
@@ -39,8 +39,7 @@ members = [
"common/cosmwasm-smart-contracts/ecash-contract",
"common/cosmwasm-smart-contracts/group-contract",
"common/cosmwasm-smart-contracts/mixnet-contract",
"common/cosmwasm-smart-contracts/multisig-contract",
"common/cosmwasm-smart-contracts/nym-performance-contract",
"common/cosmwasm-smart-contracts/multisig-contract", "common/cosmwasm-smart-contracts/nym-performance-contract",
"common/cosmwasm-smart-contracts/nym-pool-contract",
"common/cosmwasm-smart-contracts/vesting-contract",
"common/credential-storage",
@@ -50,8 +49,6 @@ members = [
"common/credentials-interface",
"common/crypto",
"common/dkg",
"common/ecash-signer-check",
"common/ecash-signer-check-types",
"common/ecash-time",
"common/execute",
"common/exit-policy",
@@ -92,7 +89,7 @@ members = [
"common/socks5/requests",
"common/statistics",
"common/store-cipher",
"common/task", "common/test-utils",
"common/task",
"common/ticketbooks-merkle",
"common/topology",
"common/tun",
@@ -103,7 +100,6 @@ members = [
"common/wasm/utils",
"common/wireguard",
"common/wireguard-types",
"common/zulip-client",
"documentation/autodoc",
"gateway",
"nym-api",
@@ -222,7 +218,7 @@ clap_complete_fig = "4.5"
colored = "2.2"
comfy-table = "7.1.4"
console = "0.15.11"
console-subscriber = "0.4.1"
console-subscriber = "0.1.1"
console_error_panic_hook = "0.1"
const-str = "0.5.6"
const_format = "0.2.34"
@@ -238,7 +234,6 @@ digest = "0.10.7"
dirs = "5.0"
doc-comment = "0.3"
dotenvy = "0.15.6"
dyn-clone = "1.0.19"
ecdsa = "0.16"
ed25519-dalek = "2.1"
encoding_rs = "0.8.35"
@@ -321,8 +316,8 @@ si-scale = "0.2.3"
snow = "0.9.6"
sphinx-packet = "=0.6.0"
sqlx = "0.8.6"
strum = "0.27.2"
strum_macros = "0.27.2"
strum = "0.26"
strum_macros = "0.26"
subtle-encoding = "0.5"
syn = "1"
sysinfo = "0.33.0"
@@ -439,9 +434,6 @@ opt-level = 'z'
# lto = true
opt-level = 'z'
[workspace.lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
[workspace.lints.clippy]
unwrap_used = "deny"
expect_used = "deny"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-client"
version = "1.1.61"
version = "1.1.58"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
description = "Implementation of the Nym Client"
edition = "2021"
+1 -1
View File
@@ -111,7 +111,7 @@ impl SocketClient {
let dkg_query_client = if self.config.base.client.disabled_credentials_mode {
None
} else {
Some(default_query_dkg_client_from_config(&self.config.base)?)
Some(default_query_dkg_client_from_config(&self.config.base))
};
let storage = self.initialise_storage().await?;
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-socks5-client"
version = "1.1.61"
version = "1.1.58"
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"
@@ -28,6 +28,8 @@ pub type HmacSha256 = Hmac<Sha256>;
pub type Nonce = u64;
pub type Taken = Option<SystemTime>;
pub const BANDWIDTH_CAP_PER_DAY: u64 = 250 * 1024 * 1024 * 1024; // 250 GB
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct IpPair {
pub ipv4: Ipv4Addr,
+2 -2
View File
@@ -207,7 +207,7 @@ where
<St as Storage>::StorageError: Send + Sync + 'static,
{
if let Some(stored) = storage
.get_expiration_date_signatures(expiration_date, epoch_id)
.get_expiration_date_signatures(expiration_date)
.await
.map_err(BandwidthControllerError::credential_storage_error)?
{
@@ -220,7 +220,7 @@ where
ecash_apis,
|api| async move {
api.api_client
.global_expiration_date_signatures(Some(expiration_date), Some(epoch_id))
.global_expiration_date_signatures(Some(expiration_date))
.await
},
format!("aggregated coin index signatures for date {expiration_date}"),
-4
View File
@@ -13,7 +13,6 @@ async-trait = { workspace = true }
base64 = { workspace = true }
bs58 = { workspace = true }
clap = { workspace = true, optional = true }
cfg-if = { workspace = true }
comfy-table = { workspace = true, optional = true }
futures = { workspace = true }
humantime = { workspace = true }
@@ -124,6 +123,3 @@ fs-surb-storage = ["nym-client-core-surb-storage/fs-surb-storage"]
fs-gateways-storage = ["nym-client-core-gateways-storage/fs-gateways-storage"]
wasm = ["nym-gateway-client/wasm"]
metrics-server = []
[lints]
workspace = true
@@ -58,7 +58,6 @@ where
Some(data) => data,
None => {
// SAFETY: one of those arguments must have been set
#[allow(clippy::unwrap_used)]
fs::read(common_args.signatures_path.unwrap())?
}
};
@@ -64,7 +64,6 @@ where
Some(data) => data,
None => {
// SAFETY: one of those arguments must have been set
#[allow(clippy::unwrap_used)]
fs::read(common_args.credential_path.unwrap())?
}
};
@@ -58,7 +58,6 @@ where
Some(data) => data,
None => {
// SAFETY: one of those arguments must have been set
#[allow(clippy::unwrap_used)]
fs::read(common_args.signatures_path.unwrap())?
}
};
@@ -58,7 +58,6 @@ where
Some(data) => data,
None => {
// SAFETY: one of those arguments must have been set
#[allow(clippy::unwrap_used)]
fs::read(common_args.key_path.unwrap())?
}
};
@@ -135,11 +135,9 @@ pub enum ClientInputStatus {
}
impl ClientInputStatus {
#[allow(clippy::panic)]
pub fn register_producer(&mut self) -> ClientInput {
match std::mem::replace(self, ClientInputStatus::Connected) {
ClientInputStatus::AwaitingProducer { client_input } => client_input,
// critical failure implying misuse of software
ClientInputStatus::Connected => panic!("producer was already registered before"),
}
}
@@ -151,11 +149,9 @@ pub enum ClientOutputStatus {
}
impl ClientOutputStatus {
#[allow(clippy::panic)]
pub fn register_consumer(&mut self) -> ClientOutput {
match std::mem::replace(self, ClientOutputStatus::Connected) {
ClientOutputStatus::AwaitingConsumer { client_output } => client_output,
// critical failure implying misuse of software
ClientOutputStatus::Connected => panic!("consumer was already registered before"),
}
}
@@ -711,14 +707,11 @@ where
})?;
let store_clone = mem_store.clone();
spawn_future!(
async move {
persistent_storage
.flush_on_shutdown(store_clone, shutdown)
.await
},
"PersistentReplyStorage::flush_on_shutdown"
);
spawn_future(async move {
persistent_storage
.flush_on_shutdown(store_clone, shutdown)
.await
});
Ok(mem_store)
}
@@ -739,7 +732,7 @@ where
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)?
.map_err(|_| ClientCoreError::HkdfDerivationError {})?
} else {
ClientKeys::generate_new(&mut rng)
};
@@ -114,32 +114,41 @@ pub async fn setup_fs_gateways_storage<P: AsRef<Path>>(
})
}
pub fn create_bandwidth_controller_with_urls<St: CredentialStorage>(
nyxd_url: Url,
storage: St,
) -> Result<BandwidthController<QueryHttpRpcNyxdClient, St>, ClientCoreError> {
let client = default_query_dkg_client(nyxd_url)?;
Ok(BandwidthController::new(storage, client))
}
pub fn default_query_dkg_client_from_config(
pub fn create_bandwidth_controller<St: CredentialStorage>(
config: &Config,
) -> Result<QueryHttpRpcNyxdClient, ClientCoreError> {
storage: St,
) -> BandwidthController<QueryHttpRpcNyxdClient, St> {
let nyxd_url = config
.get_validator_endpoints()
.pop()
.ok_or(ClientCoreError::RpcClientMissingUrl)?;
.expect("No nyxd validator endpoint provided");
create_bandwidth_controller_with_urls(nyxd_url, storage)
}
pub fn create_bandwidth_controller_with_urls<St: CredentialStorage>(
nyxd_url: Url,
storage: St,
) -> BandwidthController<QueryHttpRpcNyxdClient, St> {
let client = default_query_dkg_client(nyxd_url);
BandwidthController::new(storage, client)
}
pub fn default_query_dkg_client_from_config(config: &Config) -> QueryHttpRpcNyxdClient {
let nyxd_url = config
.get_validator_endpoints()
.pop()
.expect("No nyxd validator endpoint provided");
default_query_dkg_client(nyxd_url)
}
pub fn default_query_dkg_client(nyxd_url: Url) -> Result<QueryHttpRpcNyxdClient, ClientCoreError> {
pub fn default_query_dkg_client(nyxd_url: Url) -> QueryHttpRpcNyxdClient {
let details = nym_network_defaults::NymNetworkDetails::new_from_env();
let client_config = nyxd::Config::try_from_nym_network_details(&details)
.map_err(|source| ClientCoreError::InvalidNetworkDetails { source })?;
.expect("failed to construct validator client config");
// overwrite env configuration with config URLs
QueryHttpRpcNyxdClient::connect(client_config, nyxd_url.as_str())
.map_err(|source| ClientCoreError::RpcClientCreationFailure { source })
.expect("Could not construct query client")
}
@@ -235,7 +235,6 @@ impl LoopCoverTrafficStream<OsRng> {
tokio::task::yield_now().await;
}
#[allow(clippy::panic)]
pub fn start(mut self) {
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
@@ -252,30 +251,27 @@ impl LoopCoverTrafficStream<OsRng> {
let mut shutdown = self.task_client.fork("select");
spawn_future!(
async move {
debug!("Started LoopCoverTrafficStream with graceful shutdown support");
spawn_future(async move {
debug!("Started LoopCoverTrafficStream with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
tracing::trace!("LoopCoverTrafficStream: Received shutdown");
}
next = self.next() => {
if next.is_some() {
self.on_new_message().await;
} else {
tracing::trace!("LoopCoverTrafficStream: Stopping since channel closed");
break;
}
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
tracing::trace!("LoopCoverTrafficStream: Received shutdown");
}
next = self.next() => {
if next.is_some() {
self.on_new_message().await;
} else {
tracing::trace!("LoopCoverTrafficStream: Stopping since channel closed");
break;
}
}
}
shutdown.recv_timeout().await;
tracing::debug!("LoopCoverTrafficStream: Exiting");
},
"LoopCoverTrafficStream"
)
}
shutdown.recv_timeout().await;
tracing::debug!("LoopCoverTrafficStream: Exiting");
})
}
}
@@ -96,93 +96,72 @@ impl MixTrafficController {
mut mix_packets: Vec<MixPacket>,
) -> Result<(), ErasedGatewayError> {
debug_assert!(!mix_packets.is_empty());
let send_future = if mix_packets.len() == 1 {
// SAFETY: we just checked we have one packet
#[allow(clippy::unwrap_used)]
let result = if mix_packets.len() == 1 {
let mix_packet = mix_packets.pop().unwrap();
self.gateway_transceiver.send_mix_packet(mix_packet)
self.gateway_transceiver.send_mix_packet(mix_packet).await
} else {
self.gateway_transceiver.batch_send_mix_packets(mix_packets)
self.gateway_transceiver
.batch_send_mix_packets(mix_packets)
.await
};
tokio::select! {
biased;
_ = self.task_client.recv() => {
trace!("received shutdown while handling messages");
Ok(())
}
result = send_future => {
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;
}
result
}
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;
}
}
async fn on_client_request(&mut self, client_request: ClientRequest) {
tokio::select! {
biased;
_ = self.task_client.recv() => {
trace!("received shutdown while handling client request");
}
result = self.gateway_transceiver.send_client_request(client_request) => {
if let Err(err) = result {
error!("Failed to send client request: {err}")
}
}
}
result
}
pub fn start(mut self) {
spawn_future!(
async move {
debug!("Started MixTrafficController with graceful shutdown support");
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.task_client.recv() => {
tracing::trace!("MixTrafficController: Received shutdown");
spawn_future(async move {
debug!("Started MixTrafficController with graceful shutdown support");
while !self.task_client.is_shutdown() {
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;
}
}
},
None => {
tracing::trace!("MixTrafficController: Stopping since channel closed");
break;
}
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;
}
}
},
None => {
tracing::trace!("MixTrafficController: Stopping since channel closed");
break;
}
},
client_request = self.client_rx.recv() => match client_request {
Some(client_request) => {
self.on_client_request(client_request).await;
},
None => {
tracing::trace!("MixTrafficController, client request channel closed");
}
},
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 => {
tracing::trace!("MixTrafficController, client request channel closed");
}
},
_ = self.task_client.recv() => {
tracing::trace!("MixTrafficController: Received shutdown");
break;
}
}
self.task_client.recv_timeout().await;
tracing::debug!("MixTrafficController: Exiting");
},
"MixTrafficController"
);
}
self.task_client.recv_timeout().await;
tracing::debug!("MixTrafficController: Exiting");
});
}
}
@@ -269,8 +269,6 @@ pub struct MockGateway {
}
impl Default for MockGateway {
// test code
#[allow(clippy::unwrap_used)]
fn default() -> Self {
MockGateway {
dummy_identity: "3ebjp1Fb9hdcS1AR6AZihgeJiMHkB5jjJUsvqNnfQwU7"
@@ -194,11 +194,10 @@ impl ActionController {
trace!("{frag_id} is updating its delay");
// TODO: is it possible to solve this without either locking or temporarily removing the value?
if let Some((pending_ack_data, queue_key)) = self.pending_acks_data.remove(&frag_id) {
// SAFETY: this Action is triggered by `RetransmissionRequestListener` (for 'normal' packets)
// this Action is triggered by `RetransmissionRequestListener` (for 'normal' packets)
// or `ReplyController` (for 'reply' packets) which held the other potential
// reference to this Arc. HOWEVER, before the Action was pushed onto the queue, the reference
// was dropped hence this unwrap is safe.
#[allow(clippy::unwrap_used)]
let mut inner_data = Arc::try_unwrap(pending_ack_data).unwrap();
inner_data.update_retransmitted(delay);
@@ -210,7 +209,6 @@ impl ActionController {
}
// note: when the entry expires it's automatically removed from pending_acks_timers
#[allow(clippy::panic)]
fn handle_expired_ack_timer(&mut self, expired_ack: Expired<FragmentIdentifier>) {
let frag_id = expired_ack.into_inner();
@@ -120,7 +120,6 @@ where
}
}
#[allow(clippy::panic)]
async fn on_input_message(&mut self, msg: InputMessage) {
match msg {
InputMessage::Regular {
@@ -214,9 +213,7 @@ where
self.handle_premade_packets(msgs, lane).await
}
// MessageWrappers can't be nested
InputMessage::MessageWrapper { .. } => {
panic!("attempted to use nested MessageWrapper")
}
InputMessage::MessageWrapper { .. } => unimplemented!(),
},
};
}
@@ -226,11 +223,6 @@ where
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.task_client.recv() => {
tracing::trace!("InputMessageListener: Received shutdown");
break;
}
input_msg = self.input_receiver.recv() => match input_msg {
Some(input_msg) => {
self.on_input_message(input_msg).await;
@@ -240,7 +232,9 @@ where
break;
}
},
_ = self.task_client.recv() => {
tracing::trace!("InputMessageListener: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
@@ -298,44 +298,29 @@ where
let mut sent_notification_listener = self.sent_notification_listener;
let mut action_controller = self.action_controller;
spawn_future!(
async move {
acknowledgement_listener.run().await;
debug!("The acknowledgement listener has finished execution!");
},
"AcknowledgementController::AcknowledgementListener"
);
spawn_future(async move {
acknowledgement_listener.run().await;
debug!("The acknowledgement listener has finished execution!");
});
spawn_future!(
async move {
input_message_listener.run().await;
debug!("The input listener has finished execution!");
},
"AcknowledgementController::InputMessageListener"
);
spawn_future(async move {
input_message_listener.run().await;
debug!("The input listener has finished execution!");
});
spawn_future!(
async move {
retransmission_request_listener.run(packet_type).await;
debug!("The retransmission request listener has finished execution!");
},
"AcknowledgementController::RetransmissionRequestListener"
);
spawn_future(async move {
retransmission_request_listener.run(packet_type).await;
debug!("The retransmission request listener has finished execution!");
});
spawn_future!(
async move {
sent_notification_listener.run().await;
debug!("The sent notification listener has finished execution!");
},
"AcknowledgementController::SentNotificationListener"
);
spawn_future(async move {
sent_notification_listener.run().await;
debug!("The sent notification listener has finished execution!");
});
spawn_future!(
async move {
action_controller.run().await;
debug!("The controller has finished execution!");
},
"AcknowledgementController::ActionController"
);
spawn_future(async move {
action_controller.run().await;
debug!("The controller has finished execution!");
});
}
}
@@ -179,11 +179,6 @@ where
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.task_client.recv() => {
tracing::trace!("RetransmissionRequestListener: Received shutdown");
break;
}
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,
None => {
@@ -191,7 +186,9 @@ where
break;
}
},
_ = self.task_client.recv() => {
tracing::trace!("RetransmissionRequestListener: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
@@ -35,9 +35,6 @@ pub enum PreparationError {
#[error(transparent)]
NymTopologyError(#[from] NymTopologyError),
#[error("message wasn't split into any fragments!")]
EmptyFragments,
#[error("message too long for a single SURB, splitting into {fragments} fragments.")]
MessageTooLongForSingleSurb { fragments: usize },
@@ -323,16 +320,6 @@ where
});
}
if fragment.is_empty() {
error!("CRITICAL FAILURE: our split message didn't result in any sendable fragments");
return Err(SurbWrappedPreparationError {
source: PreparationError::EmptyFragments,
returned_surbs: Some(vec![reply_surb]),
});
}
// SAFETY: we just checked we have one fragment
#[allow(clippy::unwrap_used)]
let chunk = fragment.pop().unwrap();
let chunk_clone = chunk.clone();
let prepared_fragment = self
@@ -548,7 +535,6 @@ where
pending_acks.push(pending_ack);
}
drop(topology_permit);
self.insert_pending_acks(pending_acks);
self.forward_messages(real_messages, lane).await;
@@ -671,7 +657,6 @@ where
.zip(reply_surbs.into_iter())
.map(|(fragment, reply_surb)| {
// unwrap here is fine as we know we have a valid topology
#[allow(clippy::unwrap_used)]
self.message_preparer
.prepare_reply_chunk_for_sending(
fragment,
@@ -731,21 +716,17 @@ where
// tells real message sender (with the poisson timer) to send this to the mix network
pub(crate) async fn forward_messages(
&mut self,
&self,
messages: Vec<RealMessage>,
transmission_lane: TransmissionLane,
) {
tokio::select! {
biased;
_ = self.task_client.recv() => {
trace!("received shutdown while attempting to forward mixnet messages");
}
sending_res = self.real_message_sender.send((messages, transmission_lane)) => {
if sending_res.is_err() {
error!(
"failed to forward mixnet messages due to closed channel (outside of shutdown!)"
);
}
if let Err(err) = 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}");
}
}
}
@@ -224,20 +224,14 @@ impl RealMessagesController<OsRng> {
let ack_control = self.ack_control;
let mut reply_control = self.reply_control;
spawn_future!(
async move {
out_queue_control.run().await;
debug!("The out queue controller has finished execution!");
},
"RealMessagesController::OutQueueControl)"
);
spawn_future!(
async move {
reply_control.run().await;
debug!("The reply controller has finished execution!");
},
"RealMessagesController::ReplyController"
);
spawn_future(async move {
out_queue_control.run().await;
debug!("The out queue controller has finished execution!");
});
spawn_future(async move {
reply_control.run().await;
debug!("The reply controller has finished execution!");
});
ack_control.start(packet_type);
}
@@ -249,8 +249,6 @@ where
}
};
// SAFETY: our topology must be valid at this point
#[allow(clippy::expect_used)]
(
generate_loop_cover_packet(
&mut self.rng,
@@ -280,33 +278,17 @@ where
}
};
let sending_res = tokio::select! {
biased;
_ = self.task_client.recv() => {
trace!("received shutdown signal while attempting to send mix message");
return
}
sending_res = self.mix_tx.send(vec![next_message]) => {
sending_res
}
};
match sending_res {
Err(_) => {
if !self.task_client.is_shutdown_poll() {
tracing::error!(
"failed to send mixnet packet due to closed channel (outside of shutdown!)"
);
}
}
Ok(_) => {
let event = if fragment_id.is_some() {
PacketStatisticsEvent::RealPacketSent(packet_size)
} else {
PacketStatisticsEvent::CoverPacketSent(packet_size)
};
self.stats_tx.report(event.into());
if let Err(err) = self.mix_tx.send(vec![next_message]).await {
if !self.task_client.is_shutdown_poll() {
tracing::error!("Failed to send: {err}");
}
} else {
let event = if fragment_id.is_some() {
PacketStatisticsEvent::RealPacketSent(packet_size)
} else {
PacketStatisticsEvent::CoverPacketSent(packet_size)
};
self.stats_tx.report(event.into());
}
// notify ack controller about sending our message only after we actually managed to push it
@@ -457,8 +439,6 @@ where
tracing::trace!("handling real_messages: size: {}", real_messages.len());
self.transmission_buffer.store(&conn_id, real_messages);
// SAFETY: we just stored the message
#[allow(clippy::expect_used)]
let real_next = self.pop_next_message().expect("Just stored one");
Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
@@ -507,8 +487,6 @@ where
// First store what we got for the given connection id
self.transmission_buffer.store(&conn_id, real_messages);
// SAFETY: we just stored the message
#[allow(clippy::expect_used)]
let real_next = self.pop_next_message().expect("we just added one");
Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
@@ -198,7 +198,6 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
}
}
#[allow(clippy::panic)]
async fn disconnect_sender(&mut self) {
let mut guard = self.inner.lock().await;
if guard.message_sender.is_none() {
@@ -209,7 +208,6 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
guard.message_sender = None;
}
#[allow(clippy::panic)]
async fn connect_sender(&mut self, sender: ReconstructedMessagesSender) {
let mut guard = self.inner.lock().await;
if guard.message_sender.is_some() {
@@ -601,20 +599,14 @@ impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferControll
let mut fragmented_message_receiver = self.fragmented_message_receiver;
let mut request_receiver = self.request_receiver;
spawn_future!(
async move {
match fragmented_message_receiver.run().await {
Ok(_) => {}
Err(e) => error!("{e}"),
}
},
"ReceivedMessagesBufferController::FragmentedMessageReceiver"
);
spawn_future!(
async move {
request_receiver.run().await;
},
"ReceivedMessagesBufferController::RequestReceiver"
);
spawn_future(async move {
match fragmented_message_receiver.run().await {
Ok(_) => {}
Err(e) => error!("{e}"),
}
});
spawn_future(async move {
request_receiver.run().await;
});
}
}
@@ -155,9 +155,8 @@ where
data: Vec<Arc<PendingAcknowledgement>>,
) {
trace!("re-inserting pending retransmissions for {recipient}");
// SAFETY: the underlying entry MUST exist as we've just got data from there
// the underlying entry MUST exist as we've just got data from there
// and we hold a mut reference
#[allow(clippy::expect_used)]
let map_entry = &mut self
.surb_senders
.get_mut(recipient)
@@ -430,7 +429,6 @@ where
.pop_at_most_n_next_messages_at_random(amount)
}
#[allow(clippy::panic)]
async fn try_clear_pending_queue(&mut self, target: AnonymousSenderTag) {
trace!("trying to clear pending queue");
let available_surbs = self.surbs_storage.available_surbs(&target);
@@ -165,12 +165,9 @@ impl StatisticsControl {
}
pub(crate) fn start(mut self) {
spawn_future!(
async move {
self.run().await;
},
"StatisticsControl"
)
spawn_future(async move {
self.run().await;
})
}
pub(crate) fn create_and_start(
@@ -126,7 +126,7 @@ impl TopologyAccessor {
.map(|p| p.topology.clone())
}
pub async fn current_route_provider(&self) -> Option<RwLockReadGuard<'_, NymRouteProvider>> {
pub async fn current_route_provider(&self) -> Option<RwLockReadGuard<NymRouteProvider>> {
let provider = self.inner.topology.read().await;
if provider.topology.is_empty() {
None
@@ -145,39 +145,36 @@ impl TopologyRefresher {
}
pub fn start(mut self) {
spawn_future!(
async move {
debug!("Started TopologyRefresher with graceful shutdown support");
spawn_future(async move {
debug!("Started TopologyRefresher with graceful shutdown support");
#[cfg(not(target_arch = "wasm32"))]
let mut interval = tokio_stream::wrappers::IntervalStream::new(
tokio::time::interval(self.refresh_rate),
);
#[cfg(not(target_arch = "wasm32"))]
let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(
self.refresh_rate,
));
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
// We already have an initial topology, so no need to refresh it immediately.
// My understanding is that js setInterval does not fire immediately, so it's not
// needed there.
#[cfg(not(target_arch = "wasm32"))]
interval.next().await;
// We already have an initial topology, so no need to refresh it immediately.
// My understanding is that js setInterval does not fire immediately, so it's not
// needed there.
#[cfg(not(target_arch = "wasm32"))]
interval.next().await;
while !self.task_client.is_shutdown() {
tokio::select! {
_ = interval.next() => {
self.try_refresh().await;
},
_ = self.task_client.recv() => {
tracing::trace!("TopologyRefresher: Received shutdown");
},
}
while !self.task_client.is_shutdown() {
tokio::select! {
_ = interval.next() => {
self.try_refresh().await;
},
_ = self.task_client.recv() => {
tracing::trace!("TopologyRefresher: Received shutdown");
},
}
self.task_client.recv_timeout().await;
tracing::debug!("TopologyRefresher: Exiting");
},
"TopologyRefresher"
)
}
self.task_client.recv_timeout().await;
tracing::debug!("TopologyRefresher: Exiting");
})
}
}
+1 -15
View File
@@ -7,9 +7,7 @@ use nym_gateway_client::error::GatewayClientError;
use nym_topology::node::RoutingNodeError;
use nym_topology::{NodeId, NymTopologyError};
use nym_validator_client::nym_api::error::NymAPIError;
use nym_validator_client::nyxd::error::NyxdError;
use nym_validator_client::ValidatorClientError;
use rand::distributions::WeightedError;
use std::error::Error;
use std::path::PathBuf;
@@ -232,19 +230,7 @@ pub enum ClientCoreError {
UnexpectedKeyUpgrade { gateway_id: String },
#[error("failed to derive keys from master key")]
HkdfDerivationError,
#[error("missing url for constructing RPC client")]
RpcClientMissingUrl,
#[error("provided nym network details were malformed: {source}")]
InvalidNetworkDetails { source: NyxdError },
#[error("failed to construct RPC client: {source}")]
RpcClientCreationFailure { source: NyxdError },
#[error("failed to select valid gateway due to incomputable latency")]
GatewaySelectionFailure { source: WeightedError },
HkdfDerivationError {},
}
impl From<tungstenite::Error> for ClientCoreError {
+2 -2
View File
@@ -148,7 +148,7 @@ async fn connect(endpoint: &str) -> Result<WsConn, ClientCoreError> {
JSWebsocket::new(endpoint).map_err(|_| ClientCoreError::GatewayJsConnectionFailure)
}
async fn measure_latency<G>(gateway: &G) -> Result<GatewayWithLatency<'_, G>, ClientCoreError>
async fn measure_latency<G>(gateway: &G) -> Result<GatewayWithLatency<G>, ClientCoreError>
where
G: ConnectableGateway,
{
@@ -245,7 +245,7 @@ pub async fn choose_gateway_by_latency<R: Rng, G: ConnectableGateway + Clone>(
let gateways_with_latency = gateways_with_latency.lock().await;
let chosen = gateways_with_latency
.choose_weighted(rng, |item| 1. / item.latency.as_secs_f32())
.map_err(|source| ClientCoreError::GatewaySelectionFailure { source })?;
.expect("invalid selection weight!");
info!(
"chose gateway {} with average latency of {:?}",
+2 -38
View File
@@ -18,54 +18,18 @@ pub use nym_topology::{
};
#[cfg(target_arch = "wasm32")]
pub fn spawn_future<F>(future: F)
pub(crate) fn spawn_future<F>(future: F)
where
F: Future<Output = ()> + 'static,
{
wasm_bindgen_futures::spawn_local(future);
}
// TODO: expose similar API to the rest of the codebase,
// perhaps with some simple trait for a task to define its name
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn spawn_future<F>(future: F)
pub(crate) fn spawn_future<F>(future: F)
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
tokio::spawn(future);
}
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn spawn_named_future<F>(future: F, name: &str)
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
cfg_if::cfg_if! {if #[cfg(tokio_unstable)] {
#[allow(clippy::expect_used)]
tokio::task::Builder::new().name(name).spawn(future).expect("failed to spawn future");
} else {
let _ = name;
tracing::debug!(r#"the underlying binary hasn't been built with `RUSTFLAGS="--cfg tokio_unstable"` - the future naming won't do anything"#);
spawn_future(future);
}}
}
#[macro_export]
macro_rules! spawn_future {
($future:expr) => {{
$crate::spawn_future($future)
}};
($future:expr, $name:expr) => {{
cfg_if::cfg_if! {if #[cfg(not(target_arch = "wasm32"))] {
$crate::spawn_named_future($future, $name)
} else {
let _ = $name;
$crate::spawn_future($future)
}}
}};
}
@@ -37,7 +37,7 @@ impl StorageManager {
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.synchronous(SqliteSynchronous::Normal)
.auto_vacuum(SqliteAutoVacuum::Incremental)
.filename(database_path)
.filename(&database_path)
.create_if_missing(fresh)
.disable_statement_logging();
@@ -49,7 +49,8 @@ impl StorageManager {
}
};
let connection_pool = SqlitePoolGuard::new(connection_pool);
let connection_pool =
SqlitePoolGuard::new(database_path.as_ref().to_path_buf(), connection_pool);
if let Err(err) = sqlx::migrate!("./fs_surbs_migrations")
.run(&*connection_pool)
@@ -201,7 +201,7 @@ impl<C, St> GatewayClient<C, St> {
#[cfg(not(target_arch = "wasm32"))]
pub async fn establish_connection(&mut self) -> Result<(), GatewayClientError> {
debug!(
"Attempting to establish connection to gateway at: {}",
"Attemting to establish connection to gateway at: {}",
self.gateway_address
);
let (ws_stream, _) = connect_async(
@@ -337,7 +337,7 @@ impl PartiallyDelegatedHandle {
// check if the split stream didn't error out
let receive_res = stream_receiver
.try_recv()
.map_err(|_| GatewayClientError::ConnectionAbruptlyClosed)?;
.expect("stream sender was somehow dropped without sending anything!");
if let Some(res) = receive_res {
let _res = res?;
@@ -719,11 +719,10 @@ impl NymApiClient {
pub async fn partial_expiration_date_signatures(
&self,
expiration_date: Option<Date>,
epoch_id: Option<EpochId>,
) -> Result<PartialExpirationDateSignatureResponse, ValidatorClientError> {
Ok(self
.nym_api
.partial_expiration_date_signatures(expiration_date, epoch_id)
.partial_expiration_date_signatures(expiration_date)
.await?)
}
@@ -740,11 +739,10 @@ impl NymApiClient {
pub async fn global_expiration_date_signatures(
&self,
expiration_date: Option<Date>,
epoch_id: Option<EpochId>,
) -> Result<AggregatedExpirationDateSignatureResponse, ValidatorClientError> {
Ok(self
.nym_api
.global_expiration_date_signatures(expiration_date, epoch_id)
.global_expiration_date_signatures(expiration_date)
.await?)
}
@@ -6,18 +6,16 @@ use crate::nym_api::routes::{ecash, CORE_STATUS_COUNT, SINCE_ARG};
use async_trait::async_trait;
use nym_api_requests::ecash::models::{
AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
BatchRedeemTicketsBody, EcashBatchTicketRedemptionResponse, EcashSignerStatusResponse,
EcashTicketVerificationResponse, IssuedTicketbooksChallengeCommitmentRequest,
IssuedTicketbooksChallengeCommitmentResponse, IssuedTicketbooksDataRequest,
IssuedTicketbooksDataResponse, IssuedTicketbooksForCountResponse, IssuedTicketbooksForResponse,
VerifyEcashTicketBody,
BatchRedeemTicketsBody, EcashBatchTicketRedemptionResponse, EcashTicketVerificationResponse,
IssuedTicketbooksChallengeCommitmentRequest, IssuedTicketbooksChallengeCommitmentResponse,
IssuedTicketbooksDataRequest, IssuedTicketbooksDataResponse, IssuedTicketbooksForCountResponse,
IssuedTicketbooksForResponse, VerifyEcashTicketBody,
};
use nym_api_requests::ecash::VerificationKeyResponse;
use nym_api_requests::models::{
AnnotationResponse, ApiHealthResponse, BinaryBuildInformationOwned, ChainBlocksStatusResponse,
ChainStatusResponse, KeyRotationInfoResponse, LegacyDescribedMixNode, NodePerformanceResponse,
NodeRefreshBody, NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
SignerInformationResponse,
AnnotationResponse, ApiHealthResponse, BinaryBuildInformationOwned, ChainStatusResponse,
KeyRotationInfoResponse, LegacyDescribedMixNode, NodePerformanceResponse, NodeRefreshBody,
NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
};
use nym_api_requests::nym_nodes::{
NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponseV1,
@@ -1103,9 +1101,8 @@ pub trait NymApiClientExt: ApiClient {
async fn partial_expiration_date_signatures(
&self,
expiration_date: Option<Date>,
epoch_id: Option<EpochId>,
) -> Result<PartialExpirationDateSignatureResponse, NymAPIError> {
let mut params = match expiration_date {
let params = match expiration_date {
None => Vec::new(),
Some(exp) => vec![(
ecash::EXPIRATION_DATE_PARAM,
@@ -1113,10 +1110,6 @@ pub trait NymApiClientExt: ApiClient {
)],
};
if let Some(epoch_id) = epoch_id {
params.push((ecash::EPOCH_ID_PARAM, epoch_id.to_string()));
}
self.get_json(
&[
routes::V1_API_VERSION,
@@ -1153,9 +1146,8 @@ pub trait NymApiClientExt: ApiClient {
async fn global_expiration_date_signatures(
&self,
expiration_date: Option<Date>,
epoch_id: Option<EpochId>,
) -> Result<AggregatedExpirationDateSignatureResponse, NymAPIError> {
let mut params = match expiration_date {
let params = match expiration_date {
None => Vec::new(),
Some(exp) => vec![(
ecash::EXPIRATION_DATE_PARAM,
@@ -1163,10 +1155,6 @@ pub trait NymApiClientExt: ApiClient {
)],
};
if let Some(epoch_id) = epoch_id {
params.push((ecash::EPOCH_ID_PARAM, epoch_id.to_string()));
}
self.get_json(
&[
routes::V1_API_VERSION,
@@ -1343,22 +1331,6 @@ pub trait NymApiClientExt: ApiClient {
.await
}
async fn get_chain_blocks_status(&self) -> Result<ChainBlocksStatusResponse, NymAPIError> {
self.get_json("/v1/network/chain-blocks-status", NO_PARAMS)
.await
}
#[instrument(level = "debug", skip(self))]
async fn get_signer_status(&self) -> Result<EcashSignerStatusResponse, NymAPIError> {
self.get_json("/v1/ecash/signer-status", NO_PARAMS).await
}
#[instrument(level = "debug", skip(self))]
async fn get_signer_information(&self) -> Result<SignerInformationResponse, NymAPIError> {
self.get_json("/v1/api-status/signer-information", NO_PARAMS)
.await
}
#[instrument(level = "debug", skip(self))]
async fn get_key_rotation_info(&self) -> Result<KeyRotationInfoResponse, NymAPIError> {
self.get_json(
@@ -8,11 +8,11 @@ use crate::nyxd::CosmWasmClient;
use async_trait::async_trait;
use cosmrs::AccountId;
use cosmwasm_std::Addr;
use nym_coconut_dkg_common::dealer::RegisteredDealerDetails;
use nym_coconut_dkg_common::types::{ChunkIndex, NodeIndex, StateAdvanceResponse};
use serde::Deserialize;
use tracing::trace;
use nym_coconut_dkg_common::dealer::RegisteredDealerDetails;
pub use nym_coconut_dkg_common::{
dealer::{DealerDetailsResponse, PagedDealerIndexResponse, PagedDealerResponse},
dealing::{
@@ -21,9 +21,7 @@ pub use nym_coconut_dkg_common::{
},
msg::QueryMsg as DkgQueryMsg,
types::{DealerDetails, DealingIndex, Epoch, EpochId, EpochState, State},
verification_key::{
ContractVKShare, PagedVKSharesResponse, VerificationKeyShare, VkShareResponse,
},
verification_key::{ContractVKShare, PagedVKSharesResponse, VkShareResponse},
};
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@@ -43,11 +41,6 @@ pub trait DkgQueryClient {
self.query_dkg_contract(request).await
}
async fn get_epoch_at_height(&self, height: u64) -> Result<Option<Epoch>, NyxdError> {
let request = DkgQueryMsg::GetEpochStateAtHeight { height };
self.query_dkg_contract(request).await
}
async fn can_advance_state(&self) -> Result<StateAdvanceResponse, NyxdError> {
let request = DkgQueryMsg::CanAdvanceState {};
self.query_dkg_contract(request).await
@@ -94,34 +87,6 @@ pub trait DkgQueryClient {
self.query_dkg_contract(request).await
}
async fn get_epoch_dealers_paged(
&self,
epoch_id: EpochId,
start_after: Option<String>,
limit: Option<u32>,
) -> Result<PagedDealerResponse, NyxdError> {
let request = DkgQueryMsg::GetEpochDealers {
epoch_id,
start_after,
limit,
};
self.query_dkg_contract(request).await
}
async fn get_epoch_dealers_addresses_paged(
&self,
epoch_id: EpochId,
start_after: Option<String>,
limit: Option<u32>,
) -> Result<PagedDealerResponse, NyxdError> {
let request = DkgQueryMsg::GetEpochDealersAddresses {
epoch_id,
start_after,
limit,
};
self.query_dkg_contract(request).await
}
async fn get_dealer_indices_paged(
&self,
start_after: Option<String>,
@@ -243,20 +208,6 @@ pub trait PagedDkgQueryClient: DkgQueryClient {
collect_paged!(self, get_current_dealers_paged, dealers)
}
async fn get_all_epoch_dealers(
&self,
epoch_id: EpochId,
) -> Result<Vec<DealerDetails>, NyxdError> {
collect_paged!(self, get_epoch_dealers_paged, dealers, epoch_id)
}
async fn get_all_epoch_dealers_addresses(
&self,
epoch_id: EpochId,
) -> Result<Vec<DealerDetails>, NyxdError> {
collect_paged!(self, get_epoch_dealers_addresses_paged, dealers, epoch_id)
}
async fn get_all_dealer_indices(&self) -> Result<Vec<(Addr, NodeIndex)>, NyxdError> {
collect_paged!(self, get_dealer_indices_paged, indices)
}
@@ -306,9 +257,6 @@ mod tests {
match msg {
DkgQueryMsg::GetState {} => client.get_state().ignore(),
DkgQueryMsg::GetCurrentEpochState {} => client.get_current_epoch().ignore(),
DkgQueryMsg::GetEpochStateAtHeight { height } => {
client.get_epoch_at_height(height).ignore()
}
DkgQueryMsg::CanAdvanceState {} => client.can_advance_state().ignore(),
DkgQueryMsg::GetCurrentEpochThreshold {} => {
client.get_current_epoch_threshold().ignore()
@@ -328,20 +276,6 @@ mod tests {
DkgQueryMsg::GetCurrentDealers { limit, start_after } => client
.get_current_dealers_paged(start_after, limit)
.ignore(),
QueryMsg::GetEpochDealers {
epoch_id,
limit,
start_after,
} => client
.get_epoch_dealers_paged(epoch_id, start_after, limit)
.ignore(),
QueryMsg::GetEpochDealersAddresses {
epoch_id,
limit,
start_after,
} => client
.get_epoch_dealers_addresses_paged(epoch_id, start_after, limit)
.ignore(),
DkgQueryMsg::GetDealerIndices { limit, start_after } => {
client.get_dealer_indices_paged(start_after, limit).ignore()
}
@@ -139,22 +139,12 @@ impl NyxdClient<HttpClient> {
})
}
pub fn connect_with_network_details<U>(
endpoint: U,
network_details: NymNetworkDetails,
) -> Result<QueryHttpRpcNyxdClient, NyxdError>
where
U: TryInto<HttpClientUrl, Error = TendermintRpcError>,
{
let config = Config::try_from_nym_network_details(&network_details)?;
Self::connect(config, endpoint)
}
pub fn connect_to_default_env<U>(endpoint: U) -> Result<QueryHttpRpcNyxdClient, NyxdError>
where
U: TryInto<HttpClientUrl, Error = TendermintRpcError>,
{
Self::connect_with_network_details(endpoint, NymNetworkDetails::new_from_env())
let config = Config::try_from_nym_network_details(&NymNetworkDetails::new_from_env())?;
Self::connect(config, endpoint)
}
}
+1 -1
View File
@@ -86,7 +86,7 @@ pub async fn execute(args: Args) -> anyhow::Result<()> {
anyhow!("ticketbook got incorrectly imported - the master verification key is missing")
})?;
let expiration_signatures = persistent_storage
.get_expiration_date_signatures(expiration_date, epoch_id)
.get_expiration_date_signatures(expiration_date)
.await?
.ok_or_else(|| {
anyhow!(
@@ -120,7 +120,7 @@ async fn issue_to_file(args: Args, client: SigningClient) -> anyhow::Result<()>
if args.include_expiration_date_signatures {
let signatures = credentials_store
.get_expiration_date_signatures(expiration_date, epoch_id)
.get_expiration_date_signatures(expiration_date)
.await?
.ok_or(anyhow!("missing expiration date signatures!"))?;
@@ -241,7 +241,7 @@ pub async fn delegate_to_multiple_mixnodes(args: Args, client: SigningClient) {
let node_id = row.node_id.clone().parse::<u32>().unwrap();
let coins: Vec<Coin> = vec![];
undelegation_msgs.push((ExecuteMsg::Undelegate { node_id }, coins));
undelegation_table.add_row(std::slice::from_ref(&row.node_id));
undelegation_table.add_row(&[row.node_id.clone()]);
if row.amount.amount > 0 {
delegation_msgs
@@ -55,14 +55,6 @@ impl DealerDetailsResponse {
}
}
#[cw_serde]
pub struct PagedDealerAddressesResponse {
pub dealers: Vec<Addr>,
/// Field indicating paging information for the following queries if the caller wishes to get further entries.
pub start_next_after: Option<Addr>,
}
#[cw_serde]
pub struct PagedDealerResponse {
pub dealers: Vec<DealerDetails>,
@@ -12,8 +12,8 @@ use cosmwasm_schema::cw_serde;
#[cfg(feature = "schema")]
use crate::{
dealer::{
DealerDetailsResponse, PagedDealerAddressesResponse, PagedDealerIndexResponse,
PagedDealerResponse, RegisteredDealerDetails,
DealerDetailsResponse, PagedDealerIndexResponse, PagedDealerResponse,
RegisteredDealerDetails,
},
dealing::{
DealerDealingsStatusResponse, DealingChunkResponse, DealingChunkStatusResponse,
@@ -84,9 +84,6 @@ pub enum QueryMsg {
#[cfg_attr(feature = "schema", returns(Epoch))]
GetCurrentEpochState {},
#[cfg_attr(feature = "schema", returns(Option<Epoch>))]
GetEpochStateAtHeight { height: u64 },
#[cfg_attr(feature = "schema", returns(u64))]
GetCurrentEpochThreshold {},
@@ -105,20 +102,6 @@ pub enum QueryMsg {
#[cfg_attr(feature = "schema", returns(DealerDetailsResponse))]
GetDealerDetails { dealer_address: String },
#[cfg_attr(feature = "schema", returns(PagedDealerAddressesResponse))]
GetEpochDealersAddresses {
epoch_id: EpochId,
limit: Option<u32>,
start_after: Option<String>,
},
#[cfg_attr(feature = "schema", returns(PagedDealerResponse))]
GetEpochDealers {
epoch_id: EpochId,
limit: Option<u32>,
start_after: Option<String>,
},
#[cfg_attr(feature = "schema", returns(PagedDealerResponse))]
GetCurrentDealers {
limit: Option<u32>,
@@ -188,7 +188,7 @@ impl<C> ContractTesterBuilder<C> {
*self.app.api()
}
pub fn querier(&self) -> QuerierWrapper<'_> {
pub fn querier(&self) -> QuerierWrapper {
self.app.wrap()
}
}
@@ -57,7 +57,7 @@ pub trait NodeBond {
fn is_unbonding(&self) -> bool;
fn identity(&self) -> IdentityKeyRef<'_>;
fn identity(&self) -> IdentityKeyRef;
fn original_pledge(&self) -> &Coin;
@@ -125,7 +125,7 @@ impl NodeBond for MixNodeBond {
self.is_unbonding
}
fn identity(&self) -> IdentityKeyRef<'_> {
fn identity(&self) -> IdentityKeyRef {
self.identity()
}
@@ -178,7 +178,7 @@ impl NodeBond for NymNodeBond {
self.is_unbonding
}
fn identity(&self) -> IdentityKeyRef<'_> {
fn identity(&self) -> IdentityKeyRef {
self.identity()
}
@@ -58,7 +58,7 @@ impl<'a> PrimaryKey<'a> for Role {
type Suffix = <u8 as PrimaryKey<'a>>::Suffix;
type SuperSuffix = <u8 as PrimaryKey<'a>>::SuperSuffix;
fn key(&self) -> Vec<Key<'_>> {
fn key(&self) -> Vec<Key> {
// I'm not sure why it wasn't possible to delegate the call to
// `(*self as u8).key()` directly...
// I guess because of the `Key::Ref(&'a [u8])` variant?
@@ -1,123 +0,0 @@
/*
* Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: Apache-2.0
*/
-- 1. add temporary `epoch_id` column
ALTER TABLE pending_issuance
ADD COLUMN epoch_id INTEGER;
-- 2. populate the value
UPDATE pending_issuance
SET epoch_id = (SELECT epoch_id
FROM expiration_date_signatures
WHERE expiration_date_signatures.expiration_date = pending_issuance.expiration_date);
-- 3. create new expiration_date_signatures table (with changed constraints)
CREATE TABLE expiration_date_signatures_new
(
expiration_date DATE NOT NULL,
epoch_id INTEGER NOT NULL,
serialization_revision INTEGER NOT NULL,
-- combined signatures for all tuples issued for given day
serialised_signatures BLOB NOT NULL,
PRIMARY KEY (epoch_id, expiration_date)
);
-- 4. migrate the data
INSERT INTO expiration_date_signatures_new (expiration_date, epoch_id, serialization_revision, serialised_signatures)
SELECT expiration_date, epoch_id, serialization_revision, serialised_signatures
FROM expiration_date_signatures;
-- 5. drop and recreate the table references (due to new FK)
-- 5.1.
-- (data for ticketbooks that have an associated deposit, but failed to get issued)
CREATE TABLE pending_issuance_new
(
deposit_id INTEGER NOT NULL PRIMARY KEY,
-- introduce a way for us to introduce breaking changes in serialization of data
serialization_revision INTEGER NOT NULL,
pending_ticketbook_data BLOB NOT NULL UNIQUE,
-- for each ticketbook we MUST have corresponding expiration date signatures
expiration_date DATE NOT NULL,
epoch_id INTEGER NOT NULL,
-- for each ticketbook we MUST have corresponding expiration date signatures
FOREIGN KEY (epoch_id, expiration_date) REFERENCES expiration_date_signatures_new (epoch_id, expiration_date)
);
INSERT INTO pending_issuance_new (deposit_id, serialization_revision, pending_ticketbook_data, expiration_date,
epoch_id)
SELECT deposit_id, serialization_revision, pending_ticketbook_data, expiration_date, epoch_id
FROM pending_issuance;
-- 5.2.
CREATE TABLE ecash_ticketbook_new
(
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-- introduce a way for us to introduce breaking changes in serialization of data
serialization_revision INTEGER NOT NULL,
-- the type of the associated ticketbook
ticketbook_type TEXT NOT NULL,
-- the actual crypto data of the ticketbook (wallet, keys, etc.)
ticketbook_data BLOB NOT NULL UNIQUE,
-- for each ticketbook we MUST have corresponding expiration date signatures
expiration_date DATE NOT NULL,
-- for each ticketbook we MUST have corresponding coin index signatures
epoch_id INTEGER NOT NULL,
-- the initial number of tickets the wallet has been created for
total_tickets INTEGER NOT NULL,
-- how many tickets have been used so far (the `l` value of the wallet)
used_tickets INTEGER NOT NULL,
-- FOREIGN KEYS:
-- for each ticketbook we MUST have corresponding coin index signatures
FOREIGN KEY (epoch_id) REFERENCES coin_indices_signatures (epoch_id),
-- for each ticketbook we MUST have corresponding expiration date signatures
FOREIGN KEY (expiration_date, epoch_id) REFERENCES expiration_date_signatures_new (expiration_date, epoch_id)
);
INSERT INTO ecash_ticketbook_new (id, serialization_revision, ticketbook_type, ticketbook_data, expiration_date,
epoch_id, total_tickets, used_tickets)
SELECT id,
serialization_revision,
ticketbook_type,
ticketbook_data,
expiration_date,
epoch_id,
total_tickets,
used_tickets
FROM ecash_ticketbook;
-- 6. finally swap out the old tables
-- drop old tables
DROP TABLE expiration_date_signatures;
DROP TABLE pending_issuance;
DROP TABLE ecash_ticketbook;
-- rename new tables
ALTER TABLE expiration_date_signatures_new
RENAME TO expiration_date_signatures;
ALTER TABLE pending_issuance_new
RENAME TO pending_issuance;
ALTER TABLE ecash_ticketbook_new
RENAME TO ecash_ticketbook;
@@ -28,7 +28,7 @@ struct EcashCredentialManagerInner {
pending: HashMap<i64, RetrievedPendingTicketbook>,
master_vk: HashMap<u64, VerificationKeyAuth>,
coin_indices_sigs: HashMap<u64, Vec<AnnotatedCoinIndexSignature>>,
expiration_date_sigs: HashMap<(u64, Date), Vec<AnnotatedExpirationDateSignature>>,
expiration_date_sigs: HashMap<Date, Vec<AnnotatedExpirationDateSignature>>,
_next_id: i64,
}
@@ -242,14 +242,10 @@ impl MemoryEcachTicketbookManager {
pub(crate) async fn get_expiration_date_signatures(
&self,
expiration_date: Date,
epoch_id: u64,
) -> Option<Vec<AnnotatedExpirationDateSignature>> {
let guard = self.inner.read().await;
guard
.expiration_date_sigs
.get(&(epoch_id, expiration_date))
.cloned()
guard.expiration_date_sigs.get(&expiration_date).cloned()
}
pub(crate) async fn insert_expiration_date_signatures(
@@ -258,9 +254,8 @@ impl MemoryEcachTicketbookManager {
) {
let mut guard = self.inner.write().await;
guard.expiration_date_sigs.insert(
(sigs.epoch_id, sigs.expiration_date),
sigs.signatures.clone(),
);
guard
.expiration_date_sigs
.insert(sigs.expiration_date, sigs.signatures.clone());
}
}
@@ -39,7 +39,7 @@ impl SqliteEcashTicketbookManager {
Ok(())
}
pub(crate) async fn begin_storage_tx(&self) -> Result<Transaction<'_, Sqlite>, sqlx::Error> {
pub(crate) async fn begin_storage_tx(&self) -> Result<Transaction<Sqlite>, sqlx::Error> {
self.connection_pool.begin().await
}
@@ -260,17 +260,15 @@ impl SqliteEcashTicketbookManager {
pub(crate) async fn get_expiration_date_signatures(
&self,
expiration_date: Date,
epoch_id: i64,
) -> Result<Option<RawExpirationDateSignatures>, sqlx::Error> {
sqlx::query_as!(
RawExpirationDateSignatures,
r#"
SELECT serialised_signatures, serialization_revision as "serialization_revision: u8"
SELECT epoch_id as "epoch_id: u32", serialised_signatures, serialization_revision as "serialization_revision: u8"
FROM expiration_date_signatures
WHERE expiration_date = ? AND epoch_id = ?
WHERE expiration_date = ?
"#,
expiration_date,
epoch_id
expiration_date
)
.fetch_optional(&*self.connection_pool)
.await
@@ -166,11 +166,10 @@ impl Storage for EphemeralStorage {
async fn get_expiration_date_signatures(
&self,
expiration_date: Date,
epoch_id: u64,
) -> Result<Option<Vec<AnnotatedExpirationDateSignature>>, Self::StorageError> {
Ok(self
.storage_manager
.get_expiration_date_signatures(expiration_date, epoch_id)
.get_expiration_date_signatures(expiration_date)
.await)
}
+1
View File
@@ -60,6 +60,7 @@ pub struct StoredPendingTicketbook {
#[cfg_attr(not(target_arch = "wasm32"), derive(sqlx::FromRow))]
pub struct RawExpirationDateSignatures {
pub epoch_id: u32,
pub serialised_signatures: Vec<u8>,
pub serialization_revision: u8,
}
@@ -63,7 +63,7 @@ impl PersistentStorage {
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.synchronous(SqliteSynchronous::Normal)
.auto_vacuum(SqliteAutoVacuum::Incremental)
.filename(database_path)
.filename(&database_path)
.create_if_missing(true)
.disable_statement_logging();
@@ -75,7 +75,8 @@ impl PersistentStorage {
}
};
let connection_pool = SqlitePoolGuard::new(connection_pool);
let connection_pool =
SqlitePoolGuard::new(database_path.as_ref().to_path_buf(), connection_pool);
if let Err(err) = sqlx::migrate!("./migrations").run(&*connection_pool).await {
error!("Failed to perform migration on the SQLx database: {err}");
@@ -325,11 +326,10 @@ impl Storage for PersistentStorage {
async fn get_expiration_date_signatures(
&self,
expiration_date: Date,
epoch_id: u64,
) -> Result<Option<Vec<AnnotatedExpirationDateSignature>>, Self::StorageError> {
let Some(raw) = self
.storage_manager
.get_expiration_date_signatures(expiration_date, epoch_id as i64)
.get_expiration_date_signatures(expiration_date)
.await?
else {
return Ok(None);
-1
View File
@@ -92,7 +92,6 @@ pub trait Storage: Clone + Send + Sync {
async fn get_expiration_date_signatures(
&self,
expiration_date: Date,
epoch_id: u64,
) -> Result<Option<Vec<AnnotatedExpirationDateSignature>>, Self::StorageError>;
async fn insert_expiration_date_signatures(
@@ -11,11 +11,9 @@ rust-version.workspace = true
readme.workspace = true
[dependencies]
async-trait = { workspace = true }
bs58 = { workspace = true }
cosmwasm-std = { workspace = true }
cw-utils = { workspace = true }
dyn-clone = { workspace = true }
futures = { workspace = true }
rand = { workspace = true }
si-scale = { workspace = true }
@@ -7,36 +7,25 @@ use crate::ClientBandwidth;
use nym_credentials::ecash::utils::ecash_today;
use nym_credentials_interface::Bandwidth;
use nym_gateway_requests::ServerResponse;
use nym_gateway_storage::traits::BandwidthGatewayStorage;
use nym_gateway_storage::GatewayStorage;
use si_scale::helpers::bibytes2;
use time::OffsetDateTime;
use tracing::*;
const FREE_TESTNET_BANDWIDTH_VALUE: Bandwidth = Bandwidth::new_unchecked(64 * 1024 * 1024 * 1024); // 64GB
#[derive(Clone)]
pub struct BandwidthStorageManager {
pub(crate) storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
pub(crate) storage: GatewayStorage,
pub(crate) client_bandwidth: ClientBandwidth,
pub(crate) client_id: i64,
pub(crate) bandwidth_cfg: BandwidthFlushingBehaviourConfig,
pub(crate) only_coconut_credentials: bool,
}
impl Clone for BandwidthStorageManager {
fn clone(&self) -> Self {
Self {
storage: dyn_clone::clone_box(&*self.storage),
client_bandwidth: self.client_bandwidth.clone(),
client_id: self.client_id,
bandwidth_cfg: self.bandwidth_cfg,
only_coconut_credentials: self.only_coconut_credentials,
}
}
}
impl BandwidthStorageManager {
pub fn new(
storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
storage: GatewayStorage,
client_bandwidth: ClientBandwidth,
client_id: i64,
bandwidth_cfg: BandwidthFlushingBehaviourConfig,
+36 -124
View File
@@ -2,14 +2,12 @@
// SPDX-License-Identifier: GPL-3.0-only
use crate::Error;
use async_trait::async_trait;
use credential_sender::CredentialHandler;
use credential_sender::CredentialHandlerConfig;
use error::EcashTicketError;
use futures::channel::mpsc::{self, UnboundedSender};
use nym_credentials::CredentialSpendingData;
use nym_credentials_interface::{ClientTicket, CompactEcashError, NymPayInfo, VerificationKeyAuth};
use nym_gateway_storage::traits::BandwidthGatewayStorage;
use nym_gateway_storage::GatewayStorage;
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::DirectSigningHttpRpcNyxdClient;
@@ -22,7 +20,6 @@ pub mod credential_sender;
pub mod error;
mod helpers;
mod state;
pub mod traits;
pub const TIME_RANGE_SEC: i64 = 30;
@@ -34,21 +31,44 @@ pub struct EcashManager {
cred_sender: UnboundedSender<ClientTicket>,
}
#[async_trait]
impl traits::EcashManager for EcashManager {
async fn verification_key(
impl EcashManager {
pub async fn new(
credential_handler_cfg: CredentialHandlerConfig,
nyxd_client: DirectSigningHttpRpcNyxdClient,
pk_bytes: [u8; 32],
shutdown: nym_task::TaskClient,
storage: GatewayStorage,
) -> Result<Self, Error> {
let shared_state = SharedState::new(nyxd_client, storage).await?;
let (cred_sender, cred_receiver) = mpsc::unbounded();
let cs =
CredentialHandler::new(credential_handler_cfg, cred_receiver, shared_state.clone())
.await?;
cs.start(shutdown);
Ok(EcashManager {
shared_state,
pk_bytes,
pay_infos: Default::default(),
cred_sender,
})
}
pub async fn verification_key(
&self,
epoch_id: EpochId,
) -> Result<RwLockReadGuard<'_, VerificationKeyAuth>, EcashTicketError> {
) -> Result<RwLockReadGuard<VerificationKeyAuth>, EcashTicketError> {
self.shared_state.verification_key(epoch_id).await
}
fn storage(&self) -> Box<dyn BandwidthGatewayStorage + Send + Sync> {
dyn_clone::clone_box(&*self.shared_state.storage)
pub fn storage(&self) -> &GatewayStorage {
&self.shared_state.storage
}
//Check for duplicate pay_info, then check the payment, then insert pay_info if everything succeeded
async fn check_payment(
pub async fn check_payment(
&self,
credential: &CredentialSpendingData,
aggregated_verification_key: &VerificationKeyAuth,
@@ -68,40 +88,6 @@ impl traits::EcashManager for EcashManager {
.await
}
fn async_verify(&self, ticket: ClientTicket) {
// TODO: I guess do something for shutdowns
let _ = self
.cred_sender
.unbounded_send(ticket)
.inspect_err(|_| error!("failed to send the client ticket for verification task"));
}
}
impl EcashManager {
pub async fn new(
credential_handler_cfg: CredentialHandlerConfig,
nyxd_client: DirectSigningHttpRpcNyxdClient,
pk_bytes: [u8; 32],
shutdown: nym_task::TaskClient,
storage: GatewayStorage,
) -> Result<Self, Error> {
let shared_state = SharedState::new(nyxd_client, Box::new(storage)).await?;
let (cred_sender, cred_receiver) = mpsc::unbounded();
let cs =
CredentialHandler::new(credential_handler_cfg, cred_receiver, shared_state.clone())
.await?;
cs.start(shutdown);
Ok(EcashManager {
shared_state,
pk_bytes,
pay_infos: Default::default(),
cred_sender,
})
}
pub async fn verify_pay_info(&self, pay_info: NymPayInfo) -> Result<usize, EcashTicketError> {
//Public key check
if pay_info.pk() != self.pk_bytes {
@@ -166,86 +152,12 @@ impl EcashManager {
inner.insert(index, pay_info);
Ok(())
}
}
pub struct MockEcashManager {
verfication_key: tokio::sync::RwLock<VerificationKeyAuth>,
storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
}
impl MockEcashManager {
pub fn new(storage: Box<dyn BandwidthGatewayStorage + Send + Sync>) -> Self {
Self {
verfication_key: tokio::sync::RwLock::new(
VerificationKeyAuth::from_bytes(&[
129, 187, 76, 12, 1, 51, 46, 26, 132, 205, 148, 109, 140, 131, 50, 119, 45,
128, 51, 218, 106, 70, 181, 74, 244, 38, 162, 62, 42, 12, 5, 100, 7, 136, 32,
155, 18, 219, 195, 182, 3, 56, 168, 16, 93, 154, 249, 230, 16, 202, 90, 134,
246, 25, 98, 6, 175, 215, 188, 239, 71, 84, 66, 1, 43, 66, 197, 180, 216, 80,
55, 185, 140, 216, 14, 48, 244, 214, 20, 68, 106, 41, 48, 252, 188, 181, 231,
170, 23, 211, 215, 12, 91, 147, 47, 7, 4, 0, 0, 0, 0, 0, 0, 0, 174, 31, 237,
215, 159, 183, 71, 125, 90, 147, 84, 78, 49, 216, 66, 232, 92, 206, 41, 230,
239, 209, 211, 166, 131, 190, 148, 36, 225, 194, 146, 6, 120, 34, 194, 5, 154,
155, 234, 41, 191, 119, 227, 51, 91, 128, 151, 240, 129, 208, 253, 171, 234,
170, 71, 139, 251, 78, 49, 35, 218, 16, 77, 150, 177, 204, 83, 210, 67, 147,
66, 162, 58, 25, 96, 168, 61, 180, 92, 21, 18, 78, 194, 98, 176, 123, 122, 176,
81, 150, 187, 20, 64, 69, 0, 134, 142, 3, 84, 108, 3, 55, 107, 111, 73, 31, 46,
51, 225, 248, 202, 173, 194, 24, 104, 96, 31, 61, 24, 140, 220, 31, 176, 200,
30, 217, 66, 58, 11, 181, 158, 196, 179, 199, 177, 7, 210, 4, 119, 142, 149,
59, 3, 186, 145, 27, 230, 125, 230, 246, 197, 196, 119, 70, 239, 115, 99, 215,
63, 205, 63, 74, 108, 201, 42, 226, 150, 137, 3, 157, 45, 25, 163, 54, 107,
153, 61, 141, 64, 207, 139, 41, 203, 39, 36, 97, 181, 72, 206, 235, 221, 178,
171, 60, 4, 6, 170, 181, 213, 10, 216, 53, 28, 32, 33, 41, 224, 60, 247, 206,
137, 108, 251, 229, 234, 112, 65, 145, 124, 212, 125, 116, 154, 114, 2, 125,
202, 24, 25, 196, 219, 104, 200, 131, 133, 180, 39, 21, 144, 204, 8, 151, 218,
99, 64, 209, 47, 5, 42, 13, 214, 139, 54, 112, 224, 53, 238, 250, 56, 42, 105,
15, 21, 238, 99, 225, 79, 121, 104, 155, 230, 243, 133, 47, 39, 147, 98, 45,
113, 137, 200, 102, 151, 122, 174, 9, 250, 17, 138, 191, 129, 202, 244, 107,
75, 48, 141, 136, 89, 168, 124, 88, 174, 251, 17, 35, 146, 88, 76, 134, 102,
105, 204, 16, 176, 214, 63, 13, 170, 225, 250, 112, 7, 237, 161, 160, 15, 71,
10, 130, 137, 69, 186, 64, 223, 188, 5, 5, 228, 57, 214, 134, 247, 20, 171,
140, 43, 230, 57, 29, 127, 136, 169, 80, 14, 137, 130, 200, 205, 222, 81, 143,
40, 77, 68, 197, 91, 142, 91, 84, 164, 15, 133, 242, 149, 255, 173, 201, 108,
208, 23, 188, 230, 158, 146, 54, 198, 52, 148, 123, 202, 52, 222, 50, 4, 62,
211, 208, 176, 61, 104, 151, 227, 192, 224, 200, 132, 53, 187, 240, 254, 150,
60, 30, 140, 11, 63, 71, 12, 30, 233, 255, 144, 250, 16, 81, 38, 33, 9, 185,
195, 214, 0, 119, 117, 94, 100, 103, 144, 10, 189, 65, 113, 114, 192, 11, 177,
214, 223, 218, 36, 139, 183, 2, 206, 247, 245, 88, 62, 231, 183, 50, 46, 95,
202, 152, 82, 244, 80, 173, 192, 147, 51, 248, 46, 181, 194, 205, 233, 67, 144,
155, 250, 142, 124, 71, 9, 136, 142, 88, 29, 99, 222, 43, 181, 172, 120, 187,
179, 172, 240, 231, 57, 236, 195, 158, 182, 203, 19, 49, 220, 180, 212, 101,
105, 239, 58, 215, 0, 50, 100, 172, 29, 236, 170, 108, 129, 150, 5, 64, 238,
59, 50, 4, 21, 131, 197, 142, 191, 76, 101, 140, 133, 112, 38, 235, 113, 203,
22, 161, 204, 84, 73, 125, 219, 70, 62, 67, 119, 52, 130, 208, 180, 231, 78,
141, 181, 13, 207, 196, 126, 159, 70, 34, 195, 70,
])
.unwrap(),
),
storage: dyn_clone::clone_box(&*storage),
}
pub fn async_verify(&self, ticket: ClientTicket) {
// TODO: I guess do something for shutdowns
let _ = self
.cred_sender
.unbounded_send(ticket)
.inspect_err(|_| error!("failed to send the client ticket for verification task"));
}
}
#[async_trait]
impl traits::EcashManager for MockEcashManager {
async fn verification_key(
&self,
_epoch_id: EpochId,
) -> Result<RwLockReadGuard<'_, VerificationKeyAuth>, EcashTicketError> {
Ok(self.verfication_key.read().await)
}
fn storage(&self) -> Box<dyn BandwidthGatewayStorage + Send + Sync> {
dyn_clone::clone_box(&*self.storage)
}
async fn check_payment(
&self,
_credential: &CredentialSpendingData,
_aggregated_verification_key: &VerificationKeyAuth,
) -> Result<(), EcashTicketError> {
Ok(())
}
fn async_verify(&self, _ticket: ClientTicket) {}
}
@@ -6,7 +6,7 @@ use crate::Error;
use cosmwasm_std::{from_json, CosmosMsg, WasmMsg};
use nym_credentials_interface::VerificationKeyAuth;
use nym_ecash_contract_common::msg::ExecuteMsg;
use nym_gateway_storage::traits::BandwidthGatewayStorage;
use nym_gateway_storage::GatewayStorage;
use nym_validator_client::coconut::all_ecash_api_clients;
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::nyxd::contract_traits::{
@@ -22,28 +22,18 @@ use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use tracing::{error, trace, warn};
// state shared by different subtasks dealing with credentials
#[derive(Clone)]
pub(crate) struct SharedState {
pub(crate) nyxd_client: Arc<RwLock<DirectSigningHttpRpcNyxdClient>>,
pub(crate) address: AccountId,
pub(crate) epoch_data: Arc<RwLock<BTreeMap<EpochId, EpochState>>>,
pub(crate) storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
}
impl Clone for SharedState {
fn clone(&self) -> Self {
Self {
nyxd_client: self.nyxd_client.clone(),
address: self.address.clone(),
epoch_data: self.epoch_data.clone(),
storage: dyn_clone::clone_box(&*self.storage),
}
}
pub(crate) storage: GatewayStorage,
}
impl SharedState {
pub(crate) async fn new(
nyxd_client: DirectSigningHttpRpcNyxdClient,
storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
storage: GatewayStorage,
) -> Result<Self, Error> {
let address = nyxd_client.address();
@@ -125,7 +115,7 @@ impl SharedState {
async fn set_epoch_data(
&self,
epoch_id: EpochId,
) -> Result<RwLockWriteGuard<'_, BTreeMap<EpochId, EpochState>>, EcashTicketError> {
) -> Result<RwLockWriteGuard<BTreeMap<EpochId, EpochState>>, EcashTicketError> {
let Some(threshold) = self.threshold(epoch_id).await? else {
return Err(EcashTicketError::DKGThresholdUnavailable { epoch_id });
};
@@ -186,7 +176,7 @@ impl SharedState {
pub(crate) async fn api_clients(
&self,
epoch_id: EpochId,
) -> Result<RwLockReadGuard<'_, Vec<EcashApiClient>>, EcashTicketError> {
) -> Result<RwLockReadGuard<Vec<EcashApiClient>>, EcashTicketError> {
let guard = self.epoch_data.read().await;
// the key was already in the map
@@ -212,7 +202,7 @@ impl SharedState {
pub(crate) async fn verification_key(
&self,
epoch_id: EpochId,
) -> Result<RwLockReadGuard<'_, VerificationKeyAuth>, EcashTicketError> {
) -> Result<RwLockReadGuard<VerificationKeyAuth>, EcashTicketError> {
let guard = self.epoch_data.read().await;
// the key was already in the map
@@ -235,11 +225,11 @@ impl SharedState {
}))
}
pub(crate) async fn start_tx(&self) -> RwLockWriteGuard<'_, DirectSigningHttpRpcNyxdClient> {
pub(crate) async fn start_tx(&self) -> RwLockWriteGuard<DirectSigningHttpRpcNyxdClient> {
self.nyxd_client.write().await
}
pub(crate) async fn start_query(&self) -> RwLockReadGuard<'_, DirectSigningHttpRpcNyxdClient> {
pub(crate) async fn start_query(&self) -> RwLockReadGuard<DirectSigningHttpRpcNyxdClient> {
self.nyxd_client.read().await
}
@@ -1,23 +0,0 @@
use async_trait::async_trait;
use nym_credentials::CredentialSpendingData;
use nym_credentials_interface::{ClientTicket, VerificationKeyAuth};
use nym_gateway_storage::traits::BandwidthGatewayStorage;
use nym_validator_client::nym_api::EpochId;
use tokio::sync::RwLockReadGuard;
use crate::ecash::error::EcashTicketError;
#[async_trait]
pub trait EcashManager {
async fn verification_key(
&self,
epoch_id: EpochId,
) -> Result<RwLockReadGuard<'_, VerificationKeyAuth>, EcashTicketError>;
fn storage(&self) -> Box<dyn BandwidthGatewayStorage + Send + Sync>;
async fn check_payment(
&self,
credential: &CredentialSpendingData,
aggregated_verification_key: &VerificationKeyAuth,
) -> Result<(), EcashTicketError>;
fn async_verify(&self, ticket: ClientTicket);
}
+3 -3
View File
@@ -1,8 +1,8 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::ecash::traits::EcashManager;
use bandwidth_storage_manager::BandwidthStorageManager;
use ecash::EcashManager;
use nym_credentials::ecash::utils::{cred_exp_date, ecash_today, EcashTime};
use nym_credentials_interface::{Bandwidth, ClientTicket, TicketType};
use nym_gateway_requests::models::CredentialSpendingRequest;
@@ -20,14 +20,14 @@ pub mod error;
pub struct CredentialVerifier {
credential: CredentialSpendingRequest,
ecash_verifier: Arc<dyn EcashManager + Send + Sync>,
ecash_verifier: Arc<EcashManager>,
bandwidth_storage_manager: BandwidthStorageManager,
}
impl CredentialVerifier {
pub fn new(
credential: CredentialSpendingRequest,
ecash_verifier: Arc<dyn EcashManager + Send + Sync>,
ecash_verifier: Arc<EcashManager>,
bandwidth_storage_manager: BandwidthStorageManager,
) -> Self {
CredentialVerifier {
-1
View File
@@ -15,7 +15,6 @@ bls12_381 = { workspace = true, default-features = false }
serde = { workspace = true, features = ["derive"] }
thiserror = { workspace = true }
strum = { workspace = true, features = ["derive"] }
strum_macros = { workspace = true }
time = { workspace = true, features = ["serde"] }
utoipa = { workspace = true }
rand = { workspace = true }
+3 -3
View File
@@ -229,9 +229,9 @@ impl From<PayInfo> for NymPayInfo {
Serialize,
Deserialize,
Hash,
strum_macros::Display,
strum_macros::EnumString,
strum_macros::EnumIter,
strum::Display,
strum::EnumString,
strum::EnumIter,
)]
#[serde(rename_all = "kebab-case")]
#[strum(serialize_all = "kebab-case")]
+1 -1
View File
@@ -51,7 +51,7 @@ pub async fn obtain_expiration_date_signatures(
for ecash_api_client in ecash_api_clients.iter() {
match ecash_api_client
.api_client
.partial_expiration_date_signatures(None, None)
.partial_expiration_date_signatures(None)
.await
{
Ok(signature) => {
@@ -1,27 +0,0 @@
[package]
name = "nym-ecash-signer-check-types"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
readme.workspace = true
[dependencies]
semver = { workspace = true }
serde = { workspace = true, features = ["derive"] }
url = { workspace = true }
thiserror = { workspace = true }
time = { workspace = true }
tracing = { workspace = true }
utoipa = { workspace = true }
nym-coconut-dkg-common = { path = "../cosmwasm-smart-contracts/coconut-dkg" }
nym-crypto = { path = "../crypto", features = ["asymmetric"] }
[lints]
workspace = true
@@ -1,97 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_coconut_dkg_common::dealer::DealerDetails;
use nym_coconut_dkg_common::verification_key::{ContractVKShare, VerificationKeyShare};
use nym_crypto::asymmetric::ed25519;
use nym_crypto::asymmetric::ed25519::Ed25519RecoveryError;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use url::Url;
use utoipa::ToSchema;
#[derive(Debug, Error)]
pub enum MalformedDealer {
#[error("dealer at {dealer_url} has provided invalid ed25519 pubkey: {source}")]
InvalidDealerPubkey {
dealer_url: String,
source: Ed25519RecoveryError,
},
#[error("dealer at {dealer_url} has provided invalid announce url: {source}")]
InvalidDealerAddress {
dealer_url: String,
source: url::ParseError,
},
}
#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
pub struct RawDealerInformation {
pub announce_address: String,
pub owner_address: String,
pub node_index: u64,
pub public_key: String,
pub verification_key_share: Option<VerificationKeyShare>,
pub share_verified: bool,
}
impl RawDealerInformation {
pub fn new(
dealer_details: &DealerDetails,
contract_share: Option<&ContractVKShare>,
) -> RawDealerInformation {
RawDealerInformation {
announce_address: dealer_details.announce_address.clone(),
owner_address: dealer_details.address.to_string(),
node_index: dealer_details.assigned_index,
public_key: dealer_details.ed25519_identity.clone(),
verification_key_share: contract_share.map(|s| s.share.clone()),
share_verified: contract_share.map(|s| s.verified).unwrap_or(false),
}
}
pub fn parse(&self) -> Result<DealerInformation, MalformedDealer> {
Ok(DealerInformation {
announce_address: self.announce_address.parse().map_err(|source| {
MalformedDealer::InvalidDealerAddress {
dealer_url: self.announce_address.clone(),
source,
}
})?,
owner_address: self.owner_address.clone(),
node_index: self.node_index,
public_key: self.public_key.parse().map_err(|source| {
MalformedDealer::InvalidDealerPubkey {
dealer_url: self.announce_address.clone(),
source,
}
})?,
verification_key_share: self.verification_key_share.clone(),
share_verified: self.share_verified,
})
}
}
#[derive(Debug)]
pub struct DealerInformation {
pub announce_address: Url,
pub owner_address: String,
pub node_index: u64,
pub public_key: ed25519::PublicKey,
// no need to parse it into the full type as it doesn't get us anything
pub verification_key_share: Option<VerificationKeyShare>,
pub share_verified: bool,
}
impl From<DealerInformation> for RawDealerInformation {
fn from(d: DealerInformation) -> Self {
RawDealerInformation {
announce_address: d.announce_address.to_string(),
owner_address: d.owner_address,
node_index: d.node_index,
public_key: d.public_key.to_base58_string(),
verification_key_share: d.verification_key_share,
share_verified: d.share_verified,
}
}
}
@@ -1,127 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_coconut_dkg_common::types::EpochId;
use nym_coconut_dkg_common::verification_key::VerificationKeyShare;
use nym_crypto::asymmetric::ed25519;
use std::time::Duration;
use time::OffsetDateTime;
use tracing::{debug, warn};
pub trait Verifiable {
fn verify_signature(&self, pub_key: &ed25519::PublicKey) -> bool;
}
pub trait TimestampedResponse {
fn timestamp(&self) -> OffsetDateTime;
}
pub trait LegacyChainResponse {
fn chain_synced(&self, now: OffsetDateTime, stall_threshold: Duration) -> bool;
}
pub trait ChainResponse: Verifiable + TimestampedResponse {
fn chain_synced(&self) -> bool;
fn chain_available(
&self,
pub_key: &ed25519::PublicKey,
now: OffsetDateTime,
stale_response_threshold: Duration,
) -> bool {
if !self.verify_signature(pub_key) {
warn!("failed signature verification on chain status response");
return false;
}
// we rely on information provided from the api itself AS LONG AS it's not too outdated
if self.timestamp() + stale_response_threshold < now {
return false;
}
self.chain_synced()
}
}
pub trait LegacySignerResponse {
fn signer_identity(&self) -> &str;
fn signer_verification_key(&self) -> &Option<String>;
fn unprovable_signing_available(
&self,
pub_key: &ed25519::PublicKey,
expected_verification_key: Option<VerificationKeyShare>,
share_verified: bool,
) -> bool {
if self.signer_identity() != pub_key.to_base58_string() {
warn!("mismatched identity key on the legacy response");
return false;
}
// the contract share hasn't been verified yet, so we're probably in the middle of DKG
// thus if there's a bit of desync in the state, it's fine
if !share_verified {
return true;
}
if self.signer_verification_key() != &expected_verification_key {
warn!("mismatched [ecash] verification key on the legacy response");
return false;
}
true
}
}
pub trait SignerResponse: Verifiable + TimestampedResponse {
fn has_signing_keys(&self) -> bool;
fn signer_disabled(&self) -> bool;
fn is_ecash_signer(&self) -> bool;
fn dkg_ecash_epoch_id(&self) -> EpochId;
fn provable_signing_available(
&self,
pub_key: &ed25519::PublicKey,
dkg_epoch_id: EpochId,
now: OffsetDateTime,
stale_response_threshold: Duration,
) -> bool {
if !self.verify_signature(pub_key) {
warn!("failed signature verification on chain status response");
return false;
}
// we rely on information provided from the api itself AS LONG AS it's not too outdated
if self.timestamp() + stale_response_threshold < now {
return false;
}
if !self.has_signing_keys() {
debug!("missing signing keys");
return false;
}
if self.signer_disabled() {
debug!("signer functionalities explicitly disabled");
return false;
}
if !self.is_ecash_signer() {
debug!("signer doesn't recognise it's a signer for this epoch");
return false;
}
if dkg_epoch_id != self.dkg_ecash_epoch_id() {
debug!(
"mismatched dkg epoch id. current: {dkg_epoch_id}, signer's: {}",
self.dkg_ecash_epoch_id()
);
return false;
}
true
}
}
@@ -1,6 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod dealer_information;
pub mod helper_traits;
pub mod status;
@@ -1,303 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::dealer_information::RawDealerInformation;
use crate::helper_traits::{
ChainResponse, LegacyChainResponse, LegacySignerResponse, SignerResponse,
};
use nym_coconut_dkg_common::types::EpochId;
use nym_coconut_dkg_common::verification_key::VerificationKeyShare;
use nym_crypto::asymmetric::ed25519;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use time::OffsetDateTime;
use utoipa::ToSchema;
pub(crate) const CHAIN_STALL_THRESHOLD: Duration = Duration::from_secs(5 * 60);
pub(crate) const STALE_RESPONSE_THRESHOLD: Duration = Duration::from_secs(5 * 60);
// the reason for generics is not to remove duplication of code,
// but because without them, we'd be having problems with circular dependencies,
// i.e. nym-api-requests depending on ecash-signer-check-types and
// ecash-signer-check-types needing nym-api-requests
#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
pub enum Status<L, T> {
/// The API, even though it reports correct version, did not response to the status query
Unreachable,
/// The API is running an outdated version that does not expose the required endpoint
Outdated,
/// Response to the legacy (unsigned) status query
ReachableLegacy { response: Box<L> },
/// Response to the current (signed) status query
Reachable { response: Box<T> },
}
impl<L, T> Status<L, T>
where
L: LegacyChainResponse,
T: ChainResponse,
{
pub fn chain_available(&self, pub_key: ed25519::PublicKey) -> bool {
let now = OffsetDateTime::now_utc();
match self {
Status::Unreachable | Status::Outdated => false,
Status::ReachableLegacy { response } => {
response.chain_synced(now, CHAIN_STALL_THRESHOLD)
}
Status::Reachable { response } => {
response.chain_available(&pub_key, now, STALE_RESPONSE_THRESHOLD)
}
}
}
pub fn chain_provably_stalled(&self, pub_key: ed25519::PublicKey) -> bool {
let now = OffsetDateTime::now_utc();
match self {
Status::Unreachable | Status::Outdated | Status::ReachableLegacy { .. } => false,
Status::Reachable { response } => {
!response.chain_available(&pub_key, now, STALE_RESPONSE_THRESHOLD)
}
}
}
pub fn chain_unprovably_stalled(&self) -> bool {
let now = OffsetDateTime::now_utc();
match self {
Status::Unreachable | Status::Outdated | Status::Reachable { .. } => false,
Status::ReachableLegacy { response } => {
!response.chain_synced(now, CHAIN_STALL_THRESHOLD)
}
}
}
}
impl<L, T> Status<L, T>
where
L: LegacySignerResponse,
T: SignerResponse,
{
pub fn signing_available(
&self,
pub_key: ed25519::PublicKey,
dkg_epoch_id: u64,
expected_verification_key: Option<VerificationKeyShare>,
share_verified: bool,
) -> bool {
let now = OffsetDateTime::now_utc();
match self {
Status::Unreachable | Status::Outdated => false,
Status::ReachableLegacy { response } => response.unprovable_signing_available(
&pub_key,
expected_verification_key,
share_verified,
),
Status::Reachable { response } => response.provable_signing_available(
&pub_key,
dkg_epoch_id,
now,
STALE_RESPONSE_THRESHOLD,
),
}
}
pub fn signing_provably_unavailable(
&self,
pub_key: ed25519::PublicKey,
dkg_epoch_id: EpochId,
) -> bool {
let now = OffsetDateTime::now_utc();
match self {
Status::Unreachable | Status::Outdated | Status::ReachableLegacy { .. } => false,
Status::Reachable { response } => !response.provable_signing_available(
&pub_key,
dkg_epoch_id,
now,
STALE_RESPONSE_THRESHOLD,
),
}
}
pub fn signing_unprovably_unavailable(
&self,
pub_key: ed25519::PublicKey,
expected_verification_key: Option<VerificationKeyShare>,
share_verified: bool,
) -> bool {
match self {
Status::Unreachable | Status::Outdated | Status::Reachable { .. } => false,
Status::ReachableLegacy { response } => !response.unprovable_signing_available(
&pub_key,
expected_verification_key,
share_verified,
),
}
}
}
#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
pub struct SignerResult<LS, TS, LC, TC> {
pub dkg_epoch_id: u64,
pub information: RawDealerInformation,
pub status: SignerStatus<LS, TS, LC, TC>,
}
impl<LS, TS, LC, TC> SignerResult<LS, TS, LC, TC> {
pub fn signer_unreachable(&self) -> bool {
matches!(self.status, SignerStatus::Unreachable)
}
pub fn malformed_details(&self) -> bool {
self.information.parse().is_err()
}
}
impl<LS, TS, LC, TC> SignerResult<LS, TS, LC, TC>
where
LC: LegacyChainResponse,
TC: ChainResponse,
{
pub fn unknown_chain_status(&self) -> bool {
let Ok(_) = self.information.parse() else {
return true;
};
if let SignerStatus::Tested { .. } = &self.status {
return false;
}
true
}
pub fn chain_available(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result
.local_chain_status
.chain_available(parsed_info.public_key)
}
pub fn chain_provably_stalled(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result
.local_chain_status
.chain_provably_stalled(parsed_info.public_key)
}
pub fn chain_unprovably_stalled(&self) -> bool {
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result.local_chain_status.chain_unprovably_stalled()
}
}
impl<LS, TS, LC, TC> SignerResult<LS, TS, LC, TC>
where
LS: LegacySignerResponse,
TS: SignerResponse,
{
pub fn unknown_signing_status(&self) -> bool {
let Ok(_) = self.information.parse() else {
return true;
};
if let SignerStatus::Tested { .. } = &self.status {
return false;
}
true
}
pub fn signing_available(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result.signing_status.signing_available(
parsed_info.public_key,
self.dkg_epoch_id,
parsed_info.verification_key_share,
parsed_info.share_verified,
)
}
pub fn signing_provably_unavailable(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result
.signing_status
.signing_provably_unavailable(parsed_info.public_key, self.dkg_epoch_id)
}
pub fn signing_unprovably_unavailable(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result.signing_status.signing_unprovably_unavailable(
parsed_info.public_key,
parsed_info.verification_key_share,
parsed_info.share_verified,
)
}
}
#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
pub enum SignerStatus<LS, TS, LC, TC> {
Unreachable,
ProvidedInvalidDetails,
Tested {
result: SignerTestResult<LS, TS, LC, TC>,
},
}
impl<LS, TS, LC, TC> SignerStatus<LS, TS, LC, TC> {
pub fn with_details(
self,
information: impl Into<RawDealerInformation>,
dkg_epoch_id: u64,
) -> SignerResult<LS, TS, LC, TC> {
SignerResult {
dkg_epoch_id,
status: self,
information: information.into(),
}
}
}
#[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
pub struct SignerTestResult<LS, TS, LC, TC> {
pub reported_version: String,
pub signing_status: Status<LS, TS>,
pub local_chain_status: Status<LC, TC>,
}
-27
View File
@@ -1,27 +0,0 @@
[package]
name = "nym-ecash-signer-check"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
readme.workspace = true
[dependencies]
futures = { workspace = true }
thiserror = { workspace = true }
semver = { workspace = true }
tokio = { workspace = true, features = ["time"] }
tracing = { workspace = true }
url = { workspace = true }
nym-validator-client = { path = "../client-libs/validator-client" }
nym-network-defaults = { path = "../network-defaults" }
nym-ecash-signer-check-types = { path = "../ecash-signer-check-types" }
[lints]
workspace = true
@@ -1,225 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::{LocalChainStatus, SigningStatus, TypedSignerResult};
use nym_ecash_signer_check_types::dealer_information::RawDealerInformation;
use nym_ecash_signer_check_types::status::{SignerStatus, SignerTestResult};
use nym_validator_client::client::NymApiClientExt;
use nym_validator_client::models::BinaryBuildInformationOwned;
use nym_validator_client::nyxd::contract_traits::dkg_query_client::{
ContractVKShare, DealerDetails,
};
use nym_validator_client::NymApiClient;
use std::time::Duration;
use tracing::{error, warn};
use url::Url;
pub(crate) mod chain_status {
// Dorina
pub(crate) const MINIMUM_VERSION_LEGACY: semver::Version = semver::Version::new(1, 1, 51);
// Gruyere
pub(crate) const MINIMUM_VERSION: semver::Version = semver::Version::new(1, 1, 64);
}
pub(crate) mod signing_status {
// Magura (possibly earlier)
pub(crate) const MINIMUM_LEGACY_VERSION: semver::Version = semver::Version::new(1, 1, 46);
// Gruyere
pub(crate) const MINIMUM_VERSION: semver::Version = semver::Version::new(1, 1, 64);
}
struct ClientUnderTest {
api_client: NymApiClient,
build_info: Option<BinaryBuildInformationOwned>,
}
impl ClientUnderTest {
pub(crate) fn new(api_url: &Url) -> Self {
ClientUnderTest {
api_client: NymApiClient::new(api_url.clone()),
build_info: None,
}
}
pub(crate) async fn try_retrieve_build_information(&mut self) -> bool {
match tokio::time::timeout(
Duration::from_secs(5),
self.api_client.nym_api.build_information(),
)
.await
{
Ok(Ok(build_information)) => {
self.build_info = Some(build_information);
true
}
Ok(Err(err)) => {
warn!("{}: failed to retrieve build information: {err}. the signer is most likely down", self.api_client.api_url());
false
}
Err(_timeout) => {
warn!(
"{}: timed out while attempting to retrieve build information",
self.api_client.api_url()
);
false
}
}
}
pub(crate) fn version(&self) -> Option<semver::Version> {
self.build_info.as_ref().and_then(|build_info| {
build_info
.build_version
.parse()
.inspect_err(|err| {
error!(
"ecash signer '{}' reports invalid version {}: {err}",
self.api_client.api_url(),
build_info.build_version
)
})
.ok()
})
}
pub(crate) fn supports_legacy_signing_status_query(&self) -> bool {
let Some(version) = self.version() else {
return false;
};
version >= signing_status::MINIMUM_LEGACY_VERSION
}
pub(crate) fn supports_signing_status_query(&self) -> bool {
let Some(version) = self.version() else {
return false;
};
version >= signing_status::MINIMUM_VERSION
}
pub(crate) fn supports_chain_status_query(&self) -> bool {
let Some(version) = self.version() else {
return false;
};
version >= chain_status::MINIMUM_VERSION
}
pub(crate) fn supports_legacy_chain_status_query(&self) -> bool {
let Some(version) = self.version() else {
return false;
};
version >= chain_status::MINIMUM_VERSION_LEGACY
}
pub(crate) async fn check_local_chain(&self) -> LocalChainStatus {
// check if it at least supports legacy query
if !self.supports_legacy_chain_status_query() {
return LocalChainStatus::Outdated;
}
// check if it supports the current query
if self.supports_chain_status_query() {
return match self.api_client.nym_api.get_chain_blocks_status().await {
Ok(status) => LocalChainStatus::Reachable {
response: Box::new(status),
},
Err(err) => {
warn!(
"{}: failed to retrieve local chain status: {err}",
self.api_client.api_url()
);
LocalChainStatus::Unreachable
}
};
}
// fallback to the legacy query
match self.api_client.nym_api.get_chain_status().await {
Ok(status) => LocalChainStatus::ReachableLegacy {
response: Box::new(status),
},
Err(err) => {
warn!(
"{}: failed to retrieve [legacy] local chain status: {err}",
self.api_client.api_url()
);
LocalChainStatus::Unreachable
}
}
}
pub(crate) async fn check_signing_status(&self) -> SigningStatus {
// check if it at least supports legacy query
if !self.supports_legacy_signing_status_query() {
return SigningStatus::Outdated;
}
// check if it supports the current query
if self.supports_signing_status_query() {
return match self.api_client.nym_api.get_signer_status().await {
Ok(response) => SigningStatus::Reachable {
response: Box::new(response),
},
Err(err) => {
warn!(
"{}: failed to retrieve signer chain status: {err}",
self.api_client.api_url()
);
SigningStatus::Unreachable
}
};
}
// fallback to the legacy query
match self.api_client.nym_api.get_signer_information().await {
Ok(status) => SigningStatus::ReachableLegacy {
response: Box::new(status),
},
Err(err) => {
warn!(
"{}: failed to retrieve [legacy] signer chain status: {err}",
self.api_client.api_url()
);
// NOTE: this might equally mean the signing is disabled
SigningStatus::Unreachable
}
}
}
}
pub(crate) async fn check_client(
dealer_details: DealerDetails,
dkg_epoch: u64,
contract_share: Option<&ContractVKShare>,
) -> TypedSignerResult {
let dealer_information = RawDealerInformation::new(&dealer_details, contract_share);
// 7. attempt to construct client instances out of them
let Ok(parsed_information) = dealer_information.parse() else {
return SignerStatus::ProvidedInvalidDetails.with_details(dealer_information, dkg_epoch);
};
let mut client = ClientUnderTest::new(&parsed_information.announce_address);
// 8. check basic connection status - can you retrieve build information?
if !client.try_retrieve_build_information().await {
return SignerStatus::Unreachable.with_details(dealer_information, dkg_epoch);
}
// 9. check perceived chain status
let local_chain_status = client.check_local_chain().await;
// 10. check signer status
let signing_status = client.check_signing_status().await;
SignerStatus::Tested {
result: SignerTestResult {
reported_version: client.version().map(|v| v.to_string()).unwrap_or_default(),
signing_status,
local_chain_status,
},
}
.with_details(dealer_information, dkg_epoch)
}
@@ -1,80 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::SignerCheckError;
use nym_crypto::asymmetric::ed25519;
use nym_validator_client::nyxd::contract_traits::dkg_query_client::{
ContractVKShare, DealerDetails, VerificationKeyShare,
};
use url::Url;
#[derive(Debug)]
pub struct RawDealerInformation {
pub announce_address: String,
pub owner_address: String,
pub node_index: u64,
pub public_key: String,
pub verification_key_share: Option<VerificationKeyShare>,
pub share_verified: bool,
}
impl RawDealerInformation {
pub fn new(
dealer_details: &DealerDetails,
contract_share: Option<&ContractVKShare>,
) -> RawDealerInformation {
RawDealerInformation {
announce_address: dealer_details.announce_address.clone(),
owner_address: dealer_details.address.to_string(),
node_index: dealer_details.assigned_index,
public_key: dealer_details.ed25519_identity.clone(),
verification_key_share: contract_share.map(|s| s.share.clone()),
share_verified: contract_share.map(|s| s.verified).unwrap_or(false),
}
}
pub fn parse(&self) -> Result<DealerInformation, SignerCheckError> {
Ok(DealerInformation {
announce_address: self.announce_address.parse().map_err(|source| {
SignerCheckError::InvalidDealerAddress {
dealer_url: self.announce_address.clone(),
source,
}
})?,
owner_address: self.owner_address.clone(),
node_index: self.node_index,
public_key: self.announce_address.parse().map_err(|source| {
SignerCheckError::InvalidDealerPubkey {
dealer_url: self.announce_address.clone(),
source,
}
})?,
verification_key_share: self.verification_key_share.clone(),
share_verified: self.share_verified,
})
}
}
#[derive(Debug)]
pub struct DealerInformation {
pub announce_address: Url,
pub owner_address: String,
pub node_index: u64,
pub public_key: ed25519::PublicKey,
// no need to parse it into the full type as it doesn't get us anything
pub verification_key_share: Option<VerificationKeyShare>,
pub share_verified: bool,
}
impl From<DealerInformation> for RawDealerInformation {
fn from(d: DealerInformation) -> Self {
RawDealerInformation {
announce_address: d.announce_address.to_string(),
owner_address: d.owner_address,
node_index: d.node_index,
public_key: d.public_key.to_base58_string(),
verification_key_share: d.verification_key_share,
share_verified: d.share_verified,
}
}
}
-24
View File
@@ -1,24 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_validator_client::nyxd::error::NyxdError;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum SignerCheckError {
#[error("failed to connect to nyxd chain due to invalid connection details: {source}")]
InvalidNyxdConnectionDetails { source: NyxdError },
#[error("failed to query the DKG contract: {source}")]
DKGContractQueryFailure { source: NyxdError },
}
impl SignerCheckError {
pub fn invalid_nyxd_connection_details(source: NyxdError) -> Self {
Self::InvalidNyxdConnectionDetails { source }
}
pub fn dkg_contract_query_failure(source: NyxdError) -> Self {
Self::DKGContractQueryFailure { source }
}
}
-94
View File
@@ -1,94 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::client_check::check_client;
use futures::stream::{FuturesUnordered, StreamExt};
use nym_network_defaults::NymNetworkDetails;
use nym_validator_client::nyxd::contract_traits::{DkgQueryClient, PagedDkgQueryClient};
use nym_validator_client::QueryHttpRpcNyxdClient;
use std::collections::HashMap;
use url::Url;
pub use error::SignerCheckError;
use nym_ecash_signer_check_types::status::{SignerResult, Status};
use nym_validator_client::ecash::models::EcashSignerStatusResponse;
use nym_validator_client::models::{
ChainBlocksStatusResponse, ChainStatusResponse, SignerInformationResponse,
};
mod client_check;
pub mod error;
pub type TypedSignerResult = SignerResult<
SignerInformationResponse,
EcashSignerStatusResponse,
ChainStatusResponse,
ChainBlocksStatusResponse,
>;
pub type LocalChainStatus = Status<ChainStatusResponse, ChainBlocksStatusResponse>;
pub type SigningStatus = Status<SignerInformationResponse, EcashSignerStatusResponse>;
pub struct SignersTestResult {
pub threshold: Option<u64>,
pub results: Vec<TypedSignerResult>,
}
pub async fn check_signers(
rpc_endpoint: Url,
// details such as denoms, prefixes, etc.
network_details: NymNetworkDetails,
) -> Result<SignersTestResult, SignerCheckError> {
// 1. create nyx client instance
let client = QueryHttpRpcNyxdClient::connect_with_network_details(
rpc_endpoint.as_str(),
network_details,
)
.map_err(SignerCheckError::invalid_nyxd_connection_details)?;
check_signers_with_client(&client).await
}
pub async fn check_signers_with_client<C>(client: &C) -> Result<SignersTestResult, SignerCheckError>
where
C: DkgQueryClient + Sync,
{
// 2. retrieve current dkg epoch
let dkg_epoch = client
.get_current_epoch()
.await
.map_err(SignerCheckError::dkg_contract_query_failure)?;
// 3. retrieve the dkg threshold as reference point
let threshold = client
.get_epoch_threshold(dkg_epoch.epoch_id)
.await
.map_err(SignerCheckError::dkg_contract_query_failure)?;
// 4. retrieve information on current DKG dealers (i.e. eligible signers)
let dealers = client
.get_all_current_dealers()
.await
.map_err(SignerCheckError::dkg_contract_query_failure)?;
// 5. retrieve their published keys (if available)
let shares: HashMap<_, _> = client
.get_all_verification_key_shares(dkg_epoch.epoch_id)
.await
.map_err(SignerCheckError::dkg_contract_query_failure)?
.into_iter()
.map(|share| (share.node_index, share))
.collect();
// 6. for each dealer attempt to perform the checks
let results = dealers
.into_iter()
.map(|d| {
let share = shares.get(&d.assigned_index);
check_client(d, dkg_epoch.epoch_id, share)
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<_>>()
.await;
Ok(SignersTestResult { threshold, results })
}
-73
View File
@@ -1,73 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::chain_status::LocalChainStatus;
use crate::dealer_information::RawDealerInformation;
use crate::signing_status::SigningStatus;
use std::time::Duration;
pub(crate) const STALE_RESPONSE_THRESHOLD: Duration = Duration::from_secs(5 * 60);
#[derive(Debug)]
pub struct SignerResult {
pub dkg_epoch_id: u64,
pub information: RawDealerInformation,
pub status: SignerStatus,
}
impl SignerResult {
pub fn chain_available(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result.local_chain_status.available(parsed_info.public_key)
}
pub fn signer_available(&self) -> bool {
let Ok(parsed_info) = self.information.parse() else {
return false;
};
let SignerStatus::Tested { result } = &self.status else {
return false;
};
result.signing_status.available(
parsed_info.public_key,
self.dkg_epoch_id,
parsed_info.verification_key_share,
parsed_info.share_verified,
)
}
}
#[derive(Debug)]
pub enum SignerStatus {
Unreachable,
ProvidedInvalidDetails,
Tested { result: SignerTestResult },
}
impl SignerStatus {
pub fn with_details(
self,
information: impl Into<RawDealerInformation>,
dkg_epoch_id: u64,
) -> SignerResult {
SignerResult {
dkg_epoch_id,
status: self,
information: information.into(),
}
}
}
#[derive(Debug)]
pub struct SignerTestResult {
pub reported_version: String,
pub signing_status: SigningStatus,
pub local_chain_status: LocalChainStatus,
}
-3
View File
@@ -47,7 +47,4 @@ workspace = true
default-features = false
[dev-dependencies]
anyhow = { workspace = true }
nym-compact-ecash = { path = "../nym_offline_compact_ecash" } # we need specific imports in tests
nym-test-utils = { path = "../test-utils" }
tokio = { workspace = true, features = ["full"] }
@@ -109,85 +109,3 @@ GATEWAY -> CLIENT
DONE(status)
*/
#[cfg(test)]
mod tests {
use super::*;
use crate::ClientControlRequest;
use futures::StreamExt;
use nym_test_utils::helpers::u64_seeded_rng;
use nym_test_utils::mocks::stream_sink::mock_streams;
use nym_test_utils::traits::{Leak, Timeboxed, TimeboxedSpawnable};
use tokio::join;
use tungstenite::Message;
#[tokio::test]
async fn basic_handshake() -> anyhow::Result<()> {
use anyhow::Context as _;
// solve the lifetime issue by just leaking the contents of the boxes
// which is perfectly fine in test
let client_rng = u64_seeded_rng(42).leak();
let gateway_rng = u64_seeded_rng(69).leak();
let client_keys = ed25519::KeyPair::new(client_rng).leak();
let gateway_keys = ed25519::KeyPair::new(gateway_rng).leak();
let (client_ws, gateway_ws) = mock_streams::<Message>();
// we need streams that return Result<Message, WsError>
let client_ws = client_ws.map(Ok);
let gateway_ws = gateway_ws.map(Ok);
let client_ws = client_ws.leak();
let gateway_ws = gateway_ws.leak();
let handshake_client = client_handshake(
client_rng,
client_ws,
client_keys,
*gateway_keys.public_key(),
false,
true,
TaskClient::dummy(),
);
let client_fut = handshake_client.spawn_timeboxed();
// we need to receive the first message so that it could be propagated to the gateway side of the handshake
let ClientControlRequest::RegisterHandshakeInitRequest {
protocol_version: _,
data,
} = (gateway_ws.next())
.timeboxed()
.await
.context("timeout")?
.context("no message!")??
.into_text()?
.parse::<ClientControlRequest>()?
else {
panic!("bad message")
};
let init_msg = data;
let handshake_gateway = gateway_handshake(
gateway_rng,
gateway_ws,
gateway_keys,
init_msg,
TaskClient::dummy(),
);
let gateway_fut = handshake_gateway.spawn_timeboxed();
let (client, gateway) = join!(client_fut, gateway_fut);
let client_key = client???;
let gateway_key = gateway???;
// ensure the created keys are the same
assert_eq!(client_key, gateway_key);
Ok(())
}
}
-7
View File
@@ -9,10 +9,8 @@ edition.workspace = true
license.workspace = true
[dependencies]
async-trait = { workspace = true }
bincode = { workspace = true }
defguard_wireguard_rs = { workspace = true }
dyn-clone = { workspace = true }
sqlx = { workspace = true, features = [
"runtime-tokio-rustls",
"sqlite",
@@ -23,7 +21,6 @@ sqlx = { workspace = true, features = [
] }
time = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["sync"], optional = true }
tracing = { workspace = true }
nym-credentials-interface = { path = "../credentials-interface" }
@@ -38,7 +35,3 @@ sqlx = { workspace = true, features = [
"macros",
"migrate",
] }
[features]
default = []
mock = ["tokio"]
@@ -1,19 +0,0 @@
/*
* Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: Apache-2.0
*/
DELETE FROM wireguard_peer WHERE client_id IS NULL;
CREATE TABLE wireguard_peer_new
(
public_key TEXT NOT NULL PRIMARY KEY UNIQUE,
allowed_ips BLOB NOT NULL,
client_id INTEGER REFERENCES clients(id) NOT NULL
);
INSERT INTO wireguard_peer_new (public_key, allowed_ips, client_id)
SELECT public_key, allowed_ips, client_id FROM wireguard_peer;
DROP TABLE wireguard_peer;
ALTER TABLE wireguard_peer_new RENAME TO wireguard_peer;
-13
View File
@@ -3,8 +3,6 @@
use std::str::FromStr;
use nym_credentials_interface::TicketType;
use crate::models::Client;
#[derive(Debug, PartialEq, sqlx::Type)]
@@ -17,17 +15,6 @@ pub enum ClientType {
ExitWireguard,
}
impl From<TicketType> for ClientType {
fn from(value: TicketType) -> Self {
match value {
TicketType::V1MixnetEntry => ClientType::EntryMixnet,
TicketType::V1MixnetExit => ClientType::ExitMixnet,
TicketType::V1WireguardEntry => ClientType::EntryWireguard,
TicketType::V1WireguardExit => ClientType::ExitWireguard,
}
}
}
impl FromStr for ClientType {
type Err = &'static str;
+2 -14
View File
@@ -20,18 +20,6 @@ pub enum GatewayStorageError {
#[error("the stored data associated with ticket {ticket_id} is malformed!")]
MalformedStoredTicketData { ticket_id: i64 },
#[error("Failed to convert from type of database: {field_key}")]
TypeConversion { field_key: &'static str },
#[error("Serialization failure for {field_key}")]
Serialize {
field_key: &'static str,
source: bincode::Error,
},
#[error("Deserialization failure for {field_key}")]
Deserialize {
field_key: &'static str,
source: bincode::Error,
},
#[error("Failed to convert from type of database: {0}")]
TypeConversion(String),
}
+74 -77
View File
@@ -1,7 +1,6 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use async_trait::async_trait;
use bandwidth::BandwidthManager;
use clients::{ClientManager, ClientType};
use models::{
@@ -16,10 +15,10 @@ use sqlx::{
sqlite::{SqliteAutoVacuum, SqliteSynchronous},
ConnectOptions,
};
use std::{path::Path, time::Duration};
use std::path::Path;
use tickets::TicketStorageManager;
use time::OffsetDateTime;
use tracing::{debug, error, log::LevelFilter};
use tracing::{debug, error};
pub mod bandwidth;
mod clients;
@@ -28,21 +27,11 @@ mod inboxes;
pub mod models;
mod shared_keys;
mod tickets;
pub mod traits;
mod wireguard_peers;
pub use error::GatewayStorageError;
pub use inboxes::InboxManager;
use crate::traits::{BandwidthGatewayStorage, InboxGatewayStorage, SharedKeyGatewayStorage};
fn make_bincode_serializer() -> impl bincode::Options {
use bincode::Options;
bincode::DefaultOptions::new()
.with_big_endian()
.with_varint_encoding()
}
// note that clone here is fine as upon cloning the same underlying pool will be used
#[derive(Clone)]
pub struct GatewayStorage {
@@ -82,21 +71,6 @@ impl GatewayStorage {
&self.wireguard_peer_manager
}
pub async fn handle_forget_me(
&self,
client_address: DestinationAddressBytes,
) -> Result<(), GatewayStorageError> {
let client_id = self.get_mixnet_client_id(client_address).await?;
self.inbox_manager()
.remove_messages_for_client(&client_address.as_base58_string())
.await?;
self.bandwidth_manager().remove_client(client_id).await?;
self.shared_key_manager()
.remove_shared_keys(&client_address.as_base58_string())
.await?;
Ok(())
}
/// Initialises `PersistentStorage` using the provided path.
///
/// # Arguments
@@ -118,7 +92,6 @@ impl GatewayStorage {
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.synchronous(SqliteSynchronous::Normal)
.auto_vacuum(SqliteAutoVacuum::Incremental)
.log_slow_statements(LevelFilter::Warn, Duration::from_millis(250))
.filename(database_path)
.create_if_missing(true)
.disable_statement_logging();
@@ -150,9 +123,8 @@ impl GatewayStorage {
}
}
#[async_trait]
impl SharedKeyGatewayStorage for GatewayStorage {
async fn get_mixnet_client_id(
impl GatewayStorage {
pub async fn get_mixnet_client_id(
&self,
client_address: DestinationAddressBytes,
) -> Result<i64, GatewayStorageError> {
@@ -162,7 +134,22 @@ impl SharedKeyGatewayStorage for GatewayStorage {
.await?)
}
async fn insert_shared_keys(
pub async fn handle_forget_me(
&self,
client_address: DestinationAddressBytes,
) -> Result<(), GatewayStorageError> {
let client_id = self.get_mixnet_client_id(client_address).await?;
self.inbox_manager()
.remove_messages_for_client(&client_address.as_base58_string())
.await?;
self.bandwidth_manager().remove_client(client_id).await?;
self.shared_key_manager()
.remove_shared_keys(&client_address.as_base58_string())
.await?;
Ok(())
}
pub async fn insert_shared_keys(
&self,
client_address: DestinationAddressBytes,
shared_keys: &SharedGatewayKey,
@@ -191,7 +178,7 @@ impl SharedKeyGatewayStorage for GatewayStorage {
Ok(client_id)
}
async fn get_shared_keys(
pub async fn get_shared_keys(
&self,
client_address: DestinationAddressBytes,
) -> Result<Option<PersistedSharedKeys>, GatewayStorageError> {
@@ -203,7 +190,7 @@ impl SharedKeyGatewayStorage for GatewayStorage {
}
#[allow(dead_code)]
async fn remove_shared_keys(
pub async fn remove_shared_keys(
&self,
client_address: DestinationAddressBytes,
) -> Result<(), GatewayStorageError> {
@@ -213,7 +200,7 @@ impl SharedKeyGatewayStorage for GatewayStorage {
Ok(())
}
async fn update_last_used_authentication_timestamp(
pub async fn update_last_used_authentication_timestamp(
&self,
client_id: i64,
last_used_authentication_timestamp: OffsetDateTime,
@@ -227,15 +214,12 @@ impl SharedKeyGatewayStorage for GatewayStorage {
Ok(())
}
async fn get_client(&self, client_id: i64) -> Result<Option<Client>, GatewayStorageError> {
pub async fn get_client(&self, client_id: i64) -> Result<Option<Client>, GatewayStorageError> {
let client = self.client_manager.get_client(client_id).await?;
Ok(client)
}
}
#[async_trait]
impl InboxGatewayStorage for GatewayStorage {
async fn store_message(
pub async fn store_message(
&self,
client_address: DestinationAddressBytes,
message: Vec<u8>,
@@ -246,7 +230,7 @@ impl InboxGatewayStorage for GatewayStorage {
Ok(())
}
async fn retrieve_messages(
pub async fn retrieve_messages(
&self,
client_address: DestinationAddressBytes,
start_after: Option<i64>,
@@ -258,22 +242,19 @@ impl InboxGatewayStorage for GatewayStorage {
Ok(messages)
}
async fn remove_messages(&self, ids: Vec<i64>) -> Result<(), GatewayStorageError> {
pub async fn remove_messages(&self, ids: Vec<i64>) -> Result<(), GatewayStorageError> {
for id in ids {
self.inbox_manager.remove_message(id).await?;
}
Ok(())
}
}
#[async_trait]
impl BandwidthGatewayStorage for GatewayStorage {
async fn create_bandwidth_entry(&self, client_id: i64) -> Result<(), GatewayStorageError> {
pub async fn create_bandwidth_entry(&self, client_id: i64) -> Result<(), GatewayStorageError> {
self.bandwidth_manager.insert_new_client(client_id).await?;
Ok(())
}
async fn set_expiration(
pub async fn set_expiration(
&self,
client_id: i64,
expiration: OffsetDateTime,
@@ -284,12 +265,12 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn reset_bandwidth(&self, client_id: i64) -> Result<(), GatewayStorageError> {
pub async fn reset_bandwidth(&self, client_id: i64) -> Result<(), GatewayStorageError> {
self.bandwidth_manager.reset_bandwidth(client_id).await?;
Ok(())
}
async fn get_available_bandwidth(
pub async fn get_available_bandwidth(
&self,
client_id: i64,
) -> Result<Option<PersistedBandwidth>, GatewayStorageError> {
@@ -299,7 +280,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
.await?)
}
async fn increase_bandwidth(
pub async fn increase_bandwidth(
&self,
client_id: i64,
amount: i64,
@@ -310,7 +291,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
.await?)
}
async fn revoke_ticket_bandwidth(
pub async fn revoke_ticket_bandwidth(
&self,
ticket_id: i64,
amount: i64,
@@ -321,7 +302,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
.await?)
}
async fn decrease_bandwidth(
pub async fn decrease_bandwidth(
&self,
client_id: i64,
amount: i64,
@@ -332,7 +313,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
.await?)
}
async fn insert_epoch_signers(
pub async fn insert_epoch_signers(
&self,
epoch_id: i64,
signer_ids: Vec<i64>,
@@ -343,7 +324,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn insert_received_ticket(
pub async fn insert_received_ticket(
&self,
client_id: i64,
received_at: OffsetDateTime,
@@ -363,11 +344,11 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(ticket_id)
}
async fn contains_ticket(&self, serial_number: &[u8]) -> Result<bool, GatewayStorageError> {
pub async fn contains_ticket(&self, serial_number: &[u8]) -> Result<bool, GatewayStorageError> {
Ok(self.ticket_manager.has_ticket_data(serial_number).await?)
}
async fn insert_ticket_verification(
pub async fn insert_ticket_verification(
&self,
ticket_id: i64,
signer_id: i64,
@@ -380,7 +361,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn update_rejected_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError> {
pub async fn update_rejected_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError> {
// set the ticket as rejected
self.ticket_manager.set_rejected_ticket(ticket_id).await?;
@@ -391,7 +372,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn update_verified_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError> {
pub async fn update_verified_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError> {
// 1. insert into verified table
self.ticket_manager
.insert_verified_ticket(ticket_id)
@@ -405,7 +386,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn remove_verified_ticket_binary_data(
pub async fn remove_verified_ticket_binary_data(
&self,
ticket_id: i64,
) -> Result<(), GatewayStorageError> {
@@ -415,7 +396,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn get_all_verified_tickets_with_sn(
pub async fn get_all_verified_tickets_with_sn(
&self,
) -> Result<Vec<VerifiedTicket>, GatewayStorageError> {
Ok(self
@@ -424,7 +405,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
.await?)
}
async fn get_all_proposed_tickets_with_sn(
pub async fn get_all_proposed_tickets_with_sn(
&self,
proposal_id: u32,
) -> Result<Vec<VerifiedTicket>, GatewayStorageError> {
@@ -434,7 +415,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
.await?)
}
async fn insert_redemption_proposal(
pub async fn insert_redemption_proposal(
&self,
tickets: &[VerifiedTicket],
proposal_id: u32,
@@ -457,7 +438,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn clear_post_proposal_data(
pub async fn clear_post_proposal_data(
&self,
proposal_id: u32,
resolved_at: OffsetDateTime,
@@ -481,11 +462,13 @@ impl BandwidthGatewayStorage for GatewayStorage {
Ok(())
}
async fn latest_proposal(&self) -> Result<Option<RedemptionProposal>, GatewayStorageError> {
pub async fn latest_proposal(&self) -> Result<Option<RedemptionProposal>, GatewayStorageError> {
Ok(self.ticket_manager.get_latest_redemption_proposal().await?)
}
async fn get_all_unverified_tickets(&self) -> Result<Vec<ClientTicket>, GatewayStorageError> {
pub async fn get_all_unverified_tickets(
&self,
) -> Result<Vec<ClientTicket>, GatewayStorageError> {
self.ticket_manager
.get_unverified_tickets()
.await?
@@ -494,21 +477,21 @@ impl BandwidthGatewayStorage for GatewayStorage {
.collect()
}
async fn get_all_unresolved_proposals(&self) -> Result<Vec<i64>, GatewayStorageError> {
pub async fn get_all_unresolved_proposals(&self) -> Result<Vec<i64>, GatewayStorageError> {
Ok(self
.ticket_manager
.get_all_unresolved_redemption_proposal_ids()
.await?)
}
async fn get_votes(&self, ticket_id: i64) -> Result<Vec<i64>, GatewayStorageError> {
pub async fn get_votes(&self, ticket_id: i64) -> Result<Vec<i64>, GatewayStorageError> {
Ok(self
.ticket_manager
.get_verification_votes(ticket_id)
.await?)
}
async fn get_signers(&self, epoch_id: i64) -> Result<Vec<i64>, GatewayStorageError> {
pub async fn get_signers(&self, epoch_id: i64) -> Result<Vec<i64>, GatewayStorageError> {
Ok(self.ticket_manager.get_epoch_signers(epoch_id).await?)
}
@@ -517,20 +500,34 @@ impl BandwidthGatewayStorage for GatewayStorage {
/// # Arguments
///
/// * `peer`: wireguard peer data to be stored
async fn insert_wireguard_peer(
/// * `with_client_id`: if the peer should have a corresponding client_id
/// (created with entry wireguard ticket) or live without one (or with an
/// exiting one), for temporary backwards compatibility.
pub async fn insert_wireguard_peer(
&self,
peer: &defguard_wireguard_rs::host::Peer,
client_type: ClientType,
) -> Result<i64, GatewayStorageError> {
with_client_id: bool,
) -> Result<Option<i64>, GatewayStorageError> {
let client_id = match self
.wireguard_peer_manager
.retrieve_peer(&peer.public_key.to_string())
.await?
{
Some(peer) => peer.client_id,
None => self.client_manager.insert_client(client_type).await?,
_ => {
if with_client_id {
Some(
self.client_manager
.insert_client(ClientType::EntryWireguard)
.await?,
)
} else {
None
}
}
};
let peer = WireguardPeer::from_defguard_peer(peer.clone(), client_id)?;
let mut peer = WireguardPeer::from(peer.clone());
peer.client_id = client_id;
self.wireguard_peer_manager.insert_peer(&peer).await?;
Ok(client_id)
}
@@ -540,7 +537,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
/// # Arguments
///
/// * `peer_public_key`: wireguard public key of the peer to be retrieved.
async fn get_wireguard_peer(
pub async fn get_wireguard_peer(
&self,
peer_public_key: &str,
) -> Result<Option<WireguardPeer>, GatewayStorageError> {
@@ -552,7 +549,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
}
/// Retrieves all wireguard peers.
async fn get_all_wireguard_peers(&self) -> Result<Vec<WireguardPeer>, GatewayStorageError> {
pub async fn get_all_wireguard_peers(&self) -> Result<Vec<WireguardPeer>, GatewayStorageError> {
let ret = self.wireguard_peer_manager.retrieve_all_peers().await?;
Ok(ret)
}
@@ -562,7 +559,7 @@ impl BandwidthGatewayStorage for GatewayStorage {
/// # Arguments
///
/// * `peer_public_key`: wireguard public key of the peer to be removed.
async fn remove_wireguard_peer(
pub async fn remove_wireguard_peer(
&self,
peer_public_key: &str,
) -> Result<(), GatewayStorageError> {
+67 -25
View File
@@ -1,7 +1,9 @@
// Copyright 2021-2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::{error::GatewayStorageError, make_bincode_serializer};
use std::time::SystemTime;
use crate::error::GatewayStorageError;
use nym_credentials_interface::{AvailableBandwidth, ClientTicket, CredentialSpendingData};
use nym_gateway_requests::shared_key::{LegacySharedKeys, SharedGatewayKey, SharedSymmetricKey};
use sqlx::FromRow;
@@ -110,24 +112,35 @@ impl TryFrom<UnverifiedTicketData> for ClientTicket {
#[derive(Debug, Clone, FromRow)]
pub struct WireguardPeer {
pub public_key: String,
pub preshared_key: Option<String>,
pub protocol_version: Option<i64>,
pub endpoint: Option<String>,
pub last_handshake: Option<OffsetDateTime>,
pub tx_bytes: i64,
pub rx_bytes: i64,
pub persistent_keepalive_interval: Option<i64>,
pub allowed_ips: Vec<u8>,
pub client_id: i64,
pub client_id: Option<i64>,
}
impl WireguardPeer {
pub fn from_defguard_peer(
value: defguard_wireguard_rs::host::Peer,
client_id: i64,
) -> Result<Self, crate::error::GatewayStorageError> {
Ok(WireguardPeer {
impl From<defguard_wireguard_rs::host::Peer> for WireguardPeer {
fn from(value: defguard_wireguard_rs::host::Peer) -> Self {
WireguardPeer {
public_key: value.public_key.to_string(),
allowed_ips: bincode::Options::serialize(make_bincode_serializer(), &value.allowed_ips)
.map_err(|source| crate::error::GatewayStorageError::Serialize {
field_key: "allowed_ips",
source,
})?,
client_id,
})
preshared_key: value.preshared_key.as_ref().map(|k| k.to_string()),
protocol_version: value.protocol_version.map(|v| v as i64),
endpoint: value.endpoint.map(|e| e.to_string()),
last_handshake: value.last_handshake.map(OffsetDateTime::from),
tx_bytes: value.tx_bytes as i64,
rx_bytes: value.rx_bytes as i64,
persistent_keepalive_interval: value.persistent_keepalive_interval.map(|v| v as i64),
allowed_ips: bincode::Options::serialize(
bincode::DefaultOptions::new(),
&value.allowed_ips,
)
.unwrap_or_default(),
client_id: None,
}
}
}
@@ -136,20 +149,49 @@ impl TryFrom<WireguardPeer> for defguard_wireguard_rs::host::Peer {
fn try_from(value: WireguardPeer) -> Result<Self, Self::Error> {
Ok(Self {
public_key: value.public_key.as_str().try_into().map_err(|_| {
Self::Error::TypeConversion {
field_key: "public_key",
}
})?,
public_key: value
.public_key
.as_str()
.try_into()
.map_err(|e| Self::Error::TypeConversion(format!("public key {e}")))?,
preshared_key: value
.preshared_key
.as_deref()
.map(TryFrom::try_from)
.transpose()
.map_err(|e| Self::Error::TypeConversion(format!("preshared key {e}")))?,
protocol_version: value
.protocol_version
.map(TryFrom::try_from)
.transpose()
.map_err(|e| Self::Error::TypeConversion(format!("protocol version {e}")))?,
endpoint: value
.endpoint
.as_deref()
.map(|e| e.parse())
.transpose()
.map_err(|e| Self::Error::TypeConversion(format!("endpoint {e}")))?,
last_handshake: value.last_handshake.map(SystemTime::from),
tx_bytes: value
.tx_bytes
.try_into()
.map_err(|e| Self::Error::TypeConversion(format!("tx bytes {e}")))?,
rx_bytes: value
.rx_bytes
.try_into()
.map_err(|e| Self::Error::TypeConversion(format!("rx bytes {e}")))?,
persistent_keepalive_interval: value
.persistent_keepalive_interval
.map(TryFrom::try_from)
.transpose()
.map_err(|e| {
Self::Error::TypeConversion(format!("persistent keepalive interval {e}"))
})?,
allowed_ips: bincode::Options::deserialize(
bincode::DefaultOptions::new(),
&value.allowed_ips,
)
.map_err(|source| Self::Error::Deserialize {
field_key: "allowed_ips",
source,
})?,
..Default::default()
.map_err(|e| Self::Error::TypeConversion(format!("allowed ips {e}")))?,
})
}
}
-511
View File
@@ -1,511 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use async_trait::async_trait;
use nym_credentials_interface::ClientTicket;
use nym_gateway_requests::SharedGatewayKey;
use nym_sphinx::DestinationAddressBytes;
use time::OffsetDateTime;
use crate::{
clients::ClientType,
models::{
Client, PersistedBandwidth, PersistedSharedKeys, RedemptionProposal, StoredMessage,
VerifiedTicket, WireguardPeer,
},
GatewayStorageError,
};
#[async_trait]
pub trait SharedKeyGatewayStorage {
async fn get_mixnet_client_id(
&self,
client_address: DestinationAddressBytes,
) -> Result<i64, GatewayStorageError>;
async fn insert_shared_keys(
&self,
client_address: DestinationAddressBytes,
shared_keys: &SharedGatewayKey,
) -> Result<i64, GatewayStorageError>;
async fn get_shared_keys(
&self,
client_address: DestinationAddressBytes,
) -> Result<Option<PersistedSharedKeys>, GatewayStorageError>;
#[allow(dead_code)]
async fn remove_shared_keys(
&self,
client_address: DestinationAddressBytes,
) -> Result<(), GatewayStorageError>;
async fn update_last_used_authentication_timestamp(
&self,
client_id: i64,
last_used_authentication_timestamp: OffsetDateTime,
) -> Result<(), GatewayStorageError>;
async fn get_client(&self, client_id: i64) -> Result<Option<Client>, GatewayStorageError>;
}
#[async_trait]
pub trait InboxGatewayStorage {
async fn store_message(
&self,
client_address: DestinationAddressBytes,
message: Vec<u8>,
) -> Result<(), GatewayStorageError>;
async fn retrieve_messages(
&self,
client_address: DestinationAddressBytes,
start_after: Option<i64>,
) -> Result<(Vec<StoredMessage>, Option<i64>), GatewayStorageError>;
async fn remove_messages(&self, ids: Vec<i64>) -> Result<(), GatewayStorageError>;
}
#[async_trait]
pub trait BandwidthGatewayStorage: dyn_clone::DynClone {
async fn create_bandwidth_entry(&self, client_id: i64) -> Result<(), GatewayStorageError>;
async fn set_expiration(
&self,
client_id: i64,
expiration: OffsetDateTime,
) -> Result<(), GatewayStorageError>;
async fn reset_bandwidth(&self, client_id: i64) -> Result<(), GatewayStorageError>;
async fn get_available_bandwidth(
&self,
client_id: i64,
) -> Result<Option<PersistedBandwidth>, GatewayStorageError>;
async fn increase_bandwidth(
&self,
client_id: i64,
amount: i64,
) -> Result<i64, GatewayStorageError>;
async fn revoke_ticket_bandwidth(
&self,
ticket_id: i64,
amount: i64,
) -> Result<(), GatewayStorageError>;
async fn decrease_bandwidth(
&self,
client_id: i64,
amount: i64,
) -> Result<i64, GatewayStorageError>;
async fn insert_epoch_signers(
&self,
epoch_id: i64,
signer_ids: Vec<i64>,
) -> Result<(), GatewayStorageError>;
async fn insert_received_ticket(
&self,
client_id: i64,
received_at: OffsetDateTime,
serial_number: Vec<u8>,
data: Vec<u8>,
) -> Result<i64, GatewayStorageError>;
async fn contains_ticket(&self, serial_number: &[u8]) -> Result<bool, GatewayStorageError>;
async fn insert_ticket_verification(
&self,
ticket_id: i64,
signer_id: i64,
verified_at: OffsetDateTime,
accepted: bool,
) -> Result<(), GatewayStorageError>;
async fn update_rejected_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError>;
async fn update_verified_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError>;
async fn remove_verified_ticket_binary_data(
&self,
ticket_id: i64,
) -> Result<(), GatewayStorageError>;
async fn get_all_verified_tickets_with_sn(
&self,
) -> Result<Vec<VerifiedTicket>, GatewayStorageError>;
async fn get_all_proposed_tickets_with_sn(
&self,
proposal_id: u32,
) -> Result<Vec<VerifiedTicket>, GatewayStorageError>;
async fn insert_redemption_proposal(
&self,
tickets: &[VerifiedTicket],
proposal_id: u32,
created_at: OffsetDateTime,
) -> Result<(), GatewayStorageError>;
async fn clear_post_proposal_data(
&self,
proposal_id: u32,
resolved_at: OffsetDateTime,
rejected: bool,
) -> Result<(), GatewayStorageError>;
async fn latest_proposal(&self) -> Result<Option<RedemptionProposal>, GatewayStorageError>;
async fn get_all_unverified_tickets(&self) -> Result<Vec<ClientTicket>, GatewayStorageError>;
async fn get_all_unresolved_proposals(&self) -> Result<Vec<i64>, GatewayStorageError>;
async fn get_votes(&self, ticket_id: i64) -> Result<Vec<i64>, GatewayStorageError>;
async fn get_signers(&self, epoch_id: i64) -> Result<Vec<i64>, GatewayStorageError>;
/// Insert a wireguard peer in the storage.
///
/// # Arguments
///
/// * `peer`: wireguard peer data to be stored
async fn insert_wireguard_peer(
&self,
peer: &defguard_wireguard_rs::host::Peer,
client_type: ClientType,
) -> Result<i64, GatewayStorageError>;
/// Tries to retrieve a particular peer with the given public key.
///
/// # Arguments
///
/// * `peer_public_key`: wireguard public key of the peer to be retrieved.
async fn get_wireguard_peer(
&self,
peer_public_key: &str,
) -> Result<Option<WireguardPeer>, GatewayStorageError>;
/// Retrieves all wireguard peers.
async fn get_all_wireguard_peers(&self) -> Result<Vec<WireguardPeer>, GatewayStorageError>;
/// Remove a wireguard peer from the storage.
///
/// # Arguments
///
/// * `peer_public_key`: wireguard public key of the peer to be removed.
async fn remove_wireguard_peer(&self, peer_public_key: &str)
-> Result<(), GatewayStorageError>;
}
#[cfg(feature = "mock")]
pub mod mock {
use std::{collections::HashMap, sync::Arc};
use tokio::sync::RwLock;
use super::*;
struct EcashSigner {
_epoch_id: i64,
_signer_id: i64,
}
struct ReceivedTicket {
client_id: i64,
_received_at: OffsetDateTime,
rejected: Option<bool>,
}
struct TicketData {
serial_number: Vec<u8>,
data: Option<Vec<u8>>,
}
struct TicketVerification {
_ticket_id: i64,
_signer_id: i64,
_verified_at: OffsetDateTime,
_accepted: bool,
}
#[derive(Default)]
pub struct MockGatewayStorage {
available_bandwidth: HashMap<i64, PersistedBandwidth>,
ecash_signers: Vec<EcashSigner>,
received_ticket: HashMap<i64, ReceivedTicket>,
ticket_data: HashMap<i64, TicketData>,
ticket_verification: HashMap<i64, TicketVerification>,
verified_tickets: Vec<i64>,
wireguard_peers: HashMap<String, WireguardPeer>,
clients: HashMap<i64, String>,
}
#[async_trait]
impl BandwidthGatewayStorage for Arc<RwLock<MockGatewayStorage>> {
async fn create_bandwidth_entry(&self, client_id: i64) -> Result<(), GatewayStorageError> {
self.write().await.available_bandwidth.insert(
client_id,
PersistedBandwidth {
client_id,
available: 0,
expiration: Some(OffsetDateTime::UNIX_EPOCH),
},
);
Ok(())
}
async fn set_expiration(
&self,
client_id: i64,
expiration: OffsetDateTime,
) -> Result<(), GatewayStorageError> {
if let Some(bw) = self.write().await.available_bandwidth.get_mut(&client_id) {
bw.expiration = Some(expiration);
}
Ok(())
}
async fn reset_bandwidth(&self, client_id: i64) -> Result<(), GatewayStorageError> {
if let Some(bw) = self.write().await.available_bandwidth.get_mut(&client_id) {
bw.available = 0;
bw.expiration = Some(OffsetDateTime::UNIX_EPOCH);
}
Ok(())
}
async fn get_available_bandwidth(
&self,
client_id: i64,
) -> Result<Option<PersistedBandwidth>, GatewayStorageError> {
Ok(self
.read()
.await
.available_bandwidth
.get(&client_id)
.cloned())
}
async fn increase_bandwidth(
&self,
client_id: i64,
amount: i64,
) -> Result<i64, GatewayStorageError> {
self.write()
.await
.available_bandwidth
.get_mut(&client_id)
.map(|bw| {
bw.available += amount;
bw.available
})
.ok_or(GatewayStorageError::InternalDatabaseError(
sqlx::Error::RowNotFound,
))
}
async fn revoke_ticket_bandwidth(
&self,
ticket_id: i64,
amount: i64,
) -> Result<(), GatewayStorageError> {
let mut guard = self.write().await;
if let Some(client_id) = guard
.received_ticket
.get(&ticket_id)
.map(|ticket| ticket.client_id)
{
if let Some(bw) = guard.available_bandwidth.get_mut(&client_id) {
bw.available -= amount;
}
}
Ok(())
}
async fn decrease_bandwidth(
&self,
client_id: i64,
amount: i64,
) -> Result<i64, GatewayStorageError> {
self.write()
.await
.available_bandwidth
.get_mut(&client_id)
.map(|bw| {
bw.available -= amount;
bw.available
})
.ok_or(GatewayStorageError::InternalDatabaseError(
sqlx::Error::RowNotFound,
))
}
async fn insert_epoch_signers(
&self,
_epoch_id: i64,
signer_ids: Vec<i64>,
) -> Result<(), GatewayStorageError> {
self.write()
.await
.ecash_signers
.extend(signer_ids.iter().map(|signer_id| EcashSigner {
_epoch_id,
_signer_id: *signer_id,
}));
Ok(())
}
async fn insert_received_ticket(
&self,
client_id: i64,
_received_at: OffsetDateTime,
serial_number: Vec<u8>,
data: Vec<u8>,
) -> Result<i64, GatewayStorageError> {
let mut guard = self.write().await;
let ticket_id = guard.received_ticket.len() as i64;
guard.received_ticket.insert(
ticket_id,
ReceivedTicket {
client_id,
_received_at,
rejected: None,
},
);
guard.ticket_data.insert(
ticket_id,
TicketData {
serial_number,
data: Some(data),
},
);
Ok(ticket_id)
}
async fn contains_ticket(&self, serial_number: &[u8]) -> Result<bool, GatewayStorageError> {
Ok(self
.read()
.await
.ticket_data
.values()
.any(|ticket_data| ticket_data.serial_number == serial_number))
}
async fn insert_ticket_verification(
&self,
_ticket_id: i64,
_signer_id: i64,
_verified_at: OffsetDateTime,
_accepted: bool,
) -> Result<(), GatewayStorageError> {
self.write().await.ticket_verification.insert(
_ticket_id,
TicketVerification {
_ticket_id,
_signer_id,
_verified_at,
_accepted,
},
);
Ok(())
}
async fn update_rejected_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError> {
let mut guard = self.write().await;
if let Some(ticket) = guard.received_ticket.get_mut(&ticket_id) {
ticket.rejected = Some(true);
}
guard.ticket_data.remove(&ticket_id);
Ok(())
}
async fn update_verified_ticket(&self, ticket_id: i64) -> Result<(), GatewayStorageError> {
let mut guard = self.write().await;
guard.verified_tickets.push(ticket_id);
guard.ticket_verification.remove(&ticket_id);
Ok(())
}
async fn remove_verified_ticket_binary_data(
&self,
ticket_id: i64,
) -> Result<(), GatewayStorageError> {
if let Some(ticket) = self.write().await.ticket_data.get_mut(&ticket_id) {
ticket.data = None;
}
Ok(())
}
async fn get_all_verified_tickets_with_sn(
&self,
) -> Result<Vec<VerifiedTicket>, GatewayStorageError> {
todo!()
}
async fn get_all_proposed_tickets_with_sn(
&self,
_proposal_id: u32,
) -> Result<Vec<VerifiedTicket>, GatewayStorageError> {
todo!()
}
async fn insert_redemption_proposal(
&self,
_tickets: &[VerifiedTicket],
_proposal_id: u32,
_created_at: OffsetDateTime,
) -> Result<(), GatewayStorageError> {
todo!()
}
async fn clear_post_proposal_data(
&self,
_proposal_id: u32,
_resolved_at: OffsetDateTime,
_rejected: bool,
) -> Result<(), GatewayStorageError> {
todo!()
}
async fn latest_proposal(&self) -> Result<Option<RedemptionProposal>, GatewayStorageError> {
todo!()
}
async fn get_all_unverified_tickets(
&self,
) -> Result<Vec<ClientTicket>, GatewayStorageError> {
todo!()
}
async fn get_all_unresolved_proposals(&self) -> Result<Vec<i64>, GatewayStorageError> {
todo!()
}
async fn get_votes(&self, _ticket_id: i64) -> Result<Vec<i64>, GatewayStorageError> {
todo!()
}
async fn get_signers(&self, _epoch_id: i64) -> Result<Vec<i64>, GatewayStorageError> {
todo!()
}
async fn insert_wireguard_peer(
&self,
peer: &defguard_wireguard_rs::host::Peer,
client_type: ClientType,
) -> Result<i64, GatewayStorageError> {
let mut guard = self.write().await;
let client_id =
if let Some(peer) = guard.wireguard_peers.get(&peer.public_key.to_string()) {
peer.client_id
} else {
let client_id = guard.clients.len() as i64;
guard.clients.insert(client_id, client_type.to_string());
client_id
};
guard.wireguard_peers.insert(
peer.public_key.to_string(),
WireguardPeer::from_defguard_peer(peer.clone(), client_id)?,
);
Ok(client_id)
}
async fn get_wireguard_peer(
&self,
peer_public_key: &str,
) -> Result<Option<WireguardPeer>, GatewayStorageError> {
Ok(self
.read()
.await
.wireguard_peers
.get(peer_public_key)
.cloned())
}
async fn get_all_wireguard_peers(&self) -> Result<Vec<WireguardPeer>, GatewayStorageError> {
todo!()
}
async fn remove_wireguard_peer(
&self,
peer_public_key: &str,
) -> Result<(), GatewayStorageError> {
self.write().await.wireguard_peers.remove(peer_public_key);
Ok(())
}
}
}
@@ -27,18 +27,15 @@ impl WgPeerManager {
pub(crate) async fn insert_peer(&self, peer: &WireguardPeer) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT OR IGNORE INTO wireguard_peer(public_key, allowed_ips, client_id)
VALUES (?, ?, ?);
INSERT OR IGNORE INTO wireguard_peer(public_key, preshared_key, protocol_version, endpoint, last_handshake, tx_bytes, rx_bytes, persistent_keepalive_interval, allowed_ips, client_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
UPDATE wireguard_peer
SET allowed_ips = ?, client_id = ?
SET preshared_key = ?, protocol_version = ?, endpoint = ?, last_handshake = ?, tx_bytes = ?, rx_bytes = ?, persistent_keepalive_interval = ?, allowed_ips = ?, client_id = ?
WHERE public_key = ?
"#,
peer.public_key,
peer.allowed_ips,
peer.client_id,
peer.allowed_ips,
peer.client_id,
peer.public_key, peer.preshared_key, peer.protocol_version, peer.endpoint, peer.last_handshake, peer.tx_bytes, peer.rx_bytes, peer.persistent_keepalive_interval, peer.allowed_ips, peer.client_id,
peer.preshared_key, peer.protocol_version, peer.endpoint, peer.last_handshake, peer.tx_bytes, peer.rx_bytes, peer.persistent_keepalive_interval, peer.allowed_ips, peer.client_id,
peer.public_key,
)
.execute(&self.connection_pool)
+28 -57
View File
@@ -6,9 +6,9 @@
//! The resolver itself is the set combination of the google, cloudflare, and quad9 endpoints
//! supporting DoH and DoT.
//!
//! This resolver supports a fallback mechanism where, should the DNS-over-TLS resolution fail, a
//! This resolver implements a fallback mechanism where, should the DNS-over-TLS resolution fail, a
//! followup resolution will be done using the hosts configured default (e.g. `/etc/resolve.conf` on
//! linux). This is disabled by default and can be enabled using [`enable_system_fallback`].
//! linux).
//!
//! Requires the `dns-over-https-rustls`, `webpki-roots` feature for the
//! `hickory-resolver` crate
@@ -93,14 +93,14 @@ pub struct HickoryDnsResolver {
// Tokio Runtime in initialization, so we must delay the actual
// construction of the resolver.
state: Arc<OnceCell<TokioResolver>>,
fallback: Option<Arc<OnceCell<TokioResolver>>>,
fallback: Arc<OnceCell<TokioResolver>>,
dont_use_shared: bool,
}
impl Resolve for HickoryDnsResolver {
fn resolve(&self, name: Name) -> Resolving {
let resolver = self.state.clone();
let maybe_fallback = self.fallback.clone();
let fallback = self.fallback.clone();
let independent = self.dont_use_shared;
Box::pin(async move {
let resolver = resolver.get_or_try_init(|| {
@@ -117,30 +117,23 @@ impl Resolve for HickoryDnsResolver {
let lookup = match resolver.lookup_ip(name.as_str()).await {
Ok(res) => res,
Err(e) => {
if let Some(ref fallback) = maybe_fallback {
// on failure use the fall back system configured DNS resolver
if !e.is_no_records_found() {
warn!("primary DNS failed w/ error {e}: using system fallback");
}
let resolver = fallback.get_or_try_init(|| {
// using a closure here is slightly gross, but this makes sure that if the
// lazy-init returns an error it can be handled by the client
if independent {
new_resolver_system()
} else {
Ok(SHARED_RESOLVER
.fallback
.as_ref()
.ok_or(e)? // if the shared resolver has no fallback return the original error
.get_or_try_init(new_resolver_system)?
.clone())
}
})?;
resolver.lookup_ip(name.as_str()).await?
} else {
return Err(e.into());
// on failure use the fall back system configured DNS resolver
if !e.is_no_records_found() {
warn!("primary DNS failed w/ error {e}: using system fallback");
}
let resolver = fallback.get_or_try_init(|| {
// using a closure here is slightly gross, but this makes sure that if the
// lazy-init returns an error it can be handled by the client
if independent {
new_resolver_system()
} else {
Ok(SHARED_RESOLVER
.fallback
.get_or_try_init(new_resolver_system)?
.clone())
}
})?;
resolver.lookup_ip(name.as_str()).await?
}
};
@@ -169,17 +162,14 @@ impl HickoryDnsResolver {
let lookup = match resolver.lookup_ip(name).await {
Ok(res) => res,
Err(e) => {
if let Some(ref fallback) = self.fallback {
// on failure use the fall back system configured DNS resolver
if !e.is_no_records_found() {
warn!("primary DNS failed w/ error {e}: using system fallback");
}
let resolver = fallback.get_or_try_init(|| self.new_resolver_system())?;
resolver.lookup_ip(name).await?
} else {
return Err(e.into());
// on failure use the fall back system configured DNS resolver
if !e.is_no_records_found() {
warn!("primary DNS failed w/ error {e}: using system fallback");
}
let resolver = self
.fallback
.get_or_try_init(|| self.new_resolver_system())?;
resolver.lookup_ip(name).await?
}
};
@@ -203,34 +193,15 @@ impl HickoryDnsResolver {
}
fn new_resolver_system(&self) -> Result<TokioResolver, HickoryDnsError> {
if self.dont_use_shared || SHARED_RESOLVER.fallback.is_none() {
if self.dont_use_shared {
new_resolver_system()
} else {
Ok(SHARED_RESOLVER
.fallback
.as_ref()
.unwrap()
.get_or_try_init(new_resolver_system)?
.clone())
}
}
/// Enable fallback to the system default resolver if the primary (DoX) resolver fails
pub fn enable_system_fallback(&mut self) -> Result<(), HickoryDnsError> {
self.fallback = Some(Default::default());
let _ = self
.fallback
.as_ref()
.unwrap()
.get_or_try_init(new_resolver_system)?;
Ok(())
}
/// Disable fallback resolution. If the primary resolver fails the error is
/// returned immediately
pub fn disable_system_fallback(&mut self) {
self.fallback = None;
}
}
/// Create a new resolver with a custom DoT based configuration. The options are overridden to look
-5
View File
@@ -173,10 +173,6 @@ mod path;
#[cfg(not(target_arch = "wasm32"))]
pub use dns::{HickoryDnsError, HickoryDnsResolver};
// helper for generating user agent based on binary information
#[doc(hidden)]
pub use nym_bin_common::bin_info;
/// Default HTTP request connection timeout.
///
/// The timeout is relatively high as we are often making requests over the mixnet, where latency is
@@ -612,7 +608,6 @@ impl Client {
current_idx: Arc::new(Default::default()),
reqwest_client: self.reqwest_client.clone(),
#[cfg(feature = "tunneling")]
front: self.front.clone(),
retry_limit: self.retry_limit,
+1 -11
View File
@@ -16,20 +16,10 @@ pub struct UserAgent {
pub version: String,
/// client platform
pub platform: String,
/// source commit version for the calling crate / subsystem
/// source commit version for the calling calling crate / subsystem
pub git_commit: String,
}
/// Create `UserAgent` based on the caller's crate information
// we can't use normal function as then `application` and `version` would correspond
// of that of `nym-http-api-client` lib
#[macro_export]
macro_rules! generate_user_agent {
() => {
$crate::UserAgent::from($crate::bin_info!())
};
}
#[derive(Clone, Debug, thiserror::Error)]
#[error("invalid user agent string: {0}")]
pub struct UserAgentError(String);
@@ -146,12 +146,6 @@ pub struct OutputParams {
pub output: Option<Output>,
}
impl OutputParams {
pub fn get_output(&self) -> Output {
self.output.unwrap_or_default()
}
}
impl Output {
pub fn to_response<T: Serialize>(self, data: T) -> FormattedResponse<T> {
match self {
+2 -2
View File
@@ -25,8 +25,8 @@ pub enum NymIdError {
#[error("attempted to import an expired credential (it expired on {expiration})")]
ExpiredCredentialImport { expiration: Date },
#[error("could not import ticketbook expiring at {date} for epoch {epoch_id} since we do not have corresponding expiration date signatures")]
MissingExpirationDateSignatures { date: Date, epoch_id: u64 },
#[error("could not import ticketbook expiring at {date} since we do not have corresponding expiration date signatures")]
MissingExpirationDateSignatures { date: Date },
#[error("could not import ticketbook for epoch {epoch_id} since we do not have corresponding coin index signatures")]
MissingCoinIndexSignatures { epoch_id: u64 },
@@ -99,7 +99,7 @@ where
// in order to import the ticketbook we MUST have the appropriate signatures in the storage already
if credentials_store
.get_expiration_date_signatures(ticketbook.expiration_date(), ticketbook.epoch_id())
.get_expiration_date_signatures(ticketbook.expiration_date())
.await
.map_err(|source| NymIdError::StorageError {
source: Box::new(source),
@@ -108,7 +108,6 @@ where
{
return Err(NymIdError::MissingExpirationDateSignatures {
date: ticketbook.expiration_date(),
epoch_id: ticketbook.epoch_id(),
});
}
-2
View File
@@ -14,7 +14,6 @@ pin-project = { workspace = true }
sha2 = { workspace = true }
snow = { workspace = true }
strum = { workspace = true, features = ["derive"] }
strum_macros = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["net", "io-util", "time"] }
tokio-util = { workspace = true, features = ["codec"] }
@@ -28,7 +27,6 @@ anyhow = { workspace = true }
tokio = { workspace = true, features = ["full"] }
rand_chacha = { workspace = true }
nym-crypto = { path = "../crypto", features = ["rand"] }
nym-test-utils = { path = "../test-utils" }
[lints]
+1 -1
View File
@@ -13,7 +13,7 @@ use nym_crypto::asymmetric::x25519;
use nym_noise_keys::{NoiseVersion, VersionedNoiseKey};
use snow::params::NoiseParams;
use strum_macros::{EnumIter, FromRepr};
use strum::{EnumIter, FromRepr};
#[derive(Default, Debug, Clone, Copy, EnumIter, FromRepr, Eq, PartialEq)]
#[repr(u8)]
+1 -1
View File
@@ -5,7 +5,7 @@ use crate::config::NoisePattern;
use crate::error::NoiseError;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use nym_noise_keys::NoiseVersion;
use strum_macros::FromRepr;
use strum::FromRepr;
#[derive(Debug)]
pub struct NymNoiseFrame {

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