diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fa32e7..e2c39eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Nym mixnet transport (`transports.nym`) for outbound peer links + tunneled through a local `nym-socks5-client` SOCKS5 proxy into the + Nym mixnet, as a privacy transport alongside Tor. Outbound-only and + not platform-gated, it reuses the existing FMP framing and adds no new + crate dependencies. A single-container example + (`examples/sidecar-nostr-mixnet-relay/`) demonstrates FIPS peering + across the mixnet end to end. - Typed `RejectReason` classification for receive-path silent-rejection sites across the node. Each rejection-and-return path now passes a typed reason to `NodeStats::record_reject`, which routes it to a diff --git a/README.md b/README.md index d37dcdb..66e69aa 100644 --- a/README.md +++ b/README.md @@ -169,6 +169,12 @@ and [testing/README.md](testing/README.md). reachable exclusively over the FIPS mesh. The relay container shares the FIPS sidecar's network namespace and is isolated from the host network. +- **[examples/sidecar-nostr-mixnet-relay/](examples/sidecar-nostr-mixnet-relay/)** — + Single-container demo of FIPS peering through a **mixnet** + (implemented with [Nym](https://nym.com/)): the FIPS daemon, the mixnet + proxy, and a strfry Nostr relay all in one isolated container, with + the direct route to the peer firewalled off so traffic provably + crosses the mixnet. - **[examples/k8s-sidecar/](examples/k8s-sidecar/)** — Run FIPS as a Kubernetes Pod sidecar. The sidecar creates `fips0` in the Pod's shared network namespace so every other container in the diff --git a/examples/sidecar-nostr-mixnet-relay/.env b/examples/sidecar-nostr-mixnet-relay/.env new file mode 100644 index 0000000..31b963d --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/.env @@ -0,0 +1,37 @@ +# FIPS-over-Nym-mixnet demo configuration. +# Override these values or create a .env.local file. + +# Node identity — generate with: fipsctl keygen +# Must be set before running: export FIPS_NSEC= +FIPS_NSEC= + +# Peer configuration (leave FIPS_PEER_NPUB empty for standalone operation). +# The peer MUST expose a TCP endpoint in nym mode — the Nym SOCKS5 proxy +# tunnels TCP streams. Find more public peers at https://join.fips.network/ +# Default peer test-us03 exposes tcp:54.183.70.180:443 and udp:…:2121. +# For udp mode (see FIPS_PEER_TRANSPORT below) change FIPS_PEER_ADDR to +# 54.183.70.180:2121. +# The alias doubles as the peer's .fips hostname (.fips), so it +# must be a plain hostname label — no dots. +FIPS_PEER_NPUB=npub136yqae6na688fs75g95ppps3lxe07fvxefj77938zf47uhm6074sxw8ctm +FIPS_PEER_ADDR=54.183.70.180:443 +FIPS_PEER_ALIAS=test-us03 + +# Transport — THE switch that selects mixnet vs. direct: nym | tcp | udp +# nym : peer traffic goes through the Nym mixnet via the in-container +# nym-socks5-client (started automatically, before FIPS). DEFAULT. +# tcp : direct TCP to FIPS_PEER_ADDR; the nym client is NOT started. +# udp : direct UDP — also set FIPS_PEER_ADDR to the peer's :2121 endpoint. +# To go back to a direct link, just set this to tcp (or udp) and re-run +# `docker compose up`. Nothing else needs to change for tcp. +FIPS_PEER_TRANSPORT=nym + +# ----- Nym mixnet (only used when FIPS_PEER_TRANSPORT=nym) ----- +# Network-requester service provider. Leave empty to auto-discover the +# best-scored provider from https://harbourmaster.nymtech.net/ at startup. +NYM_SERVICE_PROVIDER= +NYM_CLIENT_ID=fips-nym-client +FIPS_NYM_SOCKS5_ADDR=127.0.0.1:1080 + +# Logging +RUST_LOG=info diff --git a/examples/sidecar-nostr-mixnet-relay/.gitignore b/examples/sidecar-nostr-mixnet-relay/.gitignore new file mode 100644 index 0000000..11ee758 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/.gitignore @@ -0,0 +1 @@ +.env.local diff --git a/examples/sidecar-nostr-mixnet-relay/Dockerfile b/examples/sidecar-nostr-mixnet-relay/Dockerfile new file mode 100644 index 0000000..e3825e5 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/Dockerfile @@ -0,0 +1,151 @@ +# Single-container FIPS-over-Nym-mixnet demo. +# +# Everything runs in ONE container: the FIPS daemon, the nym-socks5-client +# mixnet proxy, the strfry Nostr relay, nginx, and dnsmasq. The entrypoint +# starts them in strict order so the SOCKS5 proxy is up before FIPS dials +# its peer through the mixnet. +# +# Platform: the image builds NATIVE for the host. The FIPS daemon must NOT +# run under emulation — Rosetta mis-translates the amd64 ChaCha20-Poly1305 +# assembly (ring/BoringSSL), silently failing AEAD on larger frames (bloom +# filter announces are the first casualty). fips is therefore always +# compiled for the native arch. +# +# nym-socks5-client is the official prebuilt amd64 binary (Nym ships no +# other arch). On amd64 hosts it runs natively; on arm64 (Apple Silicon) +# it runs via Docker Desktop's binfmt/Rosetta handler, with the x86-64 +# loader + glibc copied in from an amd64 stage. The nym client tolerates +# emulation; fips does not. + +# ── Build stage: compile FIPS from source ── +FROM rust:1.94-slim-trixie AS builder + +# bluer (BLE) and rustables (nftables) are unconditional dependencies on +# glibc Linux: bluer needs the dbus headers, rustables runs bindgen +# (libclang) against the libnftnl headers. +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + pkg-config libdbus-1-dev libnftnl-dev libclang-dev clang && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /build +COPY Cargo.toml Cargo.lock rust-toolchain.toml build.rs ./ +COPY src ./src + +RUN cargo build --release && \ + cp target/release/fips target/release/fipsctl target/release/fipstop /usr/local/bin/ + +# ── strfry stage: collect the binary and its musl runtime ── +# The official strfry image is alpine (musl) based; the runtime stage below +# is debian (glibc), so the musl dynamic loader and the exact set of shared +# libraries strfry links against must come along. +FROM ghcr.io/hoytech/strfry:latest AS strfry +RUN mkdir -p /strfry-libs && \ + ldd /app/strfry | awk '$3 ~ /^\// {print $3}' | xargs -I{} cp {} /strfry-libs/ && \ + cp /lib/ld-musl-*.so.1 /strfry-libs/ + +# ── nym stage: official prebuilt amd64 binary + its glibc runtime ── +# Pinned amd64-only stage regardless of host arch. Collects the x86-64 +# dynamic loader and the binary's library closure so the binary can run +# inside the native (possibly arm64) runtime image via binfmt/Rosetta. +FROM --platform=linux/amd64 debian:trixie-slim AS nym +RUN apt-get update && \ + apt-get install -y --no-install-recommends ca-certificates curl && \ + rm -rf /var/lib/apt/lists/* +# Bumping the version = edit the tag in this URL. +RUN curl -sSL --retry 3 \ + "https://github.com/nymtech/nym/releases/download/nym-binaries-v2026.11-xynomizithra/nym-socks5-client" \ + -o /nym-socks5-client && \ + chmod +x /nym-socks5-client +RUN mkdir -p /nym-rt/lib64 /nym-rt/libs && \ + cp /lib64/ld-linux-x86-64.so.2 /nym-rt/lib64/ && \ + ldd /nym-socks5-client | awk '$3 ~ /^\// {print $3}' | xargs -I{} cp {} /nym-rt/libs/ + +# ── Runtime stage ── +FROM debian:trixie-slim + +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + ca-certificates \ + iproute2 iputils-ping dnsutils dnsmasq iptables \ + openssh-client openssh-server python3 \ + tcpdump netcat-openbsd curl jq iperf3 nginx && \ + rm -rf /var/lib/apt/lists/* + +# Install nak (Nostr Army Knife) — detect arch and download the correct binary. +# Asset naming: nak--linux- +RUN ARCH=$(dpkg --print-architecture) && \ + case "$ARCH" in \ + amd64) NAK_ARCH="linux-amd64" ;; \ + arm64) NAK_ARCH="linux-arm64" ;; \ + armhf) NAK_ARCH="linux-arm" ;; \ + *) echo "Unsupported arch: $ARCH" && exit 1 ;; \ + esac && \ + NAK_VERSION=$(curl -sSL --retry 3 \ + "https://api.github.com/repos/fiatjaf/nak/releases/latest" \ + | grep '"tag_name"' | head -1 | sed 's/.*"tag_name": *"\(.*\)".*/\1/') && \ + echo "Installing nak ${NAK_VERSION} for ${NAK_ARCH}" && \ + curl -sSL --retry 3 \ + "https://github.com/fiatjaf/nak/releases/download/${NAK_VERSION}/nak-${NAK_VERSION}-${NAK_ARCH}" \ + -o /usr/local/bin/nak && \ + chmod +x /usr/local/bin/nak && \ + nak --version + +# nym-socks5-client: prebuilt amd64 binary plus its x86-64 loader/glibc. +# On amd64 these COPYs overwrite identical files; on arm64 they add the +# x86-64 runtime alongside the native one (paths don't collide). The +# version check doubles as a binfmt/Rosetta smoke test on arm64 hosts. +COPY --from=nym /nym-socks5-client /usr/local/bin/nym-socks5-client +COPY --from=nym /nym-rt/lib64/ /lib64/ +COPY --from=nym /nym-rt/libs/ /lib/x86_64-linux-gnu/ +RUN echo "Installed:" && /usr/local/bin/nym-socks5-client --version + +# Setup SSH server with no authentication (test only!) +RUN mkdir -p /var/run/sshd && \ + ssh-keygen -A && \ + sed -i 's/#PermitRootLogin prohibit-password/PermitRootLogin yes/' /etc/ssh/sshd_config && \ + sed -i 's/#PermitEmptyPasswords no/PermitEmptyPasswords yes/' /etc/ssh/sshd_config && \ + sed -i 's/UsePAM yes/UsePAM no/' /etc/ssh/sshd_config && \ + passwd -d root + +# dnsmasq: forward .fips to FIPS daemon, everything else to Docker DNS +RUN printf '%s\n' \ + 'port=53' \ + 'listen-address=127.0.0.1' \ + 'bind-interfaces' \ + 'server=/fips/127.0.0.1#5354' \ + 'server=127.0.0.11' \ + 'no-resolv' \ + >> /etc/dnsmasq.conf + +# strfry: binary, musl loader, and its libraries in a private directory. +# /etc/ld-musl-.path tells the musl loader where to search, keeping +# the musl libraries invisible to the system glibc loader. +COPY --from=strfry /app/strfry /usr/local/bin/strfry +COPY --from=strfry /strfry-libs/ /opt/strfry-libs/ +RUN mv /opt/strfry-libs/ld-musl-*.so.1 /lib/ && \ + echo "/opt/strfry-libs" > /etc/ld-musl-$(uname -m).path && \ + mkdir -p /usr/src/app/strfry-db && \ + strfry --version + +# nginx: reverse proxy port 80 (IPv4 + IPv6) → strfry 127.0.0.1:7777 +RUN printf 'server {\n\ + listen 80;\n\ + listen [::]:80;\n\ + location / {\n\ + proxy_pass http://127.0.0.1:7777;\n\ + proxy_http_version 1.1;\n\ + proxy_read_timeout 1d;\n\ + proxy_send_timeout 1d;\n\ + proxy_set_header Upgrade $http_upgrade;\n\ + proxy_set_header Connection "Upgrade";\n\ + proxy_set_header Host $host;\n\ + }\n\ +}\n' > /etc/nginx/conf.d/nostr-relay.conf && \ + rm -f /etc/nginx/sites-enabled/default + +COPY --from=builder /usr/local/bin/fips /usr/local/bin/fipsctl /usr/local/bin/fipstop /usr/local/bin/ + +COPY examples/sidecar-nostr-mixnet-relay/entrypoint.sh /entrypoint.sh + +ENTRYPOINT ["/entrypoint.sh"] diff --git a/examples/sidecar-nostr-mixnet-relay/Dockerfile.dockerignore b/examples/sidecar-nostr-mixnet-relay/Dockerfile.dockerignore new file mode 100644 index 0000000..0a4314b --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/Dockerfile.dockerignore @@ -0,0 +1,7 @@ +target/ +target-*/ +.git/ +.github/ +testing/ +examples/ +!examples/sidecar-nostr-mixnet-relay/ diff --git a/examples/sidecar-nostr-mixnet-relay/README.md b/examples/sidecar-nostr-mixnet-relay/README.md new file mode 100644 index 0000000..0c796a6 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/README.md @@ -0,0 +1,173 @@ +# FIPS over a Mixnet — Single-Container Demo (Nym) + +An isolated environment demonstrating how FIPS peer traffic can travel +through a **mixnet** — a network that hides traffic patterns by routing +each packet through several relays with cover traffic and timing +obfuscation. The mixnet here is [Nym](https://nym.com/), but the FIPS side +is transport-agnostic: it just sees a SOCKS5 proxy, so any mixnet exposing +one would slot in the same way. + +**Everything runs in one Docker container**: the FIPS daemon, the mixnet +proxy (`nym-socks5-client`), a [strfry](https://github.com/hoytech/strfry) +Nostr relay behind nginx, and dnsmasq. + +``` +┌────────────────────────── one container ───────────────────────────┐ +│ │ +│ nginx :80 ──► strfry :7777 (Nostr relay, fips0-only) │ +│ │ +│ fips daemon ── transports.nym ──► nym-socks5-client :1080 │ +│ │ │ │ +│ ▼ ▼ Sphinx packets │ +│ fips0 (TUN, fd00::/8) Nym gateway ► 3 mix hops ► │ +│ network requester ► peer (TCP) │ +│ │ +│ iptables: direct route to the peer is DROPped — the FIPS link │ +│ can only exist through the mixnet. │ +└────────────────────────────────────────────────────────────────────┘ +``` + +How the pieces interlock: + +- The FIPS **nym transport** dials peers through a local SOCKS5 proxy; the + proxy routes each TCP stream through the mixnet (gateway → 3 mix hops → + network requester), which performs the final TCP connection to the peer. + The peer address must therefore be a **TCP endpoint** — find public peers + at . +- The `nym-socks5-client` is started by the entrypoint **only when the + generated FIPS config contains a `transports.nym` block** + (`FIPS_PEER_TRANSPORT=nym`), and always **before** the FIPS daemon, so + the proxy is listening by the time FIPS dials. +- In nym mode, iptables **drops the direct route to the peer**: if the peer + handshake completes, the traffic provably went through the mixnet. + +## Quick start + +```bash +# 1. Generate a node identity (any machine with fipsctl, or reuse one): +fipsctl keygen + +# 2. Put the nsec into the environment: +export FIPS_NSEC= + +# 3. Build and run (native image; FIPS compiles for your host's arch): +docker compose up --build +``` + +Watch the logs: the entrypoint auto-discovers a Nym service provider, +bootstraps the SOCKS5 client (`Nym SOCKS5 proxy ready …`), and only then +starts FIPS. After the mixnet handshake completes (can take 30–120 s): + +```bash +docker compose exec fips fipsctl show transports # nym transport: up +docker compose exec fips fipsctl show peers # test-us03: active +``` + +## Switching transport: mixnet ↔ direct (TCP/UDP) + +The single knob is `FIPS_PEER_TRANSPORT` in `.env` (or an inline override). +It selects how FIPS reaches the peer **and** whether the mixnet proxy runs +at all — the two are always in sync. + +```bash +# Default — through the Nym mixnet (anonymized, ~1-2 s RTT): +FIPS_PEER_TRANSPORT=nym docker compose up -d # or just `docker compose up -d` + +# Direct TCP (no mixnet, ~50-300 ms RTT). The nym client is NOT started: +FIPS_PEER_TRANSPORT=tcp docker compose up -d + +# Direct UDP — also point FIPS_PEER_ADDR at the peer's UDP endpoint: +FIPS_PEER_TRANSPORT=udp FIPS_PEER_ADDR=54.183.70.180:2121 docker compose up -d +``` + +What changes under the hood for each value: + +| `FIPS_PEER_TRANSPORT` | nym client | FIPS config block | peer endpoint used | direct route to peer | +| --- | --- | --- | --- | --- | +| `nym` (default) | started, before FIPS | `transports.nym` | `FIPS_PEER_ADDR` (TCP) via SOCKS5 | **firewalled off** | +| `tcp` | not started | `transports.tcp` | `FIPS_PEER_ADDR` (TCP) direct | allowed | +| `udp` | not started | `transports.udp` | `FIPS_PEER_ADDR` (UDP `:2121`) direct | allowed | + +To **switch back to a direct link**, set the value to `tcp` (no other change) +or `udp` (and swap `FIPS_PEER_ADDR` to the `:2121` endpoint), then re-run +`docker compose up -d`. To **return to the mixnet**, set it back to `nym`. +Persist your choice by editing `.env` instead of prefixing the command. +The same node can be compared both ways — direct shows ~50-300 ms RTT, +the mixnet ~1-2 s, which is the visible signature that traffic is routing +through the Sphinx mix hops. + +## Verifying the traffic really crosses the mixnet + +```bash +# The direct route to the peer is dropped — the only way packets reach +# the peer is via the nym-socks5-client: +docker compose exec fips iptables -L OUTPUT -v -n # DROP rule for peer IP + +# Mixnet activity (Sphinx packet flow) in the nym client output: +docker compose logs fips | grep -i nym + +# End-to-end data plane across the mesh. FIPS addresses every node by its +# key as .fips (each npub maps into fd00::/8); short names like +# `test-us03` are only local aliases for the peer you configured. Pick a +# node you are NOT directly linked to — grab a current npub from +# https://join.fips.network/ — so the ICMPv6 echo routes over the mixnet +# to your peer and then hop-by-hop across the mesh to the target: +docker compose exec fips ping6 -c3 .fips + +# A reply while the direct route is DROPped proves the traffic crossed the +# mixnet; the seconds-range RTT is the Sphinx path's signature, and a few +# extra hundred ms over reaching your own peer is the added mesh hops (a +# direct, non-mixnet connection would be ~30 ms). +``` + +The Nostr relay answers only over the FIPS mesh (fd00::/8) and on the +container's loopback — inbound eth0 traffic, including the host's port-80 +mapping, is dropped by the isolation rules. Check it from inside: + +```bash +docker compose exec fips curl -s -H "Accept: application/nostr+json" http://127.0.0.1/ +``` + +## Configuration (.env) + +| Variable | Default | Meaning | +| --- | --- | --- | +| `FIPS_NSEC` | *(required)* | Node identity, `fipsctl keygen` | +| `FIPS_PEER_NPUB` | test-us03's npub | Peer to dial; empty = standalone | +| `FIPS_PEER_ADDR` | `54.183.70.180:443` | **TCP** endpoint in nym/tcp mode (use `:2121` for udp) | +| `FIPS_PEER_TRANSPORT` | `nym` | `nym` \| `tcp` \| `udp` — see "Switching transport" above | +| `NYM_SERVICE_PROVIDER` | *(auto)* | Network requester; empty = pick the best-scored from [harbourmaster](https://harbourmaster.nymtech.net/) | +| `NYM_CLIENT_ID` | `fips-nym-client` | Nym client identity (kept in the `nym-data` volume) | + +With `FIPS_PEER_TRANSPORT=tcp` or `udp` the nym client is **not started at +all** and FIPS connects directly — useful as a baseline comparison. + +## Troubleshooting + +- **`could not auto-discover a Nym service provider`** — the harbourmaster + API was unreachable or returned no providers; pick one manually from + and set `NYM_SERVICE_PROVIDER`. +- **Slow or failing mixnet bootstrap** — service providers and gateways + vary in quality. Delete the client state and retry with another provider: + `docker compose down -v && NYM_SERVICE_PROVIDER= docker compose up`. + (The provider is baked into the client state at init; changing it + requires wiping the `nym-data` volume.) +- **Peer never becomes active** — confirm the peer's TCP endpoint is + reachable from the open internet (the network requester dials it from + the Nym exit side, not from your machine). +- **Never run this image under emulation** — the image builds native for + a reason: under Rosetta/qemu, the FIPS daemon's ChaCha20-Poly1305 + assembly (ring/BoringSSL) silently fails AEAD on larger frames; bloom + filter announces are dropped and multi-hop routing never converges, + while small control traffic keeps working — a maddeningly subtle + failure mode. Only the embedded amd64 `nym-socks5-client` (Nym ships no + other arch) runs emulated on Apple Silicon, which it tolerates. + +## Notes + +- The container's lifecycle follows the FIPS daemon; strfry, nginx and the + nym client run as background processes inside the same container and are + restarted with it (`restart: unless-stopped`). +- SSH (port 22, no auth) and tools like `tcpdump`, `nak`, `iperf3` are + inside the image for poking around — this is a demo image, do not expose + it beyond your machine. diff --git a/examples/sidecar-nostr-mixnet-relay/docker-compose.yml b/examples/sidecar-nostr-mixnet-relay/docker-compose.yml new file mode 100644 index 0000000..9219700 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/docker-compose.yml @@ -0,0 +1,55 @@ +networks: + fips-net: + name: ${FIPS_NETWORK:-fips-mixnet-net} + driver: bridge + ipam: + config: + - subnet: ${FIPS_SUBNET:-172.20.2.0/24} + +services: + # Single container running ALL services: fips daemon, nym-socks5-client, + # strfry Nostr relay, nginx, dnsmasq. See entrypoint.sh for start order. + fips: + # Builds NATIVE for the host — fips must not run under emulation + # (Rosetta breaks its AEAD on larger frames). Only the embedded + # amd64 nym-socks5-client is emulated on Apple Silicon. + build: + context: ../.. + dockerfile: examples/sidecar-nostr-mixnet-relay/Dockerfile + hostname: fips-mixnet + cap_add: + - NET_ADMIN + devices: + - /dev/net/tun:/dev/net/tun + sysctls: + - net.ipv6.conf.all.disable_ipv6=0 + restart: unless-stopped + ports: + - "2121:2121/udp" # FIPS UDP transport + - "8443:8443/tcp" # FIPS TCP transport + - "80:80/tcp" # Nostr relay WebSocket (via nginx) + environment: + - RUST_LOG=${RUST_LOG:-info} + - FIPS_NSEC=${FIPS_NSEC} + - FIPS_PEER_NPUB=${FIPS_PEER_NPUB:-} + - FIPS_PEER_ADDR=${FIPS_PEER_ADDR:-} + - FIPS_PEER_ALIAS=${FIPS_PEER_ALIAS:-peer} + - FIPS_PEER_TRANSPORT=${FIPS_PEER_TRANSPORT:-nym} + - FIPS_UDP_BIND=${FIPS_UDP_BIND:-0.0.0.0:2121} + - FIPS_TUN_MTU=${FIPS_TUN_MTU:-1280} + - FIPS_UDP_MTU=${FIPS_UDP_MTU:-1472} + - FIPS_NYM_SOCKS5_ADDR=${FIPS_NYM_SOCKS5_ADDR:-127.0.0.1:1080} + - NYM_CLIENT_ID=${NYM_CLIENT_ID:-fips-nym-client} + - NYM_SERVICE_PROVIDER=${NYM_SERVICE_PROVIDER:-} + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./relay/strfry.conf:/usr/src/app/strfry.conf:ro + - relay-data:/usr/src/app/strfry-db + - nym-data:/root/.nym + networks: + fips-net: + ipv4_address: ${FIPS_IPV4:-172.20.2.20} + +volumes: + relay-data: + nym-data: diff --git a/examples/sidecar-nostr-mixnet-relay/entrypoint.sh b/examples/sidecar-nostr-mixnet-relay/entrypoint.sh new file mode 100755 index 0000000..e6009b9 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/entrypoint.sh @@ -0,0 +1,221 @@ +#!/bin/bash +# Single-container entrypoint: generate the FIPS config, apply iptables +# isolation, start the Nostr relay (strfry + nginx), start the Nym SOCKS5 +# client (only when the FIPS config enables the nym transport), and launch +# FIPS last — so the mixnet proxy is provably up before FIPS dials its peer. +set -e + +# --- Generate FIPS config from environment variables --- + +FIPS_NSEC="${FIPS_NSEC:?FIPS_NSEC is required}" +FIPS_UDP_BIND="${FIPS_UDP_BIND:-0.0.0.0:2121}" +FIPS_TCP_BIND="${FIPS_TCP_BIND:-0.0.0.0:8443}" +FIPS_TUN_MTU="${FIPS_TUN_MTU:-1280}" +FIPS_UDP_MTU="${FIPS_UDP_MTU:-1472}" +FIPS_PEER_TRANSPORT="${FIPS_PEER_TRANSPORT:-nym}" +FIPS_NYM_SOCKS5_ADDR="${FIPS_NYM_SOCKS5_ADDR:-127.0.0.1:1080}" +NYM_CLIENT_ID="${NYM_CLIENT_ID:-fips-nym-client}" +NYM_STARTUP_TIMEOUT="${NYM_STARTUP_TIMEOUT:-180}" + +mkdir -p /etc/fips + +# Build peers section +PEERS_SECTION="" +if [ -n "$FIPS_PEER_NPUB" ] && [ -n "$FIPS_PEER_ADDR" ]; then + FIPS_PEER_ALIAS="${FIPS_PEER_ALIAS:-peer}" + PEERS_SECTION=" - npub: \"${FIPS_PEER_NPUB}\" + alias: \"${FIPS_PEER_ALIAS}\" + addresses: + - transport: ${FIPS_PEER_TRANSPORT} + addr: \"${FIPS_PEER_ADDR}\" + connect_policy: auto_connect" +fi + +# The nym transport block is emitted only in nym mode; the SOCKS5 client +# below starts only when this block is present in the config. +NYM_SECTION="" +if [ "$FIPS_PEER_TRANSPORT" = "nym" ]; then + NYM_SECTION=" nym: + socks5_addr: \"${FIPS_NYM_SOCKS5_ADDR}\" + startup_timeout_secs: 120" +fi + +cat > /etc/fips/fips.yaml </dev/null \ + | jq -r '[.items[] | select(.routing_score == 1.0)] + | sort_by(.last_updated_utc) | last + | .service_provider_client_id // empty' 2>/dev/null \ + || true) + if [ -z "$NYM_SERVICE_PROVIDER" ]; then + # Last resort: a provider known to work at the time of + # writing (2026-06). Providers are volatile community + # infra — if the mixnet connects but no traffic flows + # ('no node with identity … is known' warnings), this + # fallback has gone stale: pick a current one from + # https://harbourmaster.nymtech.net/ and set it in .env. + NYM_SERVICE_PROVIDER="${NYM_FALLBACK_PROVIDER:-7sfw3sEtSPwhWLmEasVmPXKxqioCo4GaXRkm9bW6yWGZ.CkhMoH85wfNcV2fwoBjc6QDbcaFZHzKqFFvXWfYMw19y@4ScsM6AVowhKTMWaH98NLntKDwbu2ZMEycUk4mZiZppG}" + echo "WARNING: harbourmaster auto-discovery failed — using the" >&2 + echo "baked-in fallback provider (may be stale; see .env):" >&2 + echo " ${NYM_SERVICE_PROVIDER}" >&2 + else + echo "Auto-selected service provider: ${NYM_SERVICE_PROVIDER}" + fi + fi + echo "Initializing Nym SOCKS5 client '${NYM_CLIENT_ID}' …" + nym-socks5-client init \ + --id "${NYM_CLIENT_ID}" \ + --provider "${NYM_SERVICE_PROVIDER}" \ + --port "${NYM_PORT}" \ + --host "${NYM_HOST}" + else + # The provider is baked into the client state at init time — a value + # set or discovered now does NOT apply to an existing client. Surface + # the one actually in effect so a stale/dead provider isn't chased + # silently (symptom: 'no node with identity … is known' warnings). + STORED_PROVIDER=$(grep -m1 -oE '[1-9A-HJ-NP-Za-km-z]{20,}\.[1-9A-HJ-NP-Za-km-z]{20,}@[1-9A-HJ-NP-Za-km-z]{20,}' \ + "${HOME}/.nym/socks5-clients/${NYM_CLIENT_ID}/config/config.toml" 2>/dev/null || true) + echo "Reusing existing Nym client state (provider: ${STORED_PROVIDER:-unknown})." + echo "To switch provider, remove the nym-data volume: docker compose down -v" + fi + + echo "Starting Nym SOCKS5 client (mixnet bootstrap may take a minute) …" + nym-socks5-client run \ + --id "${NYM_CLIENT_ID}" \ + --port "${NYM_PORT}" \ + --host "${NYM_HOST}" & + + # FIPS must not start dialing before the proxy accepts connections. + elapsed=0 + until nc -z "$NYM_HOST" "$NYM_PORT" 2>/dev/null; do + if [ "$elapsed" -ge "$NYM_STARTUP_TIMEOUT" ]; then + echo "ERROR: Nym SOCKS5 proxy not ready after ${NYM_STARTUP_TIMEOUT}s" >&2 + exit 1 + fi + sleep 2 + elapsed=$((elapsed + 2)) + done + echo "Nym SOCKS5 proxy ready at ${FIPS_NYM_SOCKS5_ADDR} (after ~${elapsed}s)" +fi + +# --- Launch FIPS (container lifecycle follows the daemon) --- + +echo "Starting FIPS daemon..." +exec fips --config /etc/fips/fips.yaml diff --git a/examples/sidecar-nostr-mixnet-relay/relay/strfry.conf b/examples/sidecar-nostr-mixnet-relay/relay/strfry.conf new file mode 100644 index 0000000..5f4c333 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/relay/strfry.conf @@ -0,0 +1,71 @@ +## +## strfry configuration for FIPS mesh deployment. +## Full reference: https://github.com/hoytech/strfry +## + +db = "/usr/src/app/strfry-db/" + +dbParams { + # Maximum size of the database (bytes). 10 GiB is a safe default. + mapsize = 10737418240 +} + +relay { + # Bind on all interfaces so the FIPS TUN (IPv4 + IPv6) can reach it. + bind = "0.0.0.0" + port = 7777 + + nofiles = 0 + + info { + name = "FIPS Nostr Relay" + description = "A Nostr relay accessible over the FIPS mesh network." + pubkey = "" + contact = "" + } + + # Maximum size of an inbound WebSocket message (bytes). + maxWebsocketPayloadSize = 131072 + + # Send a ping every N seconds to keep connections alive. + autoPingSeconds = 55 + + # Enable per-message compression. + enableTcpNoDelay = false + + rejectFutureEventsSeconds = 900 + rejectEphemeralEventsOlderThanSeconds = 60 + rejectEventsNewerThanSeconds = 900 + + maxFilterLimit = 500 + maxSubsPerConnection = 20 + + writePolicy { + # Plugin executable for write-policy decisions (leave empty to allow all). + plugin = "" + } + + compression { + enabled = true + slidingWindow = true + } + + logging { + dumpInAll = false + dumpInEvents = false + dumpInReqs = false + dbScanPerf = false + } + + numThreads { + ingester = 3 + reqWorker = 3 + reqMonitor = 3 + negentropy = 2 + } + + negentropy { + enabled = true + maxSyncEvents = 1000000 + } +} diff --git a/examples/sidecar-nostr-mixnet-relay/resolv.conf b/examples/sidecar-nostr-mixnet-relay/resolv.conf new file mode 100644 index 0000000..bbc8559 --- /dev/null +++ b/examples/sidecar-nostr-mixnet-relay/resolv.conf @@ -0,0 +1 @@ +nameserver 127.0.0.1 diff --git a/src/bin/fipstop/ui/transports.rs b/src/bin/fipstop/ui/transports.rs index 4c1ca97..bc42b95 100644 --- a/src/bin/fipstop/ui/transports.rs +++ b/src/bin/fipstop/ui/transports.rs @@ -481,6 +481,26 @@ fn draw_transport_detail(frame: &mut Frame, app: &App, area: Rect, t: &serde_jso &helpers::nested_u64(t, "stats", "connect_refused"), )); } + "nym" => { + lines.push(helpers::kv_line( + "MTU Exceeded", + &helpers::nested_u64(t, "stats", "mtu_exceeded"), + )); + lines.push(helpers::kv_line( + "SOCKS5 Errors", + &helpers::nested_u64(t, "stats", "socks5_errors"), + )); + lines.push(Line::from("")); + lines.push(helpers::section_header("Connections")); + lines.push(helpers::kv_line( + "Established", + &helpers::nested_u64(t, "stats", "connections_established"), + )); + lines.push(helpers::kv_line( + "Timeouts", + &helpers::nested_u64(t, "stats", "connect_timeouts"), + )); + } "ethernet" => { lines.push(Line::from("")); lines.push(helpers::section_header("Beacons")); diff --git a/src/config/mod.rs b/src/config/mod.rs index 4e2dde2..0097a15 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -39,8 +39,8 @@ pub use node::{ }; pub use peer::{ConnectPolicy, PeerAddress, PeerConfig}; pub use transport::{ - BleConfig, DirectoryServiceConfig, EthernetConfig, TcpConfig, TorConfig, TransportInstances, - TransportsConfig, UdpConfig, + BleConfig, DirectoryServiceConfig, EthernetConfig, NymConfig, TcpConfig, TorConfig, + TransportInstances, TransportsConfig, UdpConfig, }; /// Default config filename. diff --git a/src/config/transport.rs b/src/config/transport.rs index 94af738..1c73cc0 100644 --- a/src/config/transport.rs +++ b/src/config/transport.rs @@ -801,6 +801,77 @@ impl BleConfig { } } +// ============================================================================ +// Nym Transport Configuration +// ============================================================================ + +/// Default Nym SOCKS5 proxy address (nym-socks5-client). +const DEFAULT_NYM_SOCKS5_ADDR: &str = "127.0.0.1:1080"; + +/// Default Nym connect timeout in milliseconds (300s — Nym mixnet +/// SOCKS5 connections require multiple round-trips through 3 mix nodes +/// with timing obfuscation, which can take several minutes). +const DEFAULT_NYM_CONNECT_TIMEOUT_MS: u64 = 300_000; + +/// Default Nym MTU (same as TCP). +const DEFAULT_NYM_MTU: u16 = 1400; + +/// Default Nym startup timeout in seconds (time to wait for +/// nym-socks5-client to become ready before giving up). +const DEFAULT_NYM_STARTUP_TIMEOUT_SECS: u64 = 120; + +/// Nym transport instance configuration. +/// +/// Outbound-only connections through a nym-socks5-client SOCKS5 proxy. +/// The nym-socks5-client must be running separately (e.g., as a sidecar +/// process or container). +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct NymConfig { + /// SOCKS5 proxy address (host:port). Defaults to "127.0.0.1:1080". + #[serde(default, skip_serializing_if = "Option::is_none")] + pub socks5_addr: Option, + + /// Outbound connect timeout in milliseconds. Defaults to 300000 (300s). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub connect_timeout_ms: Option, + + /// Default MTU for Nym connections. Defaults to 1400. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mtu: Option, + + /// Seconds to wait for nym-socks5-client to become ready at startup. + /// Defaults to 120. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_timeout_secs: Option, +} + +impl NymConfig { + /// Get the SOCKS5 proxy address. Default: "127.0.0.1:1080". + pub fn socks5_addr(&self) -> &str { + self.socks5_addr + .as_deref() + .unwrap_or(DEFAULT_NYM_SOCKS5_ADDR) + } + + /// Get the connect timeout in milliseconds. Default: 300000. + pub fn connect_timeout_ms(&self) -> u64 { + self.connect_timeout_ms + .unwrap_or(DEFAULT_NYM_CONNECT_TIMEOUT_MS) + } + + /// Get the default MTU. Default: 1400. + pub fn mtu(&self) -> u16 { + self.mtu.unwrap_or(DEFAULT_NYM_MTU) + } + + /// Get the startup timeout in seconds. Default: 120. + pub fn startup_timeout_secs(&self) -> u64 { + self.startup_timeout_secs + .unwrap_or(DEFAULT_NYM_STARTUP_TIMEOUT_SECS) + } +} + // ============================================================================ // TransportsConfig // ============================================================================ @@ -827,6 +898,10 @@ pub struct TransportsConfig { #[serde(default, skip_serializing_if = "is_transport_empty")] pub tor: TransportInstances, + /// Nym transport instances. + #[serde(default, skip_serializing_if = "is_transport_empty")] + pub nym: TransportInstances, + /// BLE transport instances. #[serde(default, skip_serializing_if = "is_transport_empty")] pub ble: TransportInstances, @@ -844,6 +919,7 @@ impl TransportsConfig { && self.ethernet.is_empty() && self.tcp.is_empty() && self.tor.is_empty() + && self.nym.is_empty() && self.ble.is_empty() } @@ -863,6 +939,9 @@ impl TransportsConfig { if !other.tor.is_empty() { self.tor = other.tor; } + if !other.nym.is_empty() { + self.nym = other.nym; + } if !other.ble.is_empty() { self.ble = other.ble; } diff --git a/src/lib.rs b/src/lib.rs index 54252f3..5e5285b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,7 +30,7 @@ pub use identity::{ }; // Re-export config types -pub use config::{Config, ConfigError, IdentityConfig, TorConfig, UdpConfig}; +pub use config::{Config, ConfigError, IdentityConfig, NymConfig, TorConfig, UdpConfig}; pub use upper::config::{DnsConfig, TunConfig}; // Re-export discovery types diff --git a/src/node/mod.rs b/src/node/mod.rs index bb99f60..1120b75 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -50,6 +50,7 @@ use crate::node::session::SessionEntry; use crate::peer::{ActivePeer, PeerConnection}; #[cfg(unix)] use crate::transport::ethernet::EthernetTransport; +use crate::transport::nym::NymTransport; use crate::transport::tcp::TcpTransport; use crate::transport::tor::TorTransport; use crate::transport::udp::UdpTransport; @@ -982,6 +983,21 @@ impl Node { transports.push(TransportHandle::Tor(tor)); } + // Create Nym transport instances + let nym_instances: Vec<_> = self + .config() + .transports + .nym + .iter() + .map(|(name, config)| (name.map(|s| s.to_string()), config.clone())) + .collect(); + + for (name, nym_config) in nym_instances { + let transport_id = self.allocate_transport_id(); + let nym = NymTransport::new(transport_id, name, nym_config, packet_tx.clone()); + transports.push(TransportHandle::Nym(nym)); + } + // Create BLE transport instances #[cfg(bluer_available)] { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index e573a06..a29cd34 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -6,6 +6,7 @@ #[cfg(test)] pub mod loopback; +pub mod nym; pub mod tcp; pub mod tor; pub mod udp; @@ -22,6 +23,7 @@ use ble::DefaultBleTransport; use ethernet::EthernetTransport; #[cfg(test)] use loopback::LoopbackTransport; +use nym::NymTransport; use secp256k1::XOnlyPublicKey; use std::fmt; use std::net::SocketAddr; @@ -259,6 +261,13 @@ impl TransportType { reliable: true, // in-process channel delivery is lossless }; + /// Nym mixnet transport (via SOCKS5). + pub const NYM: TransportType = TransportType { + name: "nym", + connection_oriented: true, + reliable: true, + }; + /// Check if the transport is connectionless. pub fn is_connectionless(&self) -> bool { !self.connection_oriented @@ -891,6 +900,8 @@ pub enum TransportHandle { Tcp(TcpTransport), /// Tor transport (via SOCKS5). Tor(TorTransport), + /// Nym mixnet transport (via SOCKS5). + Nym(NymTransport), /// BLE L2CAP transport. #[cfg(target_os = "linux")] Ble(DefaultBleTransport), @@ -908,6 +919,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.start_async().await, TransportHandle::Tcp(t) => t.start_async().await, TransportHandle::Tor(t) => t.start_async().await, + TransportHandle::Nym(t) => t.start_async().await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.start_async().await, #[cfg(test)] @@ -923,6 +935,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.stop_async().await, TransportHandle::Tcp(t) => t.stop_async().await, TransportHandle::Tor(t) => t.stop_async().await, + TransportHandle::Nym(t) => t.stop_async().await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.stop_async().await, #[cfg(test)] @@ -938,6 +951,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.send_async(addr, data).await, TransportHandle::Tcp(t) => t.send_async(addr, data).await, TransportHandle::Tor(t) => t.send_async(addr, data).await, + TransportHandle::Nym(t) => t.send_async(addr, data).await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.send_async(addr, data).await, #[cfg(test)] @@ -953,6 +967,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.transport_id(), TransportHandle::Tcp(t) => t.transport_id(), TransportHandle::Tor(t) => t.transport_id(), + TransportHandle::Nym(t) => t.transport_id(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.transport_id(), #[cfg(test)] @@ -968,6 +983,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.name(), TransportHandle::Tcp(t) => t.name(), TransportHandle::Tor(t) => t.name(), + TransportHandle::Nym(t) => t.name(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.name(), #[cfg(test)] @@ -983,6 +999,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.transport_type(), TransportHandle::Tcp(t) => t.transport_type(), TransportHandle::Tor(t) => t.transport_type(), + TransportHandle::Nym(t) => t.transport_type(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.transport_type(), #[cfg(test)] @@ -998,6 +1015,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.state(), TransportHandle::Tcp(t) => t.state(), TransportHandle::Tor(t) => t.state(), + TransportHandle::Nym(t) => t.state(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.state(), #[cfg(test)] @@ -1013,6 +1031,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.mtu(), TransportHandle::Tcp(t) => t.mtu(), TransportHandle::Tor(t) => t.mtu(), + TransportHandle::Nym(t) => t.mtu(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.mtu(), #[cfg(test)] @@ -1031,6 +1050,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.link_mtu(addr), TransportHandle::Tcp(t) => t.link_mtu(addr), TransportHandle::Tor(t) => t.link_mtu(addr), + TransportHandle::Nym(t) => t.link_mtu(addr), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.link_mtu(addr), #[cfg(test)] @@ -1046,6 +1066,7 @@ impl TransportHandle { TransportHandle::Ethernet(_) => None, TransportHandle::Tcp(t) => t.local_addr(), TransportHandle::Tor(_) => None, + TransportHandle::Nym(_) => None, #[cfg(target_os = "linux")] TransportHandle::Ble(_) => None, #[cfg(test)] @@ -1061,6 +1082,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => Some(t.interface_name()), TransportHandle::Tcp(_) => None, TransportHandle::Tor(_) => None, + TransportHandle::Nym(_) => None, #[cfg(target_os = "linux")] TransportHandle::Ble(_) => None, #[cfg(test)] @@ -1100,6 +1122,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.discover(), TransportHandle::Tcp(t) => t.discover(), TransportHandle::Tor(t) => t.discover(), + TransportHandle::Nym(t) => t.discover(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.discover(), #[cfg(test)] @@ -1115,6 +1138,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.auto_connect(), TransportHandle::Tcp(t) => t.auto_connect(), TransportHandle::Tor(t) => t.auto_connect(), + TransportHandle::Nym(t) => t.auto_connect(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.auto_connect(), #[cfg(test)] @@ -1130,6 +1154,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.accept_connections(), TransportHandle::Tcp(t) => t.accept_connections(), TransportHandle::Tor(t) => t.accept_connections(), + TransportHandle::Nym(t) => t.accept_connections(), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.accept_connections(), #[cfg(test)] @@ -1139,7 +1164,7 @@ impl TransportHandle { /// Initiate a non-blocking connection to a remote address. /// - /// For connection-oriented transports (TCP, Tor), spawns a background + /// For connection-oriented transports (TCP, Tor, Nym), spawns a background /// task to establish the connection. For connectionless transports /// (UDP, Ethernet), this is a no-op that returns Ok immediately. /// @@ -1151,6 +1176,7 @@ impl TransportHandle { TransportHandle::Ethernet(_) => Ok(()), // connectionless TransportHandle::Tcp(t) => t.connect_async(addr).await, TransportHandle::Tor(t) => t.connect_async(addr).await, + TransportHandle::Nym(t) => t.connect_async(addr).await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.connect_async(addr).await, #[cfg(test)] @@ -1170,6 +1196,7 @@ impl TransportHandle { TransportHandle::Ethernet(_) => ConnectionState::Connected, TransportHandle::Tcp(t) => t.connection_state_sync(addr), TransportHandle::Tor(t) => t.connection_state_sync(addr), + TransportHandle::Nym(t) => t.connection_state_sync(addr), #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.connection_state_sync(addr), #[cfg(test)] @@ -1179,7 +1206,7 @@ impl TransportHandle { /// Close a specific connection on this transport. /// - /// No-op for connectionless transports. For TCP/Tor, removes the + /// No-op for connectionless transports. For TCP/Tor/Nym, removes the /// connection from the pool and drops the stream. pub async fn close_connection(&self, addr: &TransportAddr) { match self { @@ -1188,6 +1215,7 @@ impl TransportHandle { TransportHandle::Ethernet(t) => t.close_connection(addr), TransportHandle::Tcp(t) => t.close_connection_async(addr).await, TransportHandle::Tor(t) => t.close_connection_async(addr).await, + TransportHandle::Nym(t) => t.close_connection_async(addr).await, #[cfg(target_os = "linux")] TransportHandle::Ble(t) => t.close_connection_async(addr).await, #[cfg(test)] @@ -1212,6 +1240,7 @@ impl TransportHandle { TransportHandle::Ethernet(_) => TransportCongestion::default(), TransportHandle::Tcp(_) => TransportCongestion::default(), TransportHandle::Tor(_) => TransportCongestion::default(), + TransportHandle::Nym(_) => TransportCongestion::default(), #[cfg(target_os = "linux")] TransportHandle::Ble(_) => TransportCongestion::default(), #[cfg(test)] @@ -1249,6 +1278,9 @@ impl TransportHandle { TransportHandle::Tor(t) => { serde_json::to_value(t.stats().snapshot()).unwrap_or_default() } + TransportHandle::Nym(t) => { + serde_json::to_value(t.stats().snapshot()).unwrap_or_default() + } #[cfg(target_os = "linux")] TransportHandle::Ble(t) => { serde_json::to_value(t.stats().snapshot()).unwrap_or_default() diff --git a/src/transport/nym/mock_socks5.rs b/src/transport/nym/mock_socks5.rs new file mode 100644 index 0000000..2eff6fc --- /dev/null +++ b/src/transport/nym/mock_socks5.rs @@ -0,0 +1,203 @@ +//! Mock SOCKS5 server for testing the Nym transport's connect path. +//! +//! A copy of the Tor transport's mock, implementing just enough of the +//! SOCKS5 protocol (RFC 1928) to support the no-auth (and, defensively, +//! username/password) CONNECT flow, then proxying bytes bidirectionally to a +//! fixed target. +//! +//! Difference from the Tor mock: this one accepts connections in a loop and +//! handles each on its own task. `NymTransport::start_async` first probes the +//! proxy port for readiness (opening and immediately dropping a connection); +//! looping lets the mock shrug that probe off — its handler returns on the +//! short first read — and still serve the real data connection that follows. + +use std::net::SocketAddr; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; + +/// SOCKS5 protocol constants. +const SOCKS_VERSION: u8 = 0x05; +const AUTH_NONE: u8 = 0x00; +const AUTH_PASSWORD: u8 = 0x02; +const CMD_CONNECT: u8 = 0x01; +const ATYP_IPV4: u8 = 0x01; +const ATYP_DOMAIN: u8 = 0x03; +const REP_SUCCESS: u8 = 0x00; + +/// Username/password auth sub-negotiation version (RFC 1929). +const AUTH_SUBNEG_VERSION: u8 = 0x01; +const AUTH_SUBNEG_SUCCESS: u8 = 0x00; + +/// A minimal mock SOCKS5 proxy server for testing. +/// +/// Accepts connections in a loop, performs the SOCKS5 handshake (supporting +/// both no-auth and username/password auth), then connects to a fixed target +/// address and proxies bytes bidirectionally. +pub struct MockSocks5Server { + /// Address the mock proxy is listening on. + addr: SocketAddr, + /// The real target address to connect to (ignores SOCKS5 requested target). + target_addr: SocketAddr, + /// Listener handle. + listener: Option, +} + +impl MockSocks5Server { + /// Create a new mock SOCKS5 server that forwards to the given target. + /// + /// Binds to `127.0.0.1:0` (OS-assigned port). + pub async fn new(target_addr: SocketAddr) -> std::io::Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + Ok(Self { + addr, + target_addr, + listener: Some(listener), + }) + } + + /// Get the proxy's listen address (for `NymConfig.socks5_addr`). + pub fn addr(&self) -> SocketAddr { + self.addr + } + + /// Run the proxy, accepting connections in a loop and proxying each. + /// + /// Returns a JoinHandle for the accept loop. + pub fn spawn(mut self) -> JoinHandle<()> { + let listener = self.listener.take().expect("listener already consumed"); + let target_addr = self.target_addr; + + tokio::spawn(async move { + loop { + let (client, _) = match listener.accept().await { + Ok(c) => c, + Err(_) => break, + }; + // Handle each connection independently so the readiness probe + // (which opens and drops a connection) cannot block the real + // data connection behind it. + tokio::spawn(handle_conn(client, target_addr)); + } + }) + } +} + +/// Handle a single accepted connection: SOCKS5 handshake then byte proxy. +async fn handle_conn(mut client: tokio::net::TcpStream, target_addr: SocketAddr) { + // === Method negotiation === + // Client sends: [version, nmethods, methods...] + let mut ver_nmethods = [0u8; 2]; + if client.read_exact(&mut ver_nmethods).await.is_err() { + // Readiness probe (or any early close) — nothing to serve. + return; + } + assert_eq!(ver_nmethods[0], SOCKS_VERSION, "expected SOCKS5"); + let nmethods = ver_nmethods[1] as usize; + + let mut methods = vec![0u8; nmethods]; + client.read_exact(&mut methods).await.expect("read methods"); + + // Prefer username/password auth if offered, fall back to no-auth. + let selected = if methods.contains(&AUTH_PASSWORD) { + AUTH_PASSWORD + } else if methods.contains(&AUTH_NONE) { + AUTH_NONE + } else { + panic!("no supported auth method offered"); + }; + + // Reply: [version, selected_method] + client + .write_all(&[SOCKS_VERSION, selected]) + .await + .expect("write method reply"); + + // === Username/password sub-negotiation (RFC 1929) === + if selected == AUTH_PASSWORD { + // Client sends: [ver(1), ulen(1), uname(ulen), plen(1), passwd(plen)] + let mut subneg_header = [0u8; 2]; + client + .read_exact(&mut subneg_header) + .await + .expect("read subneg header"); + assert_eq!( + subneg_header[0], AUTH_SUBNEG_VERSION, + "expected auth subneg v1" + ); + + let ulen = subneg_header[1] as usize; + let mut uname = vec![0u8; ulen]; + client.read_exact(&mut uname).await.expect("read username"); + + let mut plen_buf = [0u8; 1]; + client.read_exact(&mut plen_buf).await.expect("read plen"); + let plen = plen_buf[0] as usize; + let mut passwd = vec![0u8; plen]; + client.read_exact(&mut passwd).await.expect("read password"); + + client + .write_all(&[AUTH_SUBNEG_VERSION, AUTH_SUBNEG_SUCCESS]) + .await + .expect("write subneg reply"); + } + + // === Connect request === + // Client sends: [version, cmd, rsv, atyp, addr..., port] + let mut header = [0u8; 4]; + client + .read_exact(&mut header) + .await + .expect("read connect header"); + assert_eq!(header[0], SOCKS_VERSION); + assert_eq!(header[1], CMD_CONNECT); + + // Read and skip the address (we connect to target_addr regardless). + match header[3] { + ATYP_IPV4 => { + let mut addr_port = [0u8; 6]; // 4 IP + 2 port + client + .read_exact(&mut addr_port) + .await + .expect("read IPv4 addr"); + } + ATYP_DOMAIN => { + let mut len_buf = [0u8; 1]; + client + .read_exact(&mut len_buf) + .await + .expect("read domain len"); + let domain_len = len_buf[0] as usize; + let mut domain_port = vec![0u8; domain_len + 2]; // domain + 2 port + client + .read_exact(&mut domain_port) + .await + .expect("read domain addr"); + } + other => panic!("unsupported ATYP: {}", other), + } + + // Connect to the real target. + let mut target = tokio::net::TcpStream::connect(target_addr) + .await + .expect("connect to target"); + + // Reply: success, bind addr = 0.0.0.0:0 + let reply = [ + SOCKS_VERSION, + REP_SUCCESS, + 0x00, // RSV + ATYP_IPV4, + 0, + 0, + 0, + 0, // bind addr + 0, + 0, // bind port + ]; + client.write_all(&reply).await.expect("write connect reply"); + + // Proxy bytes bidirectionally. + let _ = tokio::io::copy_bidirectional(&mut client, &mut target).await; +} diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs new file mode 100644 index 0000000..4996664 --- /dev/null +++ b/src/transport/nym/mod.rs @@ -0,0 +1,1141 @@ +//! Nym Mixnet Transport Implementation +//! +//! Provides Nym-based transport for FIPS peer communication using the +//! "Mixnet-As-Proxy" pattern. Traffic is routed through a local +//! nym-socks5-client SOCKS5 proxy into the Nym mixnet, providing +//! anonymity via Sphinx packet routing and timing obfuscation. +//! +//! ## Architecture +//! +//! Outbound-only: connects to remote TCP peers through the local +//! nym-socks5-client SOCKS5 proxy. Like the Tor transport, reuses FMP +//! stream framing from `tcp::stream` and follows the same connection +//! pool pattern. No inbound service is supported. + +pub mod stats; + +#[cfg(test)] +mod mock_socks5; + +use super::{ + ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, + TransportError, TransportId, TransportState, TransportType, +}; +use crate::config::NymConfig; +use crate::transport::tcp::stream::read_fmp_packet; +use stats::NymStats; + +use futures::FutureExt; +use socket2::TcpKeepalive; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpStream; +use tokio::net::tcp::OwnedWriteHalf; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use tokio::time::Instant; +use tokio_socks::tcp::Socks5Stream; +use tracing::{debug, info, trace, warn}; + +// ============================================================================ +// Connection Pool +// ============================================================================ + +/// State for a single Nym connection to a peer. +struct NymConnection { + /// Write half of the split stream. + writer: Arc>, + /// Receive task for this connection. + recv_task: JoinHandle<()>, + /// MTU for this connection. + #[allow(dead_code)] + mtu: u16, + /// When the connection was established. + #[allow(dead_code)] + established_at: Instant, +} + +/// Shared connection pool. +type ConnectionPool = Arc>>; + +/// A pending background connection attempt. +struct ConnectingEntry { + /// Background task performing SOCKS5 connect + socket configuration. + task: JoinHandle>, +} + +/// Map of addresses with background connection attempts in progress. +type ConnectingPool = Arc>>; + +// ============================================================================ +// Nym Transport +// ============================================================================ + +/// Nym mixnet transport for FIPS. +/// +/// Provides connection-oriented, reliable byte stream delivery through +/// the Nym mixnet via a local nym-socks5-client SOCKS5 proxy. +/// Outbound-only — no inbound service. +pub struct NymTransport { + /// Unique transport identifier. + transport_id: TransportId, + /// Optional instance name (for named instances in config). + name: Option, + /// Configuration. + config: NymConfig, + /// Current state. + state: TransportState, + /// Connection pool: addr -> per-connection state. + pool: ConnectionPool, + /// Pending connection attempts: addr -> background connect task. + connecting: ConnectingPool, + /// Channel for delivering received packets to Node. + packet_tx: PacketTx, + /// Transport statistics. + stats: Arc, +} + +impl NymTransport { + /// Create a new Nym transport. + pub fn new( + transport_id: TransportId, + name: Option, + config: NymConfig, + packet_tx: PacketTx, + ) -> Self { + Self { + transport_id, + name, + config, + state: TransportState::Configured, + pool: Arc::new(Mutex::new(HashMap::new())), + connecting: Arc::new(Mutex::new(HashMap::new())), + packet_tx, + stats: Arc::new(NymStats::new()), + } + } + + /// Get the instance name (if configured as a named instance). + pub fn name(&self) -> Option<&str> { + self.name.as_deref() + } + + /// Get the transport statistics. + pub fn stats(&self) -> &Arc { + &self.stats + } + + /// Start the transport asynchronously. + /// + /// Validates the SOCKS5 proxy address and transitions to Up. + /// The nym-socks5-client must already be running and listening + /// on the configured address. + pub async fn start_async(&mut self) -> Result<(), TransportError> { + if !self.state.can_start() { + return Err(TransportError::AlreadyStarted); + } + + self.state = TransportState::Starting; + + let socks5_addr = self.config.socks5_addr().to_string(); + validate_host_port(&socks5_addr, "socks5_addr")?; + + // Wait for nym-socks5-client to be ready by probing the SOCKS5 port + let ready = self.wait_for_socks5_ready(&socks5_addr).await; + if !ready { + warn!( + transport_id = %self.transport_id, + socks5_addr = %socks5_addr, + "Nym SOCKS5 client not reachable after waiting — starting anyway \ + (connections will fail until it becomes available)" + ); + } + + self.state = TransportState::Up; + + if let Some(ref name) = self.name { + info!( + name = %name, + socks5_addr = %socks5_addr, + mtu = self.config.mtu(), + "Nym mixnet transport started" + ); + } else { + info!( + socks5_addr = %socks5_addr, + mtu = self.config.mtu(), + "Nym mixnet transport started" + ); + } + + Ok(()) + } + + /// Wait for the nym-socks5-client SOCKS5 proxy to become reachable. + /// + /// Probes the TCP port with exponential backoff. Returns true if the + /// proxy is reachable within the timeout, false otherwise. + async fn wait_for_socks5_ready(&self, socks5_addr: &str) -> bool { + let max_wait = Duration::from_secs(self.config.startup_timeout_secs()); + let start = Instant::now(); + let mut delay = Duration::from_secs(1); + + info!( + transport_id = %self.transport_id, + socks5_addr = %socks5_addr, + timeout_secs = max_wait.as_secs(), + "Waiting for Nym SOCKS5 client to become ready..." + ); + + loop { + match TcpStream::connect(socks5_addr).await { + Ok(_) => { + info!( + transport_id = %self.transport_id, + socks5_addr = %socks5_addr, + elapsed_secs = start.elapsed().as_secs(), + "Nym SOCKS5 client is ready" + ); + return true; + } + Err(e) => { + if start.elapsed() >= max_wait { + warn!( + transport_id = %self.transport_id, + socks5_addr = %socks5_addr, + error = %e, + elapsed_secs = start.elapsed().as_secs(), + "Nym SOCKS5 client not ready after timeout" + ); + return false; + } + debug!( + transport_id = %self.transport_id, + socks5_addr = %socks5_addr, + error = %e, + retry_in_secs = delay.as_secs(), + "Nym SOCKS5 client not ready yet, retrying..." + ); + tokio::time::sleep(delay).await; + delay = (delay * 2).min(Duration::from_secs(10)); + } + } + } + } + + /// Stop the transport asynchronously. + pub async fn stop_async(&mut self) -> Result<(), TransportError> { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + + // Abort pending connection attempts + let mut connecting = self.connecting.lock().await; + for (addr, entry) in connecting.drain() { + entry.task.abort(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Nym connect aborted (transport stopping)" + ); + } + drop(connecting); + + // Close all connections + let mut pool = self.pool.lock().await; + for (addr, conn) in pool.drain() { + conn.recv_task.abort(); + let _ = conn.recv_task.await; + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Nym connection closed (transport stopping)" + ); + } + drop(pool); + + self.state = TransportState::Down; + + info!( + transport_id = %self.transport_id, + "Nym transport stopped" + ); + + Ok(()) + } + + /// Send a packet asynchronously. + /// + /// If no connection exists, performs connect-on-send through the + /// Nym SOCKS5 proxy. + pub async fn send_async( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + + // Pre-send MTU check + let mtu = self.config.mtu() as usize; + if data.len() > mtu { + self.stats.record_mtu_exceeded(); + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.config.mtu(), + }); + } + + // Get or create connection + let writer = { + let pool = self.pool.lock().await; + pool.get(addr).map(|c| c.writer.clone()) + }; + + let writer = match writer { + Some(w) => w, + None => { + // Connect-on-send + self.connect(addr).await? + } + }; + + // Write packet + let mut w = writer.lock().await; + match w.write_all(data).await { + Ok(()) => { + self.stats.record_send(data.len()); + trace!( + transport_id = %self.transport_id, + remote_addr = %addr, + bytes = data.len(), + "Nym packet sent" + ); + Ok(data.len()) + } + Err(e) => { + self.stats.record_send_error(); + drop(w); + // Remove failed connection from pool + let mut pool = self.pool.lock().await; + if let Some(conn) = pool.remove(addr) { + conn.recv_task.abort(); + } + Err(TransportError::SendFailed(format!("{}", e))) + } + } + } + + /// Establish a new connection through the Nym SOCKS5 proxy. + async fn connect( + &self, + addr: &TransportAddr, + ) -> Result>, TransportError> { + let target_addr = parse_target_addr(addr)?; + let proxy_addr = self.config.socks5_addr(); + let timeout_ms = self.config.connect_timeout_ms(); + + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + proxy = %proxy_addr, + timeout_secs = timeout_ms / 1000, + "Connecting via Nym mixnet SOCKS5 proxy" + ); + + let connect_start = Instant::now(); + let socks_result = tokio::time::timeout(Duration::from_millis(timeout_ms), async { + match target_addr { + TargetAddr::Ip(socket_addr) => Socks5Stream::connect(proxy_addr, socket_addr).await, + TargetAddr::Hostname(host, port) => { + Socks5Stream::connect(proxy_addr, (host.as_str(), port)).await + } + } + }) + .await; + + let stream = match socks_result { + Ok(Ok(socks_stream)) => socks_stream.into_inner(), + Ok(Err(e)) => { + self.stats.record_socks5_error(); + warn!( + transport_id = %self.transport_id, + remote_addr = %addr, + error = %e, + elapsed_secs = connect_start.elapsed().as_secs(), + "Nym SOCKS5 connection failed" + ); + return Err(TransportError::ConnectionRefused); + } + Err(_) => { + self.stats.record_connect_timeout(); + warn!( + transport_id = %self.transport_id, + remote_addr = %addr, + timeout_secs = timeout_ms / 1000, + "Nym SOCKS5 connection timed out" + ); + return Err(TransportError::Timeout); + } + }; + + // Configure socket options via socket2 + let std_stream = stream + .into_std() + .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; + configure_socket(&std_stream)?; + + // Convert back to tokio + let stream = TcpStream::from_std(std_stream) + .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; + + // Split and spawn receive task + let (read_half, write_half) = stream.into_split(); + let writer = Arc::new(Mutex::new(write_half)); + + let transport_id = self.transport_id; + let packet_tx = self.packet_tx.clone(); + let pool = self.pool.clone(); + let recv_stats = self.stats.clone(); + let remote_addr = addr.clone(); + let mtu = self.config.mtu(); + + let recv_task = tokio::spawn(async move { + nym_receive_loop( + read_half, + transport_id, + remote_addr.clone(), + packet_tx, + pool, + mtu, + recv_stats, + ) + .await; + }); + + let conn = NymConnection { + writer: writer.clone(), + recv_task, + mtu, + established_at: Instant::now(), + }; + + let mut pool = self.pool.lock().await; + pool.insert(addr.clone(), conn); + + self.stats.record_connection_established(); + + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + elapsed_secs = connect_start.elapsed().as_secs(), + "Nym mixnet connection established via SOCKS5" + ); + + Ok(writer) + } + + /// Initiate a non-blocking connection to a remote address. + pub async fn connect_async(&self, addr: &TransportAddr) -> Result<(), TransportError> { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + + // Already established? + { + let pool = self.pool.lock().await; + if pool.contains_key(addr) { + return Ok(()); + } + } + + // Already connecting? + { + let connecting = self.connecting.lock().await; + if connecting.contains_key(addr) { + return Ok(()); + } + } + + let target_addr = parse_target_addr(addr)?; + let proxy_addr = self.config.socks5_addr().to_string(); + let timeout_ms = self.config.connect_timeout_ms(); + let transport_id = self.transport_id; + let remote_addr = addr.clone(); + let config = self.config.clone(); + + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + timeout_ms, + "Initiating background Nym SOCKS5 connect" + ); + + let task = tokio::spawn(async move { + let connect_start = Instant::now(); + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + proxy = %proxy_addr, + timeout_secs = timeout_ms / 1000, + "Nym SOCKS5 CONNECT starting (this may take several minutes through mixnet)" + ); + + let socks_result = tokio::time::timeout(Duration::from_millis(timeout_ms), async { + match target_addr { + TargetAddr::Ip(socket_addr) => { + Socks5Stream::connect(proxy_addr.as_str(), socket_addr).await + } + TargetAddr::Hostname(host, port) => { + Socks5Stream::connect(proxy_addr.as_str(), (host.as_str(), port)).await + } + } + }) + .await; + + let stream = match socks_result { + Ok(Ok(socks_stream)) => { + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + elapsed_secs = connect_start.elapsed().as_secs(), + "Nym SOCKS5 CONNECT succeeded" + ); + socks_stream.into_inner() + } + Ok(Err(e)) => { + warn!( + transport_id = %transport_id, + remote_addr = %remote_addr, + error = %e, + elapsed_secs = connect_start.elapsed().as_secs(), + "Background Nym SOCKS5 connect failed" + ); + return Err(TransportError::ConnectionRefused); + } + Err(_) => { + warn!( + transport_id = %transport_id, + remote_addr = %remote_addr, + timeout_secs = timeout_ms / 1000, + elapsed_secs = connect_start.elapsed().as_secs(), + "Background Nym SOCKS5 connect timed out after {}s", + connect_start.elapsed().as_secs() + ); + return Err(TransportError::Timeout); + } + }; + + // Configure socket options via socket2 + let std_stream = stream + .into_std() + .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; + configure_socket(&std_stream)?; + + let mtu = config.mtu(); + + // Convert back to tokio + let stream = TcpStream::from_std(std_stream) + .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; + + Ok((stream, mtu)) + }); + + let mut connecting = self.connecting.lock().await; + connecting.insert(addr.clone(), ConnectingEntry { task }); + + Ok(()) + } + + /// Query the state of a connection to a remote address. + pub fn connection_state_sync(&self, addr: &TransportAddr) -> ConnectionState { + // Check established pool first + if let Ok(pool) = self.pool.try_lock() { + if pool.contains_key(addr) { + return ConnectionState::Connected; + } + } else { + return ConnectionState::Connecting; + } + + // Check connecting pool + let mut connecting = match self.connecting.try_lock() { + Ok(c) => c, + Err(_) => return ConnectionState::Connecting, + }; + + let entry = match connecting.get_mut(addr) { + Some(e) => e, + None => return ConnectionState::None, + }; + + if !entry.task.is_finished() { + return ConnectionState::Connecting; + } + + // Task is done — take the result + let addr_clone = addr.clone(); + let task = connecting.remove(&addr_clone).unwrap().task; + + match task.now_or_never() { + Some(Ok(Ok((stream, mtu)))) => { + self.promote_connection(addr, stream, mtu); + ConnectionState::Connected + } + Some(Ok(Err(e))) => ConnectionState::Failed(format!("{}", e)), + Some(Err(e)) => ConnectionState::Failed(format!("task failed: {}", e)), + None => ConnectionState::Connecting, + } + } + + /// Promote a completed background connection to the established pool. + fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) { + let (read_half, write_half) = stream.into_split(); + let writer = Arc::new(Mutex::new(write_half)); + + let transport_id = self.transport_id; + let packet_tx = self.packet_tx.clone(); + let pool = self.pool.clone(); + let recv_stats = self.stats.clone(); + let remote_addr = addr.clone(); + + let recv_task = tokio::spawn(async move { + nym_receive_loop( + read_half, + transport_id, + remote_addr.clone(), + packet_tx, + pool, + mtu, + recv_stats, + ) + .await; + }); + + let conn = NymConnection { + writer, + recv_task, + mtu, + established_at: Instant::now(), + }; + + if let Ok(mut pool) = self.pool.try_lock() { + pool.insert(addr.clone(), conn); + self.stats.record_connection_established(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Nym connection established (background connect)" + ); + } else { + conn.recv_task.abort(); + warn!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Failed to promote Nym connection (pool locked)" + ); + } + } + + /// Close a specific connection asynchronously. + pub async fn close_connection_async(&self, addr: &TransportAddr) { + let mut pool = self.pool.lock().await; + if let Some(conn) = pool.remove(addr) { + conn.recv_task.abort(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Nym connection closed" + ); + } + } +} + +impl Transport for NymTransport { + fn transport_id(&self) -> TransportId { + self.transport_id + } + + fn transport_type(&self) -> &TransportType { + &TransportType::NYM + } + + fn state(&self) -> TransportState { + self.state + } + + fn mtu(&self) -> u16 { + self.config.mtu() + } + + fn link_mtu(&self, _addr: &TransportAddr) -> u16 { + self.config.mtu() + } + + fn start(&mut self) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use start_async() for Nym transport".into(), + )) + } + + fn stop(&mut self) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use stop_async() for Nym transport".into(), + )) + } + + fn send(&self, _addr: &TransportAddr, _data: &[u8]) -> Result<(), TransportError> { + Err(TransportError::NotSupported( + "use send_async() for Nym transport".into(), + )) + } + + fn discover(&self) -> Result, TransportError> { + Ok(Vec::new()) + } + + fn accept_connections(&self) -> bool { + false + } +} + +// ============================================================================ +// Address Parsing +// ============================================================================ + +/// Target address for the SOCKS5 CONNECT request. +#[derive(Clone, Debug)] +enum TargetAddr { + /// Numeric IP:port. + Ip(SocketAddr), + /// Hostname:port (DNS resolved by the exit node). + Hostname(String, u16), +} + +/// Parse a TransportAddr string into a target address. +fn parse_target_addr(addr: &TransportAddr) -> Result { + let s = addr.as_str().ok_or_else(|| { + TransportError::InvalidAddress("Nym address must be a valid UTF-8 string".into()) + })?; + + if let Ok(socket_addr) = s.parse::() { + Ok(TargetAddr::Ip(socket_addr)) + } else { + let (host, port_str) = s.rsplit_once(':').ok_or_else(|| { + TransportError::InvalidAddress(format!("invalid address (expected host:port): {}", s)) + })?; + let port: u16 = port_str + .parse() + .map_err(|_| TransportError::InvalidAddress(format!("invalid port: {}", s)))?; + Ok(TargetAddr::Hostname(host.to_string(), port)) + } +} + +// ============================================================================ +// Receive Loop (per-connection) +// ============================================================================ + +/// Per-connection Nym receive loop. +async fn nym_receive_loop( + mut reader: tokio::net::tcp::OwnedReadHalf, + transport_id: TransportId, + remote_addr: TransportAddr, + packet_tx: PacketTx, + pool: ConnectionPool, + mtu: u16, + stats: Arc, +) { + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "Nym receive loop starting" + ); + + loop { + match read_fmp_packet(&mut reader, mtu).await { + Ok(data) => { + stats.record_recv(data.len()); + + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = data.len(), + "Nym packet received" + ); + + let packet = ReceivedPacket::new(transport_id, remote_addr.clone(), data); + + if packet_tx.send(packet).await.is_err() { + debug!( + transport_id = %transport_id, + "Packet channel closed, stopping Nym receive loop" + ); + break; + } + } + Err(e) => { + stats.record_recv_error(); + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + error = %e, + "Nym receive error, removing connection" + ); + break; + } + } + } + + // Clean up: remove ourselves from the pool + let mut pool_guard = pool.lock().await; + pool_guard.remove(&remote_addr); + + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "Nym receive loop stopped" + ); +} + +// ============================================================================ +// Socket Configuration +// ============================================================================ + +/// Configure socket options on a SOCKS5-connected stream. +fn configure_socket(stream: &std::net::TcpStream) -> Result<(), TransportError> { + let socket = socket2::SockRef::from(stream); + + // TCP_NODELAY — always enable for FIPS (latency-sensitive protocol messages) + socket + .set_tcp_nodelay(true) + .map_err(|e| TransportError::StartFailed(format!("set nodelay: {}", e)))?; + + // TCP keepalive (30s, matching TCP transport) + let keepalive = TcpKeepalive::new().with_time(Duration::from_secs(30)); + socket + .set_tcp_keepalive(&keepalive) + .map_err(|e| TransportError::StartFailed(format!("set keepalive: {}", e)))?; + + Ok(()) +} + +// ============================================================================ +// Address Validation +// ============================================================================ + +/// Validate that a string is a valid host:port address. +fn validate_host_port(addr: &str, field: &str) -> Result<(), TransportError> { + let parts: Vec<&str> = addr.rsplitn(2, ':').collect(); + if parts.len() != 2 { + return Err(TransportError::InvalidAddress(format!( + "{} must be host:port, got: {}", + field, addr + ))); + } + let _port: u16 = parts[0].parse().map_err(|_| { + TransportError::InvalidAddress(format!("{} has invalid port: {}", field, addr)) + })?; + Ok(()) +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::packet_channel; + + /// Test config: a syntactically valid loopback proxy address, with the + /// startup readiness probe disabled (no real nym-socks5-client runs in + /// unit tests) and a short connect timeout to bound any accidental dial. + fn make_config() -> NymConfig { + NymConfig { + socks5_addr: Some("127.0.0.1:1080".to_string()), + startup_timeout_secs: Some(0), + connect_timeout_ms: Some(2000), + ..Default::default() + } + } + + // ---- parse_target_addr ---- + + #[test] + fn test_parse_target_addr_ipv4() { + let addr = TransportAddr::from_string("192.0.2.10:2121"); + match parse_target_addr(&addr).unwrap() { + TargetAddr::Ip(socket_addr) => { + assert_eq!( + socket_addr, + "192.0.2.10:2121".parse::().unwrap() + ); + } + other => panic!("expected Ip variant, got {:?}", other), + } + } + + #[test] + fn test_parse_target_addr_ipv6_bracketed() { + // A bracketed IPv6 literal parses cleanly as a SocketAddr, so the + // connect path treats it as an Ip target with the brackets handled + // correctly (this is the path that actually dials peers). + let addr = TransportAddr::from_string("[2001:db8::1]:443"); + match parse_target_addr(&addr).unwrap() { + TargetAddr::Ip(socket_addr) => { + assert_eq!( + socket_addr, + "[2001:db8::1]:443".parse::().unwrap() + ); + } + other => panic!("expected Ip variant, got {:?}", other), + } + } + + #[test] + fn test_parse_target_addr_hostname() { + let addr = TransportAddr::from_string("peer.example.com:8443"); + match parse_target_addr(&addr).unwrap() { + TargetAddr::Hostname(host, port) => { + assert_eq!(host, "peer.example.com"); + assert_eq!(port, 8443); + } + other => panic!("expected Hostname variant, got {:?}", other), + } + } + + #[test] + fn test_parse_target_addr_missing_port() { + // No colon at all — cannot be split into host:port. + let addr = TransportAddr::from_string("peer.example.com"); + assert!(parse_target_addr(&addr).is_err()); + } + + #[test] + fn test_parse_target_addr_non_numeric_port() { + let addr = TransportAddr::from_string("peer.example.com:notaport"); + assert!(parse_target_addr(&addr).is_err()); + } + + // ---- validate_host_port ---- + + #[test] + fn test_validate_host_port_ok() { + assert!(validate_host_port("127.0.0.1:1080", "socks5_addr").is_ok()); + assert!(validate_host_port("proxy.local:9050", "socks5_addr").is_ok()); + } + + #[test] + fn test_validate_host_port_missing_port() { + // No colon -> not host:port. + assert!(validate_host_port("127.0.0.1", "socks5_addr").is_err()); + } + + #[test] + fn test_validate_host_port_non_numeric_port() { + assert!(validate_host_port("127.0.0.1:abc", "socks5_addr").is_err()); + } + + /// Documents a known limitation: `validate_host_port` splits on the last + /// colon, so a bracketed IPv6 literal validates with port `1080` and a + /// host of `[::1]` (stray brackets) rather than being rejected. It is + /// harmless in practice because the SOCKS5 proxy defaults to an IPv4 + /// loopback address, and the Tor transport has the same gap. Pin the + /// current behavior so any future change here is a deliberate one. + #[test] + fn test_validate_host_port_ipv6_bracket_is_accepted() { + assert!(validate_host_port("[::1]:1080", "socks5_addr").is_ok()); + } + + // ---- config defaults ---- + + #[test] + fn test_config_defaults() { + let config = NymConfig::default(); + assert_eq!(config.socks5_addr(), "127.0.0.1:1080"); + assert_eq!(config.connect_timeout_ms(), 300_000); + assert_eq!(config.mtu(), 1400); + assert_eq!(config.startup_timeout_secs(), 120); + } + + // ---- Transport trait surface ---- + + #[test] + fn test_transport_type() { + let (tx, _rx) = packet_channel(32); + let transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + let tt = transport.transport_type(); + assert_eq!(tt.name, "nym"); + assert!(tt.connection_oriented); + assert!(tt.reliable); + } + + #[test] + fn test_accept_connections_false() { + let (tx, _rx) = packet_channel(32); + let transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + assert!(!transport.accept_connections()); + } + + #[test] + fn test_discover_returns_empty() { + let (tx, _rx) = packet_channel(32); + let transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + assert!(transport.discover().unwrap().is_empty()); + } + + #[test] + fn test_sync_methods_return_not_supported() { + let (tx, _rx) = packet_channel(32); + let mut transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + assert!(transport.start().is_err()); + assert!(transport.stop().is_err()); + let addr = TransportAddr::from_string("127.0.0.1:2121"); + assert!(transport.send(&addr, &[0u8; 10]).is_err()); + } + + // ---- lifecycle ---- + + #[tokio::test] + async fn test_start_stop() { + let (tx, _rx) = packet_channel(32); + let mut transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + transport.start_async().await.unwrap(); + assert_eq!(transport.state(), TransportState::Up); + transport.stop_async().await.unwrap(); + assert_eq!(transport.state(), TransportState::Down); + } + + #[tokio::test] + async fn test_double_start_fails() { + let (tx, _rx) = packet_channel(32); + let mut transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + transport.start_async().await.unwrap(); + assert!(transport.start_async().await.is_err()); + } + + #[tokio::test] + async fn test_stop_not_started_fails() { + let (tx, _rx) = packet_channel(32); + let mut transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + assert!(transport.stop_async().await.is_err()); + } + + #[tokio::test] + async fn test_send_not_started() { + let (tx, _rx) = packet_channel(32); + let transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + let addr = TransportAddr::from_string("127.0.0.1:2121"); + assert!(transport.send_async(&addr, &[0u8; 10]).await.is_err()); + } + + #[tokio::test] + async fn test_invalid_socks5_addr_start_fails() { + let (tx, _rx) = packet_channel(32); + let config = NymConfig { + socks5_addr: Some("not-a-host-port".to_string()), + startup_timeout_secs: Some(0), + ..Default::default() + }; + let mut transport = NymTransport::new(TransportId::new(1), None, config, tx); + assert!(transport.start_async().await.is_err()); + } + + #[tokio::test] + async fn test_send_async_rejects_oversized_packet() { + let (tx, _rx) = packet_channel(32); + let mut transport = NymTransport::new(TransportId::new(1), None, make_config(), tx); + transport.start_async().await.unwrap(); + + let mtu = transport.mtu() as usize; + let addr = TransportAddr::from_string("127.0.0.1:2121"); + + // One byte over the MTU is rejected for size, before any dial. + let oversized = vec![0u8; mtu + 1]; + let result = transport.send_async(&addr, &oversized).await; + assert!(matches!(result, Err(TransportError::MtuExceeded { .. }))); + + // A packet at exactly the MTU is not rejected for size. (It still + // fails — no proxy is listening — but not with MtuExceeded.) + let at_mtu = vec![0u8; mtu]; + let result = transport.send_async(&addr, &at_mtu).await; + assert!(!matches!(result, Err(TransportError::MtuExceeded { .. }))); + + transport.stop_async().await.unwrap(); + } + + // ======================================================================== + // Integration test using MockSocks5Server (connect path), mirroring the + // Tor transport's `test_send_recv_via_socks5`. + // ======================================================================== + + use crate::config::TcpConfig; + use crate::transport::tcp::TcpTransport; + use mock_socks5::MockSocks5Server; + + /// msg1 wire size: 4 prefix + 4 sender_idx + 106 noise_msg1 = 114 bytes. + const MSG1_WIRE_SIZE: usize = 114; + /// msg1 payload_len: sender_idx(4) + noise_msg1(106) = 110. + const MSG1_PAYLOAD_LEN: u16 = (MSG1_WIRE_SIZE - 4) as u16; + + /// Build a msg1 FMP frame (114 bytes) that `read_fmp_packet` accepts. + fn build_msg1_frame() -> Vec { + let mut frame = vec![0xAA; MSG1_WIRE_SIZE]; + frame[0] = 0x01; // ver=0, phase=1 + frame[1] = 0x00; // flags + frame[2..4].copy_from_slice(&MSG1_PAYLOAD_LEN.to_le_bytes()); + frame + } + + /// End-to-end connect path: a real TCP transport is the destination, a + /// mock SOCKS5 proxy sits in front of it, and the Nym transport dials the + /// destination through the proxy. A valid FMP frame sent via the Nym + /// transport must arrive at the destination byte-for-byte. + #[tokio::test] + async fn test_send_recv_via_socks5() { + // Destination TCP transport with a real listener. + let (dest_tx, mut dest_rx) = packet_channel(32); + let dest_config = TcpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut dest = TcpTransport::new(TransportId::new(100), None, dest_config, dest_tx); + dest.start_async().await.unwrap(); + let dest_addr = dest.local_addr().unwrap(); + + // Mock SOCKS5 proxy forwarding to the destination. + let mock = MockSocks5Server::new(dest_addr).await.unwrap(); + let proxy_addr = mock.addr(); + let _proxy_handle = mock.spawn(); + + // Nym transport pointing at the mock proxy. + let (nym_tx, _nym_rx) = packet_channel(32); + let nym_config = NymConfig { + socks5_addr: Some(proxy_addr.to_string()), + startup_timeout_secs: Some(5), + connect_timeout_ms: Some(5000), + ..Default::default() + }; + let mut nym = NymTransport::new(TransportId::new(200), None, nym_config, nym_tx); + nym.start_async().await.unwrap(); + + // Send a valid FMP frame through the SOCKS5 (mixnet) path. + let frame = build_msg1_frame(); + let target = TransportAddr::from_string(&dest_addr.to_string()); + nym.send_async(&target, &frame).await.unwrap(); + + // It must arrive at the destination, byte-for-byte. + let received = tokio::time::timeout(Duration::from_secs(5), dest_rx.recv()) + .await + .expect("timeout waiting for packet") + .expect("channel closed"); + assert_eq!(received.data, frame); + + nym.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } +} diff --git a/src/transport/nym/stats.rs b/src/transport/nym/stats.rs new file mode 100644 index 0000000..ec6f287 --- /dev/null +++ b/src/transport/nym/stats.rs @@ -0,0 +1,119 @@ +//! Nym transport statistics. + +use portable_atomic::{AtomicU64, Ordering}; + +use serde::Serialize; + +/// Statistics for a Nym transport instance. +/// +/// Uses atomic counters for lock-free updates from per-connection +/// receive loops and the send path concurrently. +pub struct NymStats { + pub packets_sent: AtomicU64, + pub bytes_sent: AtomicU64, + pub packets_recv: AtomicU64, + pub bytes_recv: AtomicU64, + pub send_errors: AtomicU64, + pub recv_errors: AtomicU64, + pub mtu_exceeded: AtomicU64, + pub connections_established: AtomicU64, + pub connect_timeouts: AtomicU64, + pub socks5_errors: AtomicU64, +} + +impl NymStats { + /// Create a new stats instance with all counters at zero. + pub fn new() -> Self { + Self { + packets_sent: AtomicU64::new(0), + bytes_sent: AtomicU64::new(0), + packets_recv: AtomicU64::new(0), + bytes_recv: AtomicU64::new(0), + send_errors: AtomicU64::new(0), + recv_errors: AtomicU64::new(0), + mtu_exceeded: AtomicU64::new(0), + connections_established: AtomicU64::new(0), + connect_timeouts: AtomicU64::new(0), + socks5_errors: AtomicU64::new(0), + } + } + + /// Record a successful send. + pub fn record_send(&self, bytes: usize) { + self.packets_sent.fetch_add(1, Ordering::Relaxed); + self.bytes_sent.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Record a successful receive. + pub fn record_recv(&self, bytes: usize) { + self.packets_recv.fetch_add(1, Ordering::Relaxed); + self.bytes_recv.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Record a send error. + pub fn record_send_error(&self) { + self.send_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Record a receive error. + pub fn record_recv_error(&self) { + self.recv_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Record an MTU exceeded rejection. + pub fn record_mtu_exceeded(&self) { + self.mtu_exceeded.fetch_add(1, Ordering::Relaxed); + } + + /// Record a successful outbound connection. + pub fn record_connection_established(&self) { + self.connections_established.fetch_add(1, Ordering::Relaxed); + } + + /// Record a connect timeout. + pub fn record_connect_timeout(&self) { + self.connect_timeouts.fetch_add(1, Ordering::Relaxed); + } + + /// Record a SOCKS5 protocol error. + pub fn record_socks5_error(&self) { + self.socks5_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Take a snapshot of all counters. + pub fn snapshot(&self) -> NymStatsSnapshot { + NymStatsSnapshot { + packets_sent: self.packets_sent.load(Ordering::Relaxed), + bytes_sent: self.bytes_sent.load(Ordering::Relaxed), + packets_recv: self.packets_recv.load(Ordering::Relaxed), + bytes_recv: self.bytes_recv.load(Ordering::Relaxed), + send_errors: self.send_errors.load(Ordering::Relaxed), + recv_errors: self.recv_errors.load(Ordering::Relaxed), + mtu_exceeded: self.mtu_exceeded.load(Ordering::Relaxed), + connections_established: self.connections_established.load(Ordering::Relaxed), + connect_timeouts: self.connect_timeouts.load(Ordering::Relaxed), + socks5_errors: self.socks5_errors.load(Ordering::Relaxed), + } + } +} + +impl Default for NymStats { + fn default() -> Self { + Self::new() + } +} + +/// Point-in-time snapshot of Nym stats (non-atomic, copyable). +#[derive(Clone, Debug, Default, Serialize)] +pub struct NymStatsSnapshot { + pub packets_sent: u64, + pub bytes_sent: u64, + pub packets_recv: u64, + pub bytes_recv: u64, + pub send_errors: u64, + pub recv_errors: u64, + pub mtu_exceeded: u64, + pub connections_established: u64, + pub connect_timeouts: u64, + pub socks5_errors: u64, +}