25 Commits
Author SHA1 Message Date
Arjen 081855e8e2 fix(ble): reframe inbound L2CAP stream so packets survive non-SeqPacket backends
BLE delivery was unreliable on every backend except BlueZ. The receive
path assumed one `recv()` returned exactly one whole FIPS packet, which
only holds for BlueZ's SOCK_SEQPACKET (boundary-preserving). Android's
BluetoothSocket input stream and macOS CoreBluetooth are byte-stream
oriented: a read can return a fragment of a packet or several packets
coalesced. Under the old loop a fragment was shipped up as a runt packet
(rejected by FMP/Noise) and a coalesced tail was silently truncated and
dropped — so packets were lost and the transport thrashed.

Recover packet boundaries from the byte stream instead of trusting the
OS to preserve them. FIPS packets are self-delimiting via the 4-byte FMP
common prefix, so this reuses the exact length-prefixed framer TCP already
uses (`tcp::stream::read_fmp_packet`) — kept transport-agnostic for this
reason. A new `BleStreamRead` adapter turns the datagram-shaped `BleStream`
into the `AsyncRead` that framer expects, buffering leftover bytes across
reads. The recv future owns its scratch buffer and returns an owned Vec so
it is `'static` and storable across `poll_read` calls.

One reader is threaded through the pubkey exchange and the receive loop per
connection, so bytes a peer coalesces after the 33-byte pubkey stay buffered
rather than being lost at the handoff. The pubkey exchange now reads via
`read_exact` (reassembles a fragmented pubkey) instead of a brittle single
`recv()` with an exact-length check.

On BlueZ this is a transparent pass-through (one recv already equals one
packet); on stream backends it reassembles. Either way the layer above sees
one whole packet per read, identically on every platform.

Also bump the Android outbound SEND_QUEUE_CAP from 8 to 32. Swept against
the peer speedtest: 8 starved the radio's connection events (~half
throughput), 64 bufferbloated TCP, 32 was best (~200/500 kbps up/down). The
sweep was noisy and non-monotonic — run-to-run BLE variance (RF, 2M PHY /
connection-priority grants) rivals the knob's effect — so 32 is the
best-observed value pending re-validation with PHY/interval instrumentation,
not a proven optimum.
2026-07-14 13:53:57 +02:00
Arjen 3d81f4b1f5 style(ble): rustfmt android_io and psm
Import ordering and multi-line wrapping rustfmt would enforce; android_io
is cfg(target_os = "android") so it never compiled on the host CI matrix
and the drift went unnoticed until the mobile build was wired up.
2026-07-14 13:53:57 +02:00
Arjen 0450b11d5f docs: list Android as a supported platform
Android lands as an embedded library: the host app owns the TUN (e.g. an
Android VpnService) and FIPS does no system-TUN ops. Document where the
docs enumerate platforms as a set.

- README transport matrix: add an Android column (UDP/TCP/BLE supported;
  Ethernet has no raw sockets; Tor/Nym need an external proxy not run on
  Android). Reword the intro to distinguish standalone-daemon hosts from
  the embedded-library Android target.
- transport-layer status: BLE is now implemented on Linux/glibc and
  Android (Linux via BlueZ, Android via the embedder radio bridge).
- set-up-bluetooth-peer how-to: add Android to the BLE platform table.
2026-07-14 13:53:57 +02:00
Arjen 4afa27f056 perf(ble): shallow, backpressured outbound queue to fix bufferbloat
The per-stream outbound byte queue was 256 deep (~384 KB) on a link whose
bandwidth-delay product is ~1 packet. FSP/MMP filled it, RTT ballooned to several
seconds, and TCP above couldn't ramp. A shallow tail-drop queue is no better — it
sheds packets and collapses TCP throughput.

Give the outbound queue its own shallow cap (SEND_QUEUE_CAP=8) and make
BleStream::send backpressure — await a free slot rather than try_send-dropping —
so flow control propagates up through FSP/MMP to TCP. On-device this cut RTT from
~5s to ~1.2s with throughput preserved and 0% loss.
2026-07-14 13:53:57 +02:00
Arjen 0aa50d245c fix(ble): dial the last-learned PSM on a per-peer lookup miss
Per-peer PSM discovery keys the learned PSM by BLE address, but RPAs rotate
between the scan that learns a peer's PSM and the dial, so resolve() missed and
fell back to the legacy default 0x0085 — which no node listens on. Every outbound
L2CAP connect was rejected and the link ran inbound-only at a fraction of its rate.

A node advertises exactly one OS-assigned listener PSM, so on an exact-address
miss, dial the most-recently-learned PSM instead of the fixed default; a wrong
guess only costs a dial-retry.
2026-07-14 13:53:57 +02:00
Arjen abb0701048 feat(node): app-owned TUN seam — embedder owns the fd, FIPS uses channels
Node::enable_app_owned_tun() lets an embedder that owns the TUN fd (e.g. an
Android VpnService) exchange IPv6 packet bytes with FIPS over channels instead of
FIPS creating a system TUN device. It returns (app_outbound_tx, app_inbound_rx):
the embedder pushes packets read from its fd into the outbound sender (app ->
mesh) and pulls packets destined for its fd from the inbound receiver (mesh ->
app). Called after Node::new and before start(), mirroring control_read_handle().

start() now gates TunDevice::create on tun_tx being unset, so when the app-owned
channels are pre-installed it skips system-TUN creation and does no system-TUN
ops. The inbound IPv6-shim delivery already writes to tun_tx, and run_rx_loop
already drains tun_outbound_rx into handle_tun_outbound, so both directions reuse
the existing wiring.

Packets entering via app_outbound_tx bypass the system-TUN reader's
handle_tun_packet, so the embedder must push only fd::/8-destined packets (FIPS no
longer filters the destination) and clamp TCP MSS on outbound SYNs; the rustdoc
and the IPv6-adapter design doc spell this out.

Tests: app_owned_tun_seam_wires_channels covers the channel round-trip and the
Active state; start_skips_system_tun_when_app_owned runs start() and asserts no
named system device is created (tun_name stays unset).
2026-07-14 13:53:57 +02:00
Arjen 5c02d33ad8 feat(ble): record scan adverts (PSM + RSSI) for the developer UI
deliver_scan now carries RSSI as well, and records the latest advert per address
(PSM + RSSI) into a map exposed via advert_views() -> Vec<AdvertView>. Cleared
each scan cycle (addresses rotate with MAC privacy). This is the radio-level
view, distinct from the node's mesh-level peer table.
2026-07-14 13:53:57 +02:00
Arjen 5083a09223 fix(ble): safe AndroidBleBridge teardown + replaceable injection
- next_send: clone the per-channel receiver out before blocking (no longer holds
  the channels lock across recv_timeout), and treat a dropped sender
  (Disconnected) as closed so the Kotlin writer thread exits when the stream is
  gone.
- channel_open: also reports a closed (dropped-stream) channel, not just a
  removed one — so writers stop promptly.
- set_android_ble_bridge: replaceable (Mutex<Option>) so a stop→start cycle
  re-injects a fresh bridge for the rebuilt node.
2026-07-14 13:53:57 +02:00
Arjen 16ead248d0 feat(ble): public peer-view read API for embedders running run_rx_loop
Expose ControlReadHandle (was pub(crate)) + make Node::control_read_handle()
public, and add ControlReadHandle::peer_views() -> Vec<PeerView>
{ node_addr_hex, npub, connected }, read lock-free from the tick-published stats
snapshot (peer_meta). This lets an embedder run run_rx_loop on a background task
(which exclusively borrows &mut Node) and still poll live peer state from a
cloned handle — no &Node access needed. Used by the Myco app's developer UI.
2026-07-14 13:53:57 +02:00
Arjen 9e59659073 feat(ble): AndroidBleBridge::channel_open — let next_send tell closed from timeout 2026-07-14 13:53:57 +02:00
Arjen 1611282047 feat(ble): Android backend — BleIo over a Kotlin-radio byte-bridge
The Android BLE radio lives in Kotlin, so AndroidIo/Stream/Acceptor/Scanner
implement BleIo by delegating to AndroidBleBridge — the channel machinery shared
with the JNI layer in the embedder. The AndroidRadio trait is the object-safe
command surface Kotlin implements (listen/connect/advertise/scan/close).

- Inbound bytes/events are pushed non-blocking into tokio channels
  (deliver_recv/inbound/scan/connect_result); outbound bytes are pulled by a
  per-channel Kotlin writer thread via next_send. BleStream::send never calls
  JNI — the byte hot path is pure channel push.
- Per-peer PSM: connect() substitutes the learned PSM; advertising emits the
  OS-assigned listener PSM (16-bit LE service-data).
- Wired as DefaultBleTransport on Android, plus a node construction arm that
  reads the embedder-injected bridge (set_android_ble_bridge).

Pure Rust (no JNI here — that's in myco-core), so the channel logic unit-tests
on the host with a mock radio. Cross-compiles clean for arm64-android.
2026-07-14 13:53:57 +02:00
Arjen 786758e8c3 feat(ble): per-peer PSM discovery core + compile BLE on macOS/Android
Foundation for "BLE v2" universal per-peer PSM discovery, shared by every
BleIo backend.

- ble/psm.rs: a 16-bit little-endian service-data PSM codec + a short-lived
  BleAddr->PSM map (learn/lookup/resolve/clear), with unit tests. Every node
  advertises its OS-assigned listener PSM and every dialer reads a peer's
  advertised PSM before connect(), replacing the fixed DEFAULT_PSM (0x0085)
  assumption that only BlueZ could satisfy (Android/macOS get OS-assigned
  listener PSMs).
- Un-gate the BLE transport module from linux-only to a new `ble_available`
  cfg (linux/macos/android). The pool/discovery/psm logic and the generic
  BleTransport<I> are platform-agnostic; only the concrete BleIo backend is
  platform-specific (BluerIo on linux-glibc, else the MockBleIo fallback).
  The macOS BluestIo and Android AndroidIo backends, and the per-backend
  advertise/scan/connect wiring of the PSM map, land in follow-ups.
2026-07-14 13:53:57 +02:00
Arjen 0cf035a0e1 feat(mobile): gate desktop transports/TUN by target_os, not features
A plain `cargo build` now compiles correctly for every target with no
flags: the compiler selects platform code by `target_os`, enabling what
each platform supports and disabling what it doesn't. Android no longer
needs `--no-default-features`.

- ethernet: gated `any(target_os = "linux", target_os = "macos")` (raw
  AF_PACKET / BPF). Android is `target_os = "android"` — not "linux" — so
  the raw-socket transport self-excludes there, as it already did on Windows.
- system-tun: real platform ops gated per `target_os = "linux"`/`"macos"`;
  Windows keeps its own path. Android gets a no-op stub — the TUN is
  app-owned by the embedder (e.g. an Android VpnService).
- dns: ipi6_ifindex is i32 on Android, u32 on macOS — cast.
- node: gate the unix-only ESTABLISHED_HEADER_SIZE import so the Windows
  build is warning-clean.

No Cargo features are introduced; desktop builds are unchanged.
2026-07-14 13:53:57 +02:00
Johnathan Corgan 1dbfefc9d0 Merge branch 'maint' 2026-07-02 22:03:41 +00:00
Johnathan Corgan 1d277e67c7 fix(tree): invalidate coord cache on parent-lost via peer removal
handle_peer_removal_tree_cleanup reparents or self-roots the node when
its parent link drops, but omitted the coordinate-cache invalidation that
every other position-change path performs. Cached entries for downstream
destinations kept the node's now-stale coordinate prefix, and because
find_next_hop refreshes the TTL on every routing access, an actively
routed stale entry never self-expired — corrected only by a fresh insert.

Mirror the loop-detection branch: inside the parent-loss changed block,
invalidate both classes — invalidate_via_node (reparent) and
invalidate_other_roots (self-root). Add regression tests for a parent-link
removal that reparents (via-node entry dropped, same-root sibling
preserved) and one that self-roots (both via-node and stale old-root
entries dropped).
2026-07-02 22:03:36 +00:00
Johnathan Corgan 793f844448 Merge branch 'maint' 2026-06-29 16:47:43 +00:00
Johnathan Corgan 2491091868 docs(testing): document ci-local.sh run isolation, preemption, and cleanup
Add a "Running CI locally" section to testing/README.md covering the
ci-local.sh orchestrator: the per-run FIPS_CI_RUN_ID override and how it
scopes compose projects, image tags, and per-chaos-child subnets so
simultaneous runs on one host never collide; the cancellation exit codes
(130/143 vs 0/1) and how a preempting worker interprets them; and the
label-based cleanup via --reap / ci-cleanup.sh.
2026-06-29 16:47:37 +00:00
Johnathan Corgan 243bd7985a Merge branch 'maint' 2026-06-29 14:23:57 +00:00
Johnathan Corgan 965de26239 testing: make ci-local.sh preemption-safe with per-run docker isolation
A CI worker may preempt an in-flight ci-local.sh run (SIGTERM, then SIGKILL
after a grace period) to restart on a newer commit. For that kill to be safe,
the script must clean up after itself and never let a dying run collide with
its restart. It previously had no signal handling, shared the default compose
project name across runs, and tore down each suite only at the suite end.

- Derive a per-run id (honoring FIPS_CI_RUN_ID, else short-sha+random) and
  namespace every docker resource to it: a fipsci_<run>_<suite> compose project
  per suite and per parallel chaos child, and per-run image tags
  (fips-test:<run>, fips-test-app:<run>) retagged to :latest only after both
  builds succeed so :latest never points at a half-built image.
- Install a bounded, idempotent teardown trap on SIGTERM/SIGINT (+ EXIT): reap
  parallel chaos children, then force-remove this run's docker resources via
  the new ci-cleanup.sh, wrapped in timeout so a stuck down cannot wedge it.
- Exit 143 (SIGTERM) / 130 (SIGINT), distinct from 0 (pass) / 1 (failed), so a
  preempting worker tells a cancelled run from a real failure.
- Add ci-cleanup.sh (also ci-local.sh --reap): force-removes leftover CI
  resources by the com.corganlabs.fips-ci=1 label and the fipsci_ project
  prefix, robust to however a prior run died.
- Label every per-suite docker resource so the label sweep reaps it after a
  SIGKILL regardless of network name: direct docker run/network resources, the
  sidecar compose services, and every per-suite compose network (acl-allowlist,
  boringtun, firewall, nat, static, both tor suites, and the chaos generator
  template). Parametrize the static/sidecar compose image refs so the per-run
  tags are honored.
- Give each parallel chaos child a unique /24 from 10.30.x (a new --subnet
  override on the sim CLI, assigned per-child in ci-local.sh) so parallel
  children never collide on a shared docker subnet, and a chaos net can never
  span a fixed-subnet suite (sidecar/static in 172.20.x). 10.30.x sits outside
  docker's default-address-pool range, so an auto-assigned net cannot land on
  it either; node IPs derive from the subnet, so no scenario config changes.
2026-06-29 14:23:46 +00:00
Johnathan Corgan ab915d0479 testing: convert remaining fixed-sleep settles to progress-aware polling
Replace the fixed post-first-rekey and post-second-rekey settle sleeps in
the rekey integration test, and the fixed-deadline baseline wait in the
interop test, with the deterministic wait_until_connected progress-aware
polling helper already used for the initial baseline convergence.

Each converted site fails fast with structured diagnostics when the mesh
is stuck and extends its deadline only while pairwise reachability is
still climbing, removing the wall-clock settle windows that produced
intermittent connectivity failures after a rekey.
2026-06-29 00:40:06 +00:00
Johnathan Corgan 30c5808e09 Merge maint into master after the v0.4.0 rollover
maint carries only its 0.4.1-dev version bump; master keeps its own
0.5.0-dev development version. Recorded as a linkage merge so later
forward-merges of real fixes land cleanly.
2026-06-27 18:35:41 +00:00
Johnathan Corgan 8f30924fc7 Open 0.4.1-dev cycle on maint after v0.4.0 release
Fast-forward maint to the v0.4.0 release and open the 0.4.x patch
series: bump the version to 0.4.1-dev and update the status badge.
2026-06-27 18:29:34 +00:00
Johnathan Corgan 3c9a629ad4 Open 0.5.0-dev cycle on master after v0.4.0 release
Bump the version to 0.5.0-dev and update the status badge.
2026-06-27 18:28:52 +00:00
Johnathan Corgan d5ee526f0e Release v0.4.0: bump version and finalize CHANGELOG date
Bump the crate version from 0.4.0-dev to 0.4.0 and stamp the 0.4.0
CHANGELOG heading with the release date.
2026-06-27 15:57:58 +00:00
ArjenandJohnathan Corgan 22a5b3e5c6 packaging(openwrt): default .apk WAN port to DSA name 'wan'
The shared fips.yaml ships ethernet.wan.interface: "eth0", the default
WAN port on OpenWrt 24 and earlier. OpenWrt 25 (DSA) boards, which the
.apk package targets, name the WAN port "wan" instead. Rewrite the
staged copy at build time so the as-installed config binds the Ethernet
transport to the right port out of the box, without maintaining a second
copy of the config file. The .ipk package keeps "eth0".

Update the openwrt README to document eth0 (24) vs wan (25/DSA) and add
a CHANGELOG entry.
2026-06-26 03:31:17 +00:00
44 changed files with 2101 additions and 161 deletions
+4 -1
View File
@@ -13,7 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
## [0.4.0] - 2026-06-21
## [0.4.0] - 2026-06-27
### Added
@@ -327,6 +327,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
example is placed under `/usr/share/fips`, deliberately outside
`/usr/share/doc`, which minimal and container installs path-exclude
(so the install-time seed source is never dropped).
- openwrt: the `.apk` package now defaults `ethernet.wan` to the
OpenWrt 25 DSA port name `wan`; the `.ipk` package keeps `eth0` for
OpenWrt 24 and earlier.
#### CI & test-harness reliability
Generated
+1 -1
View File
@@ -1074,7 +1074,7 @@ checksum = "9844ddc3a6e533d62bba727eb6c28b5d360921d5175e9ff0f1e621a5c590a4d5"
[[package]]
name = "fips"
version = "0.4.0-dev"
version = "0.5.0-dev"
dependencies = [
"arc-swap",
"bech32",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "fips"
version = "0.4.0-dev"
version = "0.5.0-dev"
edition = "2024"
description = "A distributed, decentralized network routing protocol for mesh nodes connecting over arbitrary transports"
license = "MIT"
+20 -16
View File
@@ -3,7 +3,7 @@
![banner](docs/logos/fips_banner.png)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
[![Rust](https://img.shields.io/badge/rust-1.85%2B-orange.svg)](https://www.rust-lang.org/)
[![Status](https://img.shields.io/badge/status-v0.4.0-green.svg)](#status--roadmap)
[![Status](https://img.shields.io/badge/status-v0.5.0--dev-green.svg)](#status--roadmap)
A self-organizing encrypted mesh network built on Nostr identities,
capable of operating over arbitrary transports without central
@@ -112,17 +112,19 @@ tutorial progression starting at
cargo build --release
```
Requires Rust 1.94.1+ (edition 2024). Linux, macOS, and Windows are
supported; transport availability varies by platform.
Requires Rust 1.94.1+ (edition 2024). Linux, macOS, and Windows run as
standalone daemons; Android is supported as an embedded library (the host
app owns the TUN, e.g. a `VpnService`). Transport availability varies by
platform.
| Transport | Linux | macOS | Windows | OpenWrt |
|-----------|:-----:|:-----:|:-------:|:-------:|
| UDP | ✅ | ✅ | ✅ | ✅ |
| TCP | ✅ | ✅ | ✅ | ✅ |
| Ethernet | ✅ | ✅ | ❌ | ✅ |
| Tor | ✅ | ✅ | ✅ | ✅ |
| Nym | ✅ | ✅ | ✅ | ❌ |
| BLE | ✅ | ❌ | ❌ | ❌ |
| Transport | Linux | macOS | Windows | Android | OpenWrt |
|-----------|:-----:|:-----:|:-------:|:-------:|:-------:|
| UDP | ✅ | ✅ | ✅ | ✅ | ✅ |
| TCP | ✅ | ✅ | ✅ | ✅ | ✅ |
| Ethernet | ✅ | ✅ | ❌ | ❌ | ✅ |
| Tor | ✅ | ✅ | ✅ | ❌ | ✅ |
| Nym | ✅ | ✅ | ✅ | ❌ | ❌ |
| BLE | ✅ | ❌ | ❌ | ✅ | ❌ |
On Linux, a source build requires `libclang` — the LAN gateway's
nftables bindings are generated by `bindgen` at build time, which
@@ -210,12 +212,14 @@ testing/ Docker-based integration test harnesses + chaos simulation
## Status & roadmap
FIPS is at **v0.4.0**. The core protocol works end-to-end over
FIPS is at **v0.5.0-dev** on the `master` branch.
[v0.4.0](https://github.com/jmcorgan/fips/releases/tag/v0.4.0) has
shipped; this development line continues the testing-and-polishing
track toward v0.5.0. The core protocol works end-to-end over
UDP, TCP, Ethernet, Tor, Nym, and Bluetooth on a global, public test
mesh of thousands of nodes. v0.4.0 builds on the v0.3.0 testing-and-polishing
track, adding the Nym mixnet transport and mDNS LAN discovery
alongside the existing Nostr-mediated peer discovery, UDP NAT
traversal, peer ACL, and packaging hardening. New wire-format work
mesh of thousands of nodes. v0.4.0 added the Nym mixnet transport and
mDNS LAN discovery alongside the existing Nostr-mediated peer discovery,
UDP NAT traversal, peer ACL, and packaging hardening. New wire-format work
continues to be staged on the `next` branch for the subsequent
release line.
+9
View File
@@ -41,9 +41,18 @@ fn main() {
// satisfy libdbus-sys's pkg-config cross-compile requirement, and musl
// router targets don't run BlueZ by default anyway.
println!("cargo:rustc-check-cfg=cfg(bluer_available)");
// `ble_available` gates the platform-agnostic BLE transport module (pool,
// discovery, per-peer PSM, the generic `BleTransport`). The module compiles
// on every platform that has — or will have — a concrete `BleIo` backend;
// the backend is selected per-platform (BluerIo on linux-glibc, BluestIo on
// macOS, AndroidIo on Android, else the in-memory MockBleIo fallback).
println!("cargo:rustc-check-cfg=cfg(ble_available)");
let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default();
let target_env = std::env::var("CARGO_CFG_TARGET_ENV").unwrap_or_default();
if target_os == "linux" && target_env != "musl" {
println!("cargo:rustc-cfg=bluer_available");
}
if matches!(target_os.as_str(), "linux" | "macos" | "android") {
println!("cargo:rustc-cfg=ble_available");
}
}
+28
View File
@@ -302,6 +302,34 @@ alternative — running under a dedicated unprivileged service
account with the capability granted on the binary — see
[../how-to/run-as-unprivileged-user.md](../how-to/run-as-unprivileged-user.md).
### App-Owned TUN (embedded hosts)
On platforms where FIPS is embedded rather than run as a daemon — notably
Android, where the `VpnService` owns the TUN fd and the app has no
`CAP_NET_ADMIN` — FIPS does not create `fips0` itself. Instead the embedder owns
the fd and exchanges IPv6 packet bytes with FIPS over channels.
`Node::enable_app_owned_tun()` sets this up. It is called after `Node::new` and
before `start()` (and before the node is moved into a background task), mirroring
`control_read_handle()`, and returns two app-side channel ends:
- **app → mesh** — the embedder pushes IPv6 packets read from its fd into
`app_outbound_tx`. These are drained by `run_rx_loop` into `handle_tun_outbound`
and routed exactly as the Reader Thread's output would be.
- **mesh → app** — inbound mesh traffic on port 256 is reconstructed and written
to the node's `tun_tx` (the same sink the Writer Thread reads); the embedder
pulls from `app_inbound_rx` and writes to its fd.
With the channels installed, `start()` skips system-TUN creation (it gates on
`tun_tx` being unset), so FIPS does no `CAP_NET_ADMIN` operations.
Because packets enter via `app_outbound_tx` rather than the Reader Thread, they
**bypass `handle_tun_packet`** — the `fd00::/8` destination filter, the ICMPv6
Destination Unreachable for off-mesh dests (see [Reader Thread](#reader-thread)),
and the [TUN-Side TCP MSS Clamping](#tun-side-tcp-mss-clamping). The embedder is
therefore responsible for routing only `fd00::/8` to its TUN (so only mesh-bound
packets arrive) and for clamping TCP MSS on outbound SYNs.
## Implementation Status
| Feature | Status |
+1 -1
View File
@@ -899,7 +899,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to
| WiFi | **Implemented** (via Ethernet transport, infrastructure mode) | mac80211 translates 802.11↔802.3; broadcast beacons unreliable through APs |
| Tor | **Implemented** | Outbound SOCKS5, inbound via onion service, .onion and clearnet addressing |
| Nym | **Implemented** | Outbound-only SOCKS5 through nym-socks5-client, mixnet anonymity, IP/hostname addressing |
| BLE | **Implemented** (Linux/glibc only; experimental) | L2CAP CoC, ATT_MTU negotiation, per-link MTU; musl/macOS/Windows skip |
| BLE | **Implemented** (Linux/glibc and Android; experimental) | L2CAP CoC, ATT_MTU negotiation, per-link MTU; Linux via BlueZ, Android via the embedder radio bridge; macOS/Windows/musl skip |
| Radio | Future direction | Constrained MTU (51222 bytes) |
| Serial | Future direction | SLIP/COBS framing, point-to-point |
+1
View File
@@ -44,6 +44,7 @@ crate accordingly).
| -------- | -------------- |
| Linux (glibc) | Supported. |
| Linux (musl, OpenWrt) | Disabled at build time. |
| Android | Supported (native Android BLE, via the embedder's radio bridge). |
| macOS | Not supported. |
| Windows | Not supported. |
+7
View File
@@ -191,6 +191,13 @@ install -d "$STAGE_DIR/etc/fips"
install -m 0600 "$FILES_DIR/etc/fips/fips.yaml" "$STAGE_DIR/etc/fips/fips.yaml"
install -m 0755 "$FILES_DIR/etc/fips/firewall.sh" "$STAGE_DIR/etc/fips/firewall.sh"
# The shared fips.yaml ships ethernet.wan.interface: "eth0", the OpenWrt 24
# default. This .apk package targets OpenWrt 25+ (DSA), where the WAN port is
# named "wan", so ship "wan" as the default. Patching the staged copy keeps the
# as-installed config correct for the platform without maintaining a second copy
# of the file; operators can still edit /etc/fips/fips.yaml for non-standard boards.
sed -i 's|interface: "eth0"|interface: "wan"|' "$STAGE_DIR/etc/fips/fips.yaml"
install -d "$STAGE_DIR/etc/dnsmasq.d"
install -m 0644 "$FILES_DIR/etc/dnsmasq.d/fips.conf" "$STAGE_DIR/etc/dnsmasq.d/fips.conf"
+4 -1
View File
@@ -132,7 +132,10 @@ The default config enables:
For Ethernet transport, uncomment the `ethernet:` section and set the correct
physical interface names for your router. **Always use physical port names
(`eth0`, `eth1`), never bridge names (`br-lan`).** See
(`eth0`, `eth1`, or DSA port names like `wan`/`lan1`), never bridge names
(`br-lan`).** The shipped default WAN port is `eth0` (OpenWrt 24); on OpenWrt
25 (DSA) boards the WAN port is named `wan` — the `.apk` package ships that
default. Run `ip link show` to confirm the names on your board. See
[`deploy/native/README.md`](../../deploy/native/README.md) for details.
## Service management
+35 -1
View File
@@ -41,7 +41,7 @@ use super::snapshot::{EntitySnapshot, RoutingSnapshot, StatsSnapshot};
/// starting R1 as `show_*` queries cut over to off-loop rendering; until then
/// they are wired but unread.
#[derive(Clone)]
pub(crate) struct ControlReadHandle {
pub struct ControlReadHandle {
/// Effectively-immutable node context (config, identity, limits).
context: Arc<NodeContext>,
/// Metrics registry (counters / gauges) for `show_stats_*`.
@@ -108,6 +108,40 @@ impl ControlReadHandle {
}
}
/// A minimal, public view of one peer for embedders (e.g. an app UI), read
/// lock-free from the tick-published snapshot. See [`ControlReadHandle::peer_views`].
#[derive(Debug, Clone)]
pub struct PeerView {
/// The peer's `node_addr`, hex-encoded.
pub node_addr_hex: String,
/// Resolved npub (or the `node_addr` hex when not yet resolved to a peer).
pub npub: String,
/// Whether the peer is currently in the live authenticated-peer table.
pub connected: bool,
}
impl ControlReadHandle {
/// A lock-free snapshot of known peers (node_addr / npub / connected),
/// read from the tick-published stats snapshot.
///
/// Intended for embedders that run [`crate::Node::run_rx_loop`] on a
/// background task (so the node is exclusively borrowed there) and poll peer
/// state from a clone of this handle — the read touches only an `ArcSwap`
/// load, never the `Node`. See the Myco app for the reference embedding.
pub fn peer_views(&self) -> Vec<PeerView> {
self.stats
.load()
.peer_meta
.iter()
.map(|(addr, meta)| PeerView {
node_addr_hex: addr.to_string(),
npub: meta.npub.clone(),
connected: meta.is_active,
})
.collect()
}
}
/// Attempt to serve a request entirely from the read handle, off the rx_loop.
///
/// Returns `Some(response)` when the command is a pure-snapshot query that has
+4 -2
View File
@@ -1201,8 +1201,10 @@ impl Node {
// This allows handshake messages to be sent before we start accepting packets
self.initiate_peer_connections().await;
// Initialize TUN interface last, after transports and peers are ready
if self.config().tun.enabled {
// Initialize TUN interface last, after transports and peers are ready.
// Skip when the TUN is app-owned (the embedder pre-set `tun_tx` via
// `enable_app_owned_tun`) — then FIPS does no system-TUN ops.
if self.config().tun.enabled && self.tun_tx.is_none() {
let address = *self.identity().address();
match TunDevice::create(&self.config().tun, address).await {
Ok(device) => {
+83 -9
View File
@@ -41,14 +41,18 @@ use self::routing_error_rate_limit::RoutingErrorRateLimiter;
/// `node.rekey.after_secs` remains the nominal interval (mean preserved).
pub(crate) const REKEY_JITTER_SECS: i64 = 15;
use self::wire::{
ESTABLISHED_HEADER_SIZE, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, build_encrypted,
build_established_header, prepend_inner_header,
FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, build_encrypted, build_established_header,
prepend_inner_header,
};
// Only referenced by the unix UDP fast-path block below; on Windows the wire
// buffer is sized through build_encrypted, leaving this import otherwise unused.
#[cfg(unix)]
use self::wire::ESTABLISHED_HEADER_SIZE;
use crate::bloom::{BloomFilter, BloomState};
use crate::cache::CoordCache;
use crate::node::session::SessionEntry;
use crate::peer::{ActivePeer, PeerConnection};
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
use crate::transport::ethernet::EthernetTransport;
use crate::transport::nym::NymTransport;
use crate::transport::tcp::TcpTransport;
@@ -61,7 +65,7 @@ use crate::transport::{
use crate::tree::TreeState;
use crate::upper::hosts::HostMap;
use crate::upper::icmp_rate_limit::IcmpRateLimiter;
use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx};
use crate::upper::tun::{TunError, TunOutboundRx, TunOutboundTx, TunState, TunTx};
use crate::utils::index::IndexAllocator;
use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity};
use rand::Rng;
@@ -928,7 +932,7 @@ impl Node {
}
// Create Ethernet transport instances (Unix only — requires raw sockets)
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
{
let eth_instances: Vec<_> = self
.config()
@@ -1039,6 +1043,43 @@ impl Node {
}
}
// Android BLE: the radio lives in Kotlin; build AndroidIo over the bridge
// injected by the embedder (see ble::android_io::set_android_ble_bridge).
#[cfg(all(target_os = "android", not(test)))]
{
let ble_instances: Vec<_> = self
.config()
.transports
.ble
.iter()
.map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
.collect();
match crate::transport::ble::android_io::android_ble_bridge() {
Some(bridge) => {
for (name, ble_config) in ble_instances {
let transport_id = self.allocate_transport_id();
let io = crate::transport::ble::android_io::AndroidIo::new(
std::sync::Arc::clone(&bridge),
);
let mut ble = crate::transport::ble::BleTransport::new(
transport_id,
name,
ble_config,
io,
packet_tx.clone(),
);
ble.set_local_pubkey(self.identity().pubkey().serialize());
transports.push(TransportHandle::Ble(ble));
}
}
None => {
if !ble_instances.is_empty() {
tracing::warn!("BLE configured but no Android radio bridge injected");
}
}
}
}
transports
}
@@ -1062,7 +1103,7 @@ impl Node {
&self,
addr_str: &str,
) -> Result<(TransportId, TransportAddr), NodeError> {
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
{
let (iface, mac_str) = addr_str.split_once('/').ok_or_else(|| {
NodeError::NoTransportForType(format!(
@@ -1094,7 +1135,7 @@ impl Node {
Ok((transport_id, TransportAddr::from_bytes(&mac)))
}
#[cfg(not(unix))]
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
{
Err(NodeError::NoTransportForType(
"Ethernet transport is not supported on this platform".to_string(),
@@ -1435,8 +1476,10 @@ impl Node {
/// over this node's already-shared `NodeContext` and `MetricsRegistry`.
///
/// Used at control-socket spawn time so pure-snapshot `show_*` queries
/// render off the rx_loop. Cloneable; cheap (all `Arc` clones).
pub(crate) fn control_read_handle(&self) -> crate::control::read_handle::ControlReadHandle {
/// render off the rx_loop. Cloneable; cheap (all `Arc` clones). Also the
/// public seam for embedders that run [`Self::run_rx_loop`] on a background
/// task and poll peer state via [`crate::control::read_handle::ControlReadHandle::peer_views`].
pub fn control_read_handle(&self) -> crate::control::read_handle::ControlReadHandle {
crate::control::read_handle::ControlReadHandle::new(
self.context.clone(),
self.metrics.clone(),
@@ -2813,6 +2856,37 @@ impl Node {
self.tun_tx.as_ref()
}
/// Set up an **app-owned TUN**: rather than FIPS creating a system TUN
/// device, the embedder (e.g. an Android `VpnService`) owns the fd and
/// exchanges IPv6 packet bytes with FIPS over the returned channels. Call
/// this after [`Node::new`] and **before** [`Self::start`] — and before
/// moving the node into a background task — exactly like
/// [`Self::control_read_handle`].
///
/// Returns `(app_outbound_tx, app_inbound_rx)`:
/// - push IPv6 packets read from the app's TUN fd into `app_outbound_tx`
/// (app → mesh); FIPS routes them to the destination node.
/// - pull IPv6 packets destined for the app's TUN fd from `app_inbound_rx`
/// (mesh → app) and write them to the fd (`recv_timeout` for clean stop).
///
/// With this set, [`Self::start`] skips system-TUN creation (it gates on
/// `tun_tx` being unset). Packets pushed into `app_outbound_tx` bypass the
/// system-TUN reader's `handle_tun_packet`, so the embedder must do what that
/// path otherwise would: push only `fd::/8`-destined IPv6 packets — FIPS no
/// longer filters the destination or emits ICMPv6 unreachable for off-mesh
/// dests — and clamp TCP MSS on outbound SYNs.
pub fn enable_app_owned_tun(&mut self) -> (TunOutboundTx, std::sync::mpsc::Receiver<Vec<u8>>) {
let tun_channel_size = self.config().node.buffers.tun_channel;
// app → mesh: the app pushes; `run_rx_loop` drains `tun_outbound_rx`.
let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(tun_channel_size);
// mesh → app: the node writes inbound packets to `tun_tx`; the app pulls.
let (tun_tx, tun_rx) = std::sync::mpsc::channel();
self.tun_tx = Some(tun_tx);
self.tun_outbound_rx = Some(outbound_rx);
self.tun_state = TunState::Active;
(outbound_tx, tun_rx)
}
// === Sending ===
/// Encrypt and send a link-layer message to an authenticated peer.
+122
View File
@@ -1333,3 +1333,125 @@ fn test_route_class_partition_sums_to_forwarded() {
assert_eq!(snap.route_crosslink_ascend, 3);
assert_eq!(snap.route_direct_peer, 1);
}
// === Coord-cache invalidation on parent loss ===
//
// Parent-lost-via-peer-removal is a genuine position change and must
// surgically invalidate the coordinate cache like every other such path
// (reparent → invalidate_via_node; self-root → invalidate_other_roots).
// `make_node_addr(0)` is the network minimum, so the node's random identity
// addr is always greater than it — the reparent/child geometry is deterministic.
#[test]
fn test_parent_loss_reparent_invalidates_coord_cache() {
let mut node = make_node();
let my_addr = *node.node_addr();
let root = make_node_addr(0);
let parent = make_node_addr(1);
let alt = make_node_addr(2);
// Current parent and an alternative, both rooted at `root`.
node.tree_state_mut().update_peer(
ParentDeclaration::new(parent, root, 1, 1000),
TreeCoordinate::from_addrs(vec![parent, root]).unwrap(),
);
node.tree_state_mut().update_peer(
ParentDeclaration::new(alt, root, 1, 1000),
TreeCoordinate::from_addrs(vec![alt, root]).unwrap(),
);
// Adopt `parent`; our coords become [my_addr, parent, root], root = `root`.
node.tree_state_mut().set_parent(parent, 1, 1000);
node.tree_state_mut().recompute_coords();
assert!(!node.tree_state().is_root());
assert_eq!(node.tree_state().root(), &root);
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
// via-node class: a downstream destination that routes through us.
let downstream = make_node_addr(10);
node.coord_cache_mut().insert(
downstream,
TreeCoordinate::from_addrs(vec![downstream, my_addr, root]).unwrap(),
now_ms,
);
// survivor: same root, does not route through us.
let sibling_dest = make_node_addr(11);
node.coord_cache_mut().insert(
sibling_dest,
TreeCoordinate::from_addrs(vec![sibling_dest, alt, root]).unwrap(),
now_ms,
);
// Parent link drops; node reparents onto `alt` (still rooted at `root`).
let changed = node.handle_peer_removal_tree_cleanup(&parent);
assert!(changed);
assert_eq!(node.tree_state().my_declaration().parent_id(), &alt);
assert_eq!(node.tree_state().root(), &root);
assert!(
!node.coord_cache().contains(&downstream, now_ms),
"entry routing through us must be invalidated after reparent"
);
assert!(
node.coord_cache().contains(&sibling_dest, now_ms),
"same-root entry not routing through us must survive (surgical, not a flush)"
);
}
#[test]
fn test_parent_loss_selfroot_invalidates_coord_cache() {
let mut node = make_node();
let my_addr = *node.node_addr();
let old_root = make_node_addr(0);
let parent = make_node_addr(1);
// Adopt `parent` (rooted at `old_root`); no alternative peers exist, so a
// parent loss self-roots the node.
node.tree_state_mut().update_peer(
ParentDeclaration::new(parent, old_root, 1, 1000),
TreeCoordinate::from_addrs(vec![parent, old_root]).unwrap(),
);
node.tree_state_mut().set_parent(parent, 1, 1000);
node.tree_state_mut().recompute_coords();
assert!(!node.tree_state().is_root());
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
// via-node class: routes through us.
let downstream = make_node_addr(10);
node.coord_cache_mut().insert(
downstream,
TreeCoordinate::from_addrs(vec![downstream, my_addr, old_root]).unwrap(),
now_ms,
);
// other-roots class: on the old root, does not route through us.
let foreign = make_node_addr(11);
node.coord_cache_mut().insert(
foreign,
TreeCoordinate::from_addrs(vec![foreign, parent, old_root]).unwrap(),
now_ms,
);
// Parent link drops; no alternative parent → node self-roots.
let changed = node.handle_peer_removal_tree_cleanup(&parent);
assert!(changed);
assert!(node.tree_state().is_root());
assert_eq!(node.tree_state().root(), &my_addr);
assert!(
!node.coord_cache().contains(&downstream, now_ms),
"via-node entry must be invalidated after self-root"
);
assert!(
!node.coord_cache().contains(&foreign, now_ms),
"stale old-root entry must be invalidated after self-root"
);
}
+55
View File
@@ -1995,3 +1995,58 @@ async fn handle_msg1_admits_existing_peer_at_cap() {
"rate limiter must rebalance after the (bypass-admitted) handler returns"
);
}
/// App-owned TUN seam: `enable_app_owned_tun` wires the embedder's packet
/// channels (an Android `VpnService` owns the fd) and marks the TUN active so
/// `start()` skips system-TUN creation.
#[test]
fn app_owned_tun_seam_wires_channels() {
let mut config = crate::Config::new();
config.tun.enabled = true;
let mut node = make_node_with(config);
let (outbound_tx, tun_rx) = node.enable_app_owned_tun();
// TUN is active and the inbound (mesh→app) sender is installed, so `start()`
// will skip `TunDevice::create` (it gates on `tun_tx.is_none()`).
assert_eq!(node.tun_state(), crate::upper::tun::TunState::Active);
assert!(node.tun_tx().is_some(), "inbound sender installed");
// mesh → app: a packet the node delivers to its `tun_tx` reaches the app's rx.
let pkt = vec![0x60u8, 0, 0, 0, 0, 0];
node.tun_tx().unwrap().send(pkt.clone()).unwrap();
assert_eq!(
tun_rx
.recv_timeout(std::time::Duration::from_millis(200))
.unwrap(),
pkt,
"the app pulls the same bytes the node wrote",
);
// app → mesh: the returned sender is live (its matching rx is held by the node
// and drained by `run_rx_loop` → `handle_tun_outbound`).
assert!(outbound_tx.try_send(vec![0x60]).is_ok());
}
/// With an app-owned TUN configured, `start()` must NOT create a system TUN
/// device: it leaves `tun_name` unset (a real device records its interface name)
/// and keeps the TUN `Active` with the app-owned channels.
#[tokio::test]
async fn start_skips_system_tun_when_app_owned() {
let mut config = crate::Config::new();
config.tun.enabled = true;
let mut node = make_node_with(config);
let (_outbound_tx, _tun_rx) = node.enable_app_owned_tun();
node.start().await.unwrap();
// No system device was created (that path records the interface name); the
// app-owned TUN stayed active.
assert!(
node.tun_name().is_none(),
"app-owned TUN must not create a named system device",
);
assert_eq!(node.tun_state(), crate::upper::tun::TunState::Active);
node.stop().await.unwrap();
}
+10
View File
@@ -620,6 +620,16 @@ impl Node {
.tree
.record_reject(TreeReject::OutboundSignFailed);
}
// handle_parent_lost may promote to root OR find new parent;
// cover both invalidation classes (same as the loop-detection
// branch above). Without this, cached downstream entries keep
// our now-stale coordinate prefix until TTL — and get_and_touch
// refreshes the TTL on every routing access, so an actively
// routed stale entry never self-expires.
self.coord_cache
.invalidate_via_node(our_identity.node_addr());
self.coord_cache
.invalidate_other_roots(self.tree_state.root());
info!(
new_root = %self.tree_state.root(),
is_root = self.tree_state.is_root(),
+734
View File
@@ -0,0 +1,734 @@
//! Android BLE backend: a [`BleIo`] whose radio lives in Kotlin.
//!
//! Android's BLE APIs are Java-only, so Kotlin owns the radio (scan, advertise,
//! L2CAP listen/connect, socket read/write) and exchanges **raw bytes** with this
//! Rust backend over a byte-bridge — symmetric to how nostr-vpn's `MobileTunnel`
//! exchanges TUN packet bytes across the FFI. FIPS keeps everything above the
//! `BleIo` trait (the pool, the cross-probe tiebreaker, the pubkey exchange,
//! Noise); this backend only moves bytes and surfaces adverts.
//!
//! ## Layering
//!
//! FIPS cannot depend on the app crate (`myco-core`), so the split is:
//!
//! - [`AndroidRadio`] — an object-safe trait for the few **commands** the radio
//! must run (listen/connect/advertise/scan/close). `myco-core` implements it
//! via JNI calls into the Kotlin radio object.
//! - [`AndroidBleBridge`] — the channel machinery shared by this backend and the
//! JNI layer. `myco-core` constructs it, injects it via
//! [`set_android_ble_bridge`], and drives its `deliver_*` / `next_send`
//! methods from its `Java_..._NativeCore_*` exports.
//! - [`AndroidIo`] / [`AndroidStream`] / [`AndroidAcceptor`] / [`AndroidScanner`]
//! — the `BleIo` impl, delegating to the bridge.
//!
//! ## Direction of blocking (matches nostr-vpn's MobileTunnel)
//!
//! - **Inbound** bytes/events (Kotlin → Rust) are **pushed** non-blocking into
//! tokio channels (`deliver_recv`, `deliver_inbound`, `deliver_scan`,
//! `deliver_connect_result`); the awaiting FIPS task wakes.
//! - **Outbound** bytes (Rust → Kotlin) are **pulled, blocking with timeout**, by
//! a per-channel Kotlin writer thread via [`AndroidBleBridge::next_send`].
//! `BleStream::send` only pushes into a std channel — it never calls JNI — so
//! the byte hot path never blocks a tokio worker on a JNI upcall.
//!
//! This module is platform-agnostic Rust (no JNI here — that lives in
//! `myco-core`), so it compiles and unit-tests on the host with a mock radio.
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU16, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Mutex as AsyncMutex;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use crate::transport::TransportError;
use super::DEFAULT_PSM;
use super::addr::BleAddr;
use super::io::{BleAcceptor, BleIo, BleScanner, BleStream};
use super::psm::PsmMap;
/// Synthetic adapter label (Android does not expose a BlueZ-style adapter name;
/// identity is the pubkey, never the MAC — see ble-interop.md).
const ANDROID_ADAPTER: &str = "ble0";
/// Bound on a per-channel inbound queue and the accept/scan fan-in. Generous so
/// control events (accept/scan) are not dropped under burst; L2CAP data drops are
/// tolerable since FMP/Noise above retransmits.
const CHANNEL_CAP: usize = 256;
/// Bound on the **outbound byte** queue (Rust → Kotlin writer), fixed at channel
/// creation. A bounded queue makes `BleStream::send` backpressure (the `SyncSender`
/// blocks), propagating flow control up through FSP/MMP to the TUN and TCP rather
/// than letting an unbounded queue bufferbloat the link.
///
/// 32 is empirical, swept against the peer speedtest: 8 starved the radio's
/// connection events (~half throughput), 64 bufferbloated TCP (regressed), and 32
/// was best (~200/500 kbps up/down). The sweep was noisy and non-monotonic across
/// single runs, though — run-to-run BLE variance (RF, and whether the OS grants 2M
/// PHY / high connection priority that session) rivals the effect of this knob — so
/// 32 is the best-observed working value, to be re-validated with repeated runs +
/// PHY/interval instrumentation, not a proven optimum.
const SEND_QUEUE_CAP: usize = 32;
/// Transport default MTU, used when the OS reports an unknown (0) channel MTU.
/// Matches `DEFAULT_BLE_MTU` in `config/transport.rs`.
const DEFAULT_BLE_MTU: u16 = 2048;
// ============================================================================
// AndroidRadio — the Kotlin-implemented command surface
// ============================================================================
/// The radio commands the bridge issues to the platform. `myco-core` implements
/// this via JNI `call_method` on the Kotlin `BleRadio` object. Object-safe so the
/// bridge can hold `Arc<dyn AndroidRadio>`.
///
/// These are the **control** plane only — never the byte hot path. Outbound bytes
/// are pulled by Kotlin via [`AndroidBleBridge::next_send`]; inbound bytes are
/// pushed by Kotlin via [`AndroidBleBridge::deliver_recv`].
pub trait AndroidRadio: Send + Sync {
/// Open an insecure L2CAP listener and return the OS-assigned PSM (0 = failure).
fn listen(&self) -> u16;
/// Begin dialing `addr` at `psm`. The outcome is delivered asynchronously via
/// [`AndroidBleBridge::deliver_connect_result`] keyed by `connect_id`.
fn connect(&self, connect_id: i64, addr: &BleAddr, psm: u16);
/// Advertise the FIPS service UUID plus our listener `psm` (16-bit LE
/// service-data — see [`super::psm`]).
fn start_advertising(&self, psm: u16);
fn stop_advertising(&self);
/// Scan for the FIPS UUID; deliver hits via [`AndroidBleBridge::deliver_scan`].
fn start_scanning(&self);
fn stop_scanning(&self);
/// Close the L2CAP socket for `ch_id` (called when FIPS drops the stream).
fn close_channel(&self, ch_id: i64);
}
/// One discovered scan advert (address / learned PSM / RSSI), for the developer
/// UI's "discovered devices" list.
#[derive(Debug, Clone)]
pub struct AdvertView {
/// `BleAddr` string (`adapter/AA:BB:..`); the MAC rotates with privacy.
pub addr: String,
/// The peer's advertised listener PSM (0 if not present in the advert).
pub psm: u16,
/// Signal strength in dBm (negative; closer ≈ less negative).
pub rssi: i32,
}
// ============================================================================
// AndroidBleBridge — the shared channel machinery
// ============================================================================
/// The half of a channel kept by the bridge (the JNI-facing ends).
struct ChannelState {
/// Kotlin-pushed inbound bytes land here; the stream's `recv` awaits them.
recv_tx: mpsc::Sender<Vec<u8>>,
/// `BleStream::send` pushes here; the Kotlin writer thread pulls via `next_send`.
/// `Arc` so `next_send` can clone it out and release the channels lock before
/// blocking on `recv_timeout`.
send_rx: Arc<Mutex<std::sync::mpsc::Receiver<Vec<u8>>>>,
closed: Arc<AtomicBool>,
}
/// The half of a channel handed to the `BleStream` (the FIPS-facing ends).
struct StreamEndpoints {
ch_id: i64,
remote: BleAddr,
send_mtu: u16,
recv_mtu: u16,
recv_rx: mpsc::Receiver<Vec<u8>>,
send_tx: std::sync::mpsc::SyncSender<Vec<u8>>,
closed: Arc<AtomicBool>,
}
/// Channel machinery shared between [`AndroidIo`] and the JNI layer in `myco-core`.
///
/// Constructed by `myco-core` with a concrete [`AndroidRadio`], injected via
/// [`set_android_ble_bridge`], and driven by its `deliver_*` / `next_send`
/// methods from the JNI exports.
pub struct AndroidBleBridge {
radio: Arc<dyn AndroidRadio>,
next_id: AtomicI64,
/// Our own OS-assigned listener PSM, learned from `radio.listen()`.
local_psm: AtomicU16,
/// Learned peer PSMs (advert service-data), consulted on `connect`.
psm_map: PsmMap,
/// Latest scan advert per address (PSM + RSSI), for the developer UI's
/// "discovered" list. Re-learned each scan cycle (addresses rotate).
adverts: Mutex<HashMap<BleAddr, (u16, i32)>>,
channels: Mutex<HashMap<i64, ChannelState>>,
/// connect_id → result slot for an in-flight outbound dial.
connects: Mutex<HashMap<i64, oneshot::Sender<StreamEndpoints>>>,
/// Inbound-accept fan-in; the acceptor takes the receiver once.
accept_tx: mpsc::Sender<StreamEndpoints>,
accept_rx: Mutex<Option<mpsc::Receiver<StreamEndpoints>>>,
/// Scan fan-in; the scanner takes the receiver once.
scan_tx: mpsc::Sender<BleAddr>,
scan_rx: Mutex<Option<mpsc::Receiver<BleAddr>>>,
}
impl AndroidBleBridge {
/// Build a bridge over a concrete radio.
pub fn new(radio: Arc<dyn AndroidRadio>) -> Arc<Self> {
let (accept_tx, accept_rx) = mpsc::channel(CHANNEL_CAP);
let (scan_tx, scan_rx) = mpsc::channel(CHANNEL_CAP);
Arc::new(Self {
radio,
next_id: AtomicI64::new(1),
local_psm: AtomicU16::new(DEFAULT_PSM),
psm_map: PsmMap::new(),
adverts: Mutex::new(HashMap::new()),
channels: Mutex::new(HashMap::new()),
connects: Mutex::new(HashMap::new()),
accept_tx,
accept_rx: Mutex::new(Some(accept_rx)),
scan_tx,
scan_rx: Mutex::new(Some(scan_rx)),
})
}
fn lock_channels(&self) -> std::sync::MutexGuard<'_, HashMap<i64, ChannelState>> {
self.channels.lock().unwrap_or_else(|e| e.into_inner())
}
/// Allocate a channel id and wire its two halves, registering the
/// bridge-facing half and returning the FIPS-facing half.
fn make_channel(&self, remote: BleAddr, send_mtu: u16, recv_mtu: u16) -> StreamEndpoints {
let ch_id = self.next_id.fetch_add(1, Ordering::Relaxed);
let (recv_tx, recv_rx) = mpsc::channel(CHANNEL_CAP);
let (send_tx, send_rx) = std::sync::mpsc::sync_channel(SEND_QUEUE_CAP);
let closed = Arc::new(AtomicBool::new(false));
self.lock_channels().insert(
ch_id,
ChannelState {
recv_tx,
send_rx: Arc::new(Mutex::new(send_rx)),
closed: Arc::clone(&closed),
},
);
StreamEndpoints {
ch_id,
remote,
send_mtu: if send_mtu == 0 {
DEFAULT_BLE_MTU
} else {
send_mtu
},
recv_mtu: if recv_mtu == 0 {
DEFAULT_BLE_MTU
} else {
recv_mtu
},
recv_rx,
send_tx,
closed,
}
}
// --- JNI-facing push/pull surface (called by myco-core's exports) ---
/// Kotlin accepted a new inbound L2CAP channel. Returns the allocated `ch_id`.
pub fn deliver_inbound(&self, remote: BleAddr, send_mtu: u16, recv_mtu: u16) -> i64 {
let ep = self.make_channel(remote, send_mtu, recv_mtu);
let ch_id = ep.ch_id;
if self.accept_tx.try_send(ep).is_err() {
// Acceptor gone or saturated: reclaim the half-registered channel.
self.lock_channels().remove(&ch_id);
return 0;
}
ch_id
}
/// Kotlin finished (or failed) an outbound dial started by `radio.connect`.
/// Returns the allocated `ch_id` on success, else 0.
pub fn deliver_connect_result(
&self,
connect_id: i64,
ok: bool,
remote: BleAddr,
send_mtu: u16,
recv_mtu: u16,
) -> i64 {
let waiter = self
.connects
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&connect_id);
let Some(tx) = waiter else { return 0 };
if !ok {
drop(tx); // dropping the sender wakes the awaiting connect() as an error
return 0;
}
let ep = self.make_channel(remote, send_mtu, recv_mtu);
let ch_id = ep.ch_id;
if tx.send(ep).is_err() {
self.lock_channels().remove(&ch_id);
return 0;
}
ch_id
}
/// Kotlin discovered a FIPS peer advertising `psm` (its OS-assigned listener
/// PSM) at signal strength `rssi` (dBm). Learns the per-peer PSM, records the
/// advert for the developer UI, and surfaces the address to the scanner.
pub fn deliver_scan(&self, addr: BleAddr, psm: u16, rssi: i32) {
if psm != 0 {
self.psm_map.learn(&addr, psm);
}
self.adverts
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(addr.clone(), (psm, rssi));
let _ = self.scan_tx.try_send(addr);
}
/// Snapshot of the current scan adverts (address / PSM / RSSI) for the
/// developer UI.
pub fn advert_views(&self) -> Vec<AdvertView> {
self.adverts
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.map(|(addr, (psm, rssi))| AdvertView {
addr: addr.to_string_repr(),
psm: *psm,
rssi: *rssi,
})
.collect()
}
/// Drop recorded adverts (called at the start of a scan cycle).
pub fn clear_adverts(&self) {
self.adverts
.lock()
.unwrap_or_else(|e| e.into_inner())
.clear();
}
/// Kotlin read one L2CAP packet for `ch_id`. Returns false if the channel is
/// unknown/closed (Kotlin should then stop its reader).
pub fn deliver_recv(&self, ch_id: i64, data: &[u8]) -> bool {
let tx = self.lock_channels().get(&ch_id).map(|c| c.recv_tx.clone());
match tx {
Some(tx) => tx.try_send(data.to_vec()).is_ok(),
None => false,
}
}
/// Kotlin's per-channel writer thread pulls the next outbound packet, blocking
/// up to `timeout`. `None` = timed out (Kotlin loops) or the channel is gone.
pub fn next_send(&self, ch_id: i64, timeout: Duration) -> Option<Vec<u8>> {
// Clone the per-channel receiver Arc + closed flag, then release the
// channels lock before blocking on recv_timeout (so close/create on other
// channels aren't stalled for up to `timeout`).
let (send_rx, closed) = {
let guard = self.lock_channels();
let state = guard.get(&ch_id)?;
(Arc::clone(&state.send_rx), Arc::clone(&state.closed))
};
let rx = send_rx.lock().unwrap_or_else(|e| e.into_inner());
match rx.recv_timeout(timeout) {
Ok(bytes) => Some(bytes),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => None,
// The send half (the BleStream) was dropped — the stream is gone
// (e.g. the node stopped). Mark closed so the next channel_open()
// returns false and the Kotlin writer thread exits.
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
closed.store(true, Ordering::Relaxed);
None
}
}
}
/// Kotlin reports `ch_id` closed (EOF / socket gone). Wakes the stream's
/// `recv` with a zero-length read (FIPS treats that as peer-closed).
pub fn channel_closed(&self, ch_id: i64) {
if let Some(state) = self.lock_channels().remove(&ch_id) {
state.closed.store(true, Ordering::Relaxed);
// Dropping recv_tx closes the stream's recv_rx → recv() returns Ok(0).
drop(state);
}
}
/// Whether `ch_id` is still open (registered and not marked closed). The JNI
/// `next_send` export uses this to tell a timeout (loop again) from a closed
/// channel (stop the writer thread).
pub fn channel_open(&self, ch_id: i64) -> bool {
self.lock_channels()
.get(&ch_id)
.map(|s| !s.closed.load(Ordering::Relaxed))
.unwrap_or(false)
}
}
// ============================================================================
// Global injection seam
// ============================================================================
static BRIDGE: Mutex<Option<Arc<AndroidBleBridge>>> = Mutex::new(None);
/// Inject the process-wide bridge before `Node::new` / start (one radio per
/// process). Replaceable so a stop-then-start cycle (BLE toggled off then on)
/// can inject a fresh bridge for the rebuilt node.
pub fn set_android_ble_bridge(bridge: Arc<AndroidBleBridge>) {
*BRIDGE.lock().unwrap_or_else(|e| e.into_inner()) = Some(bridge);
}
/// The injected bridge, if any. The node's BLE construction arm reads this.
pub fn android_ble_bridge() -> Option<Arc<AndroidBleBridge>> {
BRIDGE.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
// ============================================================================
// BleIo implementation
// ============================================================================
/// macOS/Android-style external-radio backend over [`AndroidBleBridge`].
pub struct AndroidIo {
bridge: Arc<AndroidBleBridge>,
}
impl AndroidIo {
pub fn new(bridge: Arc<AndroidBleBridge>) -> Self {
Self { bridge }
}
}
/// One live L2CAP channel.
pub struct AndroidStream {
ch_id: i64,
remote: BleAddr,
send_mtu: u16,
recv_mtu: u16,
recv_rx: AsyncMutex<mpsc::Receiver<Vec<u8>>>,
send_tx: std::sync::mpsc::SyncSender<Vec<u8>>,
closed: Arc<AtomicBool>,
radio: Arc<dyn AndroidRadio>,
}
impl AndroidStream {
fn from_endpoints(ep: StreamEndpoints, radio: Arc<dyn AndroidRadio>) -> Self {
Self {
ch_id: ep.ch_id,
remote: ep.remote,
send_mtu: ep.send_mtu,
recv_mtu: ep.recv_mtu,
recv_rx: AsyncMutex::new(ep.recv_rx),
send_tx: ep.send_tx,
closed: ep.closed,
radio,
}
}
}
impl Drop for AndroidStream {
fn drop(&mut self) {
self.closed.store(true, Ordering::Relaxed);
self.radio.close_channel(self.ch_id);
}
}
impl BleStream for AndroidStream {
async fn send(&self, data: &[u8]) -> Result<(), TransportError> {
if self.closed.load(Ordering::Relaxed) {
return Err(TransportError::Io(std::io::Error::other(
"BLE channel closed",
)));
}
// Pure channel push — no JNI on the hot path. The Kotlin writer thread
// pulls this via the bridge's next_send and writes the socket.
//
// Backpressure rather than drop on a full queue. The queue is deliberately
// shallow (SEND_QUEUE_CAP) so it can't bufferbloat the BLE link; awaiting a
// free slot here propagates flow control up through FSP/MMP to the TUN and
// TCP — keeping RTT low *without* the loss-driven throughput collapse that a
// shallow tail-drop queue causes.
let mut payload = data.to_vec();
loop {
if self.closed.load(Ordering::Relaxed) {
return Err(TransportError::Io(std::io::Error::other(
"BLE channel closed",
)));
}
match self.send_tx.try_send(payload) {
Ok(()) => return Ok(()),
Err(std::sync::mpsc::TrySendError::Full(p)) => {
payload = p;
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
}
Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
return Err(TransportError::Io(std::io::Error::other(
"BLE send: peer closed",
)));
}
}
}
}
async fn recv(&self, buf: &mut [u8]) -> Result<usize, TransportError> {
match self.recv_rx.lock().await.recv().await {
Some(packet) => {
let n = packet.len().min(buf.len());
buf[..n].copy_from_slice(&packet[..n]);
Ok(n)
}
// Sender dropped (channel closed) → peer-closed, per the BleStream
// contract (a zero-length recv means the peer closed).
None => Ok(0),
}
}
fn send_mtu(&self) -> u16 {
self.send_mtu
}
fn recv_mtu(&self) -> u16 {
self.recv_mtu
}
fn remote_addr(&self) -> &BleAddr {
&self.remote
}
}
/// Yields inbound channels Kotlin accepted.
pub struct AndroidAcceptor {
rx: Option<mpsc::Receiver<StreamEndpoints>>,
radio: Arc<dyn AndroidRadio>,
}
impl BleAcceptor for AndroidAcceptor {
type Stream = AndroidStream;
async fn accept(&mut self) -> Result<AndroidStream, TransportError> {
match self.rx.as_mut() {
Some(rx) => match rx.recv().await {
Some(ep) => Ok(AndroidStream::from_endpoints(ep, Arc::clone(&self.radio))),
None => std::future::pending().await, // fan-in closed; idle
},
None => std::future::pending().await, // acceptor already consumed
}
}
}
/// Yields discovered FIPS peers (the learned PSM is captured into the bridge map).
pub struct AndroidScanner {
rx: Option<mpsc::Receiver<BleAddr>>,
}
impl BleScanner for AndroidScanner {
async fn next(&mut self) -> Option<BleAddr> {
match self.rx.as_mut() {
Some(rx) => rx.recv().await,
None => None,
}
}
}
impl BleIo for AndroidIo {
type Stream = AndroidStream;
type Acceptor = AndroidAcceptor;
type Scanner = AndroidScanner;
async fn listen(&self, _psm: u16) -> Result<AndroidAcceptor, TransportError> {
// Android assigns the listener PSM; the `psm` arg from FIPS is ignored.
let os_psm = self.bridge.radio.listen();
if os_psm != 0 {
self.bridge.local_psm.store(os_psm, Ordering::Relaxed);
}
let rx = self
.bridge
.accept_rx
.lock()
.unwrap_or_else(|e| e.into_inner())
.take();
Ok(AndroidAcceptor {
rx,
radio: Arc::clone(&self.bridge.radio),
})
}
async fn connect(&self, addr: &BleAddr, psm: u16) -> Result<AndroidStream, TransportError> {
// Substitute the learned per-peer PSM for this address, if known.
let dial_psm = self.bridge.psm_map.resolve(addr, psm);
let connect_id = self.bridge.next_id.fetch_add(1, Ordering::Relaxed);
let (tx, rx) = oneshot::channel();
self.bridge
.connects
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(connect_id, tx);
self.bridge.radio.connect(connect_id, addr, dial_psm);
// FIPS already wraps connect() in a timeout, so we just await the result.
match rx.await {
Ok(ep) => Ok(AndroidStream::from_endpoints(
ep,
Arc::clone(&self.bridge.radio),
)),
Err(_) => {
self.bridge
.connects
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&connect_id);
Err(TransportError::Io(std::io::Error::other(format!(
"BLE connect to {addr} failed"
))))
}
}
}
async fn start_advertising(&self) -> Result<(), TransportError> {
self.bridge
.radio
.start_advertising(self.bridge.local_psm.load(Ordering::Relaxed));
Ok(())
}
async fn stop_advertising(&self) -> Result<(), TransportError> {
self.bridge.radio.stop_advertising();
Ok(())
}
async fn start_scanning(&self) -> Result<AndroidScanner, TransportError> {
// Re-learn PSMs / adverts each scan cycle (addresses rotate with MAC).
self.bridge.psm_map.clear();
self.bridge.clear_adverts();
self.bridge.radio.start_scanning();
let rx = self
.bridge
.scan_rx
.lock()
.unwrap_or_else(|e| e.into_inner())
.take();
Ok(AndroidScanner { rx })
}
fn local_addr(&self) -> Result<BleAddr, TransportError> {
Ok(BleAddr {
adapter: ANDROID_ADAPTER.to_string(),
device: [0, 0, 0, 0, 0, 0],
})
}
fn adapter_name(&self) -> &str {
ANDROID_ADAPTER
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::AtomicU16 as TestAtomicU16;
/// A mock radio that records commands and lets the test drive the bridge.
#[derive(Default)]
struct MockRadio {
listen_psm: TestAtomicU16,
scanning: AtomicBool,
advertising_psm: TestAtomicU16,
}
impl AndroidRadio for MockRadio {
fn listen(&self) -> u16 {
self.listen_psm.load(Ordering::Relaxed)
}
fn connect(&self, _connect_id: i64, _addr: &BleAddr, _psm: u16) {}
fn start_advertising(&self, psm: u16) {
self.advertising_psm.store(psm, Ordering::Relaxed);
}
fn stop_advertising(&self) {}
fn start_scanning(&self) {
self.scanning.store(true, Ordering::Relaxed);
}
fn stop_scanning(&self) {}
fn close_channel(&self, _ch_id: i64) {}
}
fn addr(n: u8) -> BleAddr {
BleAddr {
adapter: ANDROID_ADAPTER.to_string(),
device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
}
}
#[tokio::test]
async fn inbound_channel_recv_and_close() {
let radio = Arc::new(MockRadio::default());
let bridge = AndroidBleBridge::new(radio.clone());
let io = AndroidIo::new(Arc::clone(&bridge));
let mut acceptor = io.listen(0).await.unwrap();
// Kotlin accepts an inbound channel, then pushes a packet.
let ch_id = bridge.deliver_inbound(addr(1), 512, 512);
assert!(ch_id > 0);
assert!(bridge.deliver_recv(ch_id, b"hello"));
let stream = acceptor.accept().await.unwrap();
assert_eq!(stream.remote_addr(), &addr(1));
assert_eq!(stream.send_mtu(), 512);
let mut buf = [0u8; 64];
let n = stream.recv(&mut buf).await.unwrap();
assert_eq!(&buf[..n], b"hello");
// Closing the channel makes the next recv return 0 (peer closed).
bridge.channel_closed(ch_id);
let n = stream.recv(&mut buf).await.unwrap();
assert_eq!(n, 0);
}
#[tokio::test]
async fn outbound_send_is_pulled_by_next_send() {
let radio = Arc::new(MockRadio::default());
let bridge = AndroidBleBridge::new(radio.clone());
// Simulate an accepted channel and grab its stream.
let mut acceptor = AndroidIo::new(Arc::clone(&bridge)).listen(0).await.unwrap();
let ch_id = bridge.deliver_inbound(addr(2), 0, 0);
let stream = acceptor.accept().await.unwrap();
// 0 MTU falls back to the transport default.
assert!(stream.send_mtu() > 0);
stream.send(b"out").await.unwrap();
let pulled = bridge.next_send(ch_id, Duration::from_millis(100)).unwrap();
assert_eq!(pulled, b"out");
// Nothing more queued → times out (None).
assert!(bridge.next_send(ch_id, Duration::from_millis(10)).is_none());
}
#[tokio::test]
async fn scan_learns_psm_and_connect_substitutes_it() {
let radio = Arc::new(MockRadio::default());
let bridge = AndroidBleBridge::new(radio.clone());
let io = AndroidIo::new(Arc::clone(&bridge));
let mut scanner = io.start_scanning().await.unwrap();
assert!(radio.scanning.load(Ordering::Relaxed));
bridge.deliver_scan(addr(3), 0x00C1, -50);
assert_eq!(scanner.next().await, Some(addr(3)));
// The advert is also recorded for the developer UI.
let adverts = bridge.advert_views();
assert_eq!(adverts.len(), 1);
assert_eq!(adverts[0].psm, 0x00C1);
assert_eq!(adverts[0].rssi, -50);
// The learned PSM is what a later dial would use (over FIPS's default).
assert_eq!(bridge.psm_map.resolve(&addr(3), DEFAULT_PSM), 0x00C1);
}
#[tokio::test]
async fn advertise_uses_os_assigned_listen_psm() {
let radio = Arc::new(MockRadio::default());
radio.listen_psm.store(0x0099, Ordering::Relaxed);
let bridge = AndroidBleBridge::new(radio.clone());
let io = AndroidIo::new(Arc::clone(&bridge));
let _ = io.listen(0).await.unwrap(); // learns the OS PSM
io.start_advertising().await.unwrap();
assert_eq!(radio.advertising_psm.load(Ordering::Relaxed), 0x0099);
}
}
+94 -50
View File
@@ -1,9 +1,11 @@
//! BLE L2CAP Transport Implementation
//!
//! Provides BLE-based transport for FIPS peer communication using L2CAP
//! Connection-Oriented Channels (CoC) in SeqPacket mode. L2CAP CoC
//! preserves message boundaries (unlike TCP byte streams), so no FMP
//! framing is needed — each send/recv is one FIPS packet.
//! Connection-Oriented Channels (CoC). BlueZ (SeqPacket) preserves message
//! boundaries, but stream-oriented backends (Android `BluetoothSocket`,
//! CoreBluetooth) do not, so the receive path recovers FIPS packet boundaries
//! with the shared FMP framer (`tcp::stream::read_fmp_packet`) over a
//! [`stream_read::BleStreamRead`] adapter — reliable on every platform.
//!
//! ## Architecture
//!
@@ -22,7 +24,15 @@ pub mod addr;
pub mod discovery;
pub mod io;
pub mod pool;
pub mod psm;
pub mod stats;
pub mod stream_read;
// The Android backend (radio in Kotlin, bytes over a bridge). Compiled on
// Android, and under `cfg(test)` on any host so its channel logic is unit-tested
// without a device. Not built into non-test desktop builds.
#[cfg(any(target_os = "android", test))]
pub mod android_io;
use super::{
ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr,
@@ -35,6 +45,9 @@ use discovery::DiscoveryBuffer;
use io::{BleIo, BleScanner, BleStream};
use pool::{BleConnection, ConnectionPool};
use stats::BleStats;
use stream_read::BleStreamRead;
use crate::transport::tcp::stream::{StreamError, read_fmp_packet};
use secp256k1::XOnlyPublicKey;
use std::collections::HashMap;
@@ -55,7 +68,12 @@ pub const DEFAULT_PSM: u16 = 0x0085;
#[cfg(all(bluer_available, not(test)))]
pub type DefaultBleTransport = BleTransport<io::BluerIo>;
#[cfg(any(not(bluer_available), test))]
// Android: the Kotlin-radio backend over the byte-bridge.
#[cfg(all(target_os = "android", not(test)))]
pub type DefaultBleTransport = BleTransport<android_io::AndroidIo>;
// Everything else (macOS/musl/host) and all test builds: the in-memory mock.
#[cfg(any(all(not(bluer_available), not(target_os = "android")), test))]
pub type DefaultBleTransport = BleTransport<io::MockBleIo>;
// ============================================================================
@@ -364,9 +382,15 @@ impl<I: BleIo> BleTransport<I> {
}
};
// One buffering reader per connection, shared by the pubkey exchange
// and the receive loop so coalesced bytes are never dropped.
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
// Pre-handshake pubkey exchange (temporary, pre-XX)
if let Some(ref our_pubkey) = self.local_pubkey {
match pubkey_exchange(&stream, our_pubkey).await {
match pubkey_exchange(&mut reader, our_pubkey).await {
Ok(peer_pubkey) => {
debug!(addr = %addr, "BLE outbound pubkey exchange complete");
self.discovery_buffer
@@ -379,24 +403,26 @@ impl<I: BleIo> BleTransport<I> {
}
}
self.promote_connection(addr, &ble_addr, stream).await
self.promote_connection(addr, &ble_addr, stream, reader)
.await
}
/// Promote a newly established stream into the connection pool.
///
/// Spawns the receive loop and inserts into the pool with eviction.
/// Spawns the receive loop (driven by `reader`) and inserts into the pool
/// with eviction.
async fn promote_connection(
&self,
addr: &TransportAddr,
ble_addr: &BleAddr,
stream: I::Stream,
stream: Arc<I::Stream>,
reader: BleStreamRead<I::Stream>,
) -> Result<(), TransportError> {
let send_mtu = stream.send_mtu();
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
let recv_task = tokio::spawn(receive_loop(
Arc::clone(&stream),
reader,
addr.clone(),
Arc::clone(&self.pool),
self.packet_tx.clone(),
@@ -484,9 +510,16 @@ impl<I: BleIo> BleTransport<I> {
match result {
Ok(Ok(stream)) => {
let send_mtu = stream.send_mtu();
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
// Shared reader: pubkey exchange + receive loop read the
// same buffer so coalesced bytes survive the handoff.
let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
// Pre-handshake pubkey exchange (temporary, pre-XX)
if let Some(ref our_pubkey) = local_pubkey {
match pubkey_exchange(&stream, our_pubkey).await {
match pubkey_exchange(&mut reader, our_pubkey).await {
Ok(peer_pubkey) => {
debug!(addr = %addr_clone, "BLE outbound pubkey exchange complete");
discovery_buffer.add_peer_with_pubkey(&ble_addr, peer_pubkey);
@@ -501,12 +534,8 @@ impl<I: BleIo> BleTransport<I> {
}
}
let send_mtu = stream.send_mtu();
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
let recv_task = tokio::spawn(receive_loop(
Arc::clone(&stream),
reader,
addr_clone.clone(),
Arc::clone(&pool),
packet_tx,
@@ -677,31 +706,38 @@ const PUBKEY_EXCHANGE_TIMEOUT_SECS: u64 = 5;
/// Exchange public keys over a newly established L2CAP connection.
///
/// Both sides send `[0x00][our_pubkey:32]` and receive the peer's.
/// Both sides send `[0x00][our_pubkey:32]` and receive the peer's. Reads go
/// through the connection's [`BleStreamRead`] buffer so a peer that fragments
/// the 33-byte message (stream-oriented backends) is read correctly, and any
/// bytes it coalesced after the pubkey stay buffered for the receive loop.
/// Returns the peer's XOnlyPublicKey on success.
async fn pubkey_exchange<S: BleStream>(
stream: &S,
reader: &mut BleStreamRead<S>,
local_pubkey: &[u8; 32],
) -> Result<XOnlyPublicKey, TransportError> {
use tokio::io::AsyncReadExt;
// Send our pubkey
let mut msg = [0u8; PUBKEY_EXCHANGE_SIZE];
msg[0] = PUBKEY_EXCHANGE_PREFIX;
msg[1..].copy_from_slice(local_pubkey);
stream.send(&msg).await?;
reader.stream().send(&msg).await?;
// Receive peer's pubkey (with timeout to prevent indefinite blocking)
// Receive peer's pubkey (with timeout to prevent indefinite blocking).
// read_exact reassembles across fragmented recvs and leaves any trailing
// bytes in the reader's buffer.
let mut buf = [0u8; PUBKEY_EXCHANGE_SIZE];
let timeout = std::time::Duration::from_secs(PUBKEY_EXCHANGE_TIMEOUT_SECS);
let n = match tokio::time::timeout(timeout, stream.recv(&mut buf)).await {
Ok(result) => result?,
match tokio::time::timeout(timeout, reader.read_exact(&mut buf)).await {
Ok(Ok(_)) => {}
Ok(Err(e)) => {
return Err(TransportError::RecvFailed(format!(
"pubkey exchange: {}",
e
)));
}
Err(_) => return Err(TransportError::Timeout),
};
if n != PUBKEY_EXCHANGE_SIZE {
return Err(TransportError::RecvFailed(format!(
"pubkey exchange: expected {} bytes, got {}",
PUBKEY_EXCHANGE_SIZE, n
)));
}
if buf[0] != PUBKEY_EXCHANGE_PREFIX {
return Err(TransportError::RecvFailed(format!(
"pubkey exchange: bad prefix 0x{:02X}",
@@ -751,10 +787,13 @@ async fn accept_loop<A>(
let send_mtu = stream.send_mtu();
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
// Shared reader for pubkey exchange + receive loop.
let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
// Pre-handshake pubkey exchange (temporary, pre-XX)
if let Some(ref our_pubkey) = local_pubkey {
match pubkey_exchange(&stream, our_pubkey).await {
match pubkey_exchange(&mut reader, our_pubkey).await {
Ok(peer_pubkey) => {
debug!(addr = %ta, "BLE inbound pubkey exchange complete");
discovery_buffer.add_peer_with_pubkey(&addr, peer_pubkey);
@@ -780,11 +819,9 @@ async fn accept_loop<A>(
}
}
let stream = Arc::new(stream);
// Spawn receive loop
let recv_task = tokio::spawn(receive_loop(
Arc::clone(&stream),
reader,
ta.clone(),
Arc::clone(&pool),
packet_tx.clone(),
@@ -829,8 +866,14 @@ async fn accept_loop<A>(
}
/// Receive loop: reads packets from a BLE stream and delivers to node.
async fn receive_loop<S: BleStream>(
stream: Arc<S>,
///
/// Recovers FIPS packet boundaries from the byte stream via the shared FMP
/// framer ([`read_fmp_packet`]) rather than assuming one `recv` is one packet.
/// L2CAP only preserves SDU boundaries on BlueZ (SeqPacket); stream-oriented
/// backends (Android, CoreBluetooth) fragment and coalesce, so the framer's
/// length-prefixed reads are what make delivery reliable across platforms.
async fn receive_loop<S: BleStream + 'static>(
mut reader: BleStreamRead<S>,
addr: TransportAddr,
pool: Arc<Mutex<ConnectionPool<Arc<S>>>>,
packet_tx: PacketTx,
@@ -838,21 +881,21 @@ async fn receive_loop<S: BleStream>(
stats: Arc<BleStats>,
recv_mtu: u16,
) {
let mut buf = vec![0u8; recv_mtu as usize];
loop {
match stream.recv(&mut buf).await {
Ok(0) => {
debug!(addr = %addr, "BLE connection closed by peer");
break;
}
Ok(n) => {
stats.record_recv(n);
let packet = ReceivedPacket::new(transport_id, addr.clone(), buf[..n].to_vec());
if packet_tx.send(packet).await.is_err() {
match read_fmp_packet(&mut reader, recv_mtu).await {
Ok(packet) => {
stats.record_recv(packet.len());
let received = ReceivedPacket::new(transport_id, addr.clone(), packet);
if packet_tx.send(received).await.is_err() {
trace!("BLE packet_tx closed, stopping receive loop");
break;
}
}
// Clean peer close (EOF at a packet boundary) is expected, not an error.
Err(StreamError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
debug!(addr = %addr, "BLE connection closed by peer");
break;
}
Err(e) => {
debug!(addr = %addr, error = %e, "BLE receive error");
stats.record_recv_error();
@@ -986,7 +1029,12 @@ async fn scan_probe_loop<I: io::BleIo>(
// Pubkey exchange, then promote connection to pool
let ta = addr.to_transport_addr();
match pubkey_exchange(&stream, &our_pubkey).await {
let send_mtu = stream.send_mtu();
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
// Shared reader for pubkey exchange + receive loop.
let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
match pubkey_exchange(&mut reader, &our_pubkey).await {
Ok(peer_pubkey) => {
debug!(addr = %addr, "BLE probe complete");
@@ -1005,12 +1053,8 @@ async fn scan_probe_loop<I: io::BleIo>(
}
// Promote connection to pool — no second L2CAP connect needed
let send_mtu = stream.send_mtu();
let recv_mtu = stream.recv_mtu();
let stream = Arc::new(stream);
let recv_task = tokio::spawn(receive_loop(
Arc::clone(&stream),
reader,
ta.clone(),
Arc::clone(&pool),
packet_tx.clone(),
+192
View File
@@ -0,0 +1,192 @@
//! Per-peer PSM discovery (BLE v2).
//!
//! On the platforms that matter the L2CAP listener PSM is **OS-assigned**, not
//! chosen by the app (Android `listenUsingInsecureL2capChannel`, macOS
//! `CBPeripheralManager.publishL2CAPChannel`); only BlueZ can bind a fixed one.
//! So rather than relying on the fixed `DEFAULT_PSM` (0x0085), every node
//! **advertises its own listener PSM** as a 16-bit little-endian value in the
//! service-data field keyed on the FIPS service UUID, and every dialer **reads a
//! peer's advertised PSM before `connect()`**. The fixed-PSM assumption was a
//! BlueZ quirk; this makes discovery symmetric across all backends.
//!
//! This module holds the platform-agnostic pieces shared by every `BleIo`
//! backend (`BluerIo`, `BluestIo`, `AndroidIo`): the service-data **codec** and
//! the short-lived **`BleAddr → PSM` map**. The per-backend advertise/scan/dial
//! wiring lives in the backends. See
//! [`docs/reference/ble-wire.md`](../../../../docs/reference/ble-wire.md) and
//! [`docs/design/ble-interop.md`](../../../../docs/design/ble-interop.md).
use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU16, Ordering};
use super::addr::BleAddr;
/// Encode a listener PSM as the 2-byte little-endian service-data payload
/// advertised under the FIPS service UUID.
///
/// The legacy advertising PDU caps at ~31 bytes, so a 128-bit UUID + this
/// 2-byte value is the tight, legacy-safe layout (see ble-wire.md).
pub fn encode_psm(psm: u16) -> [u8; 2] {
psm.to_le_bytes()
}
/// Decode a peer's advertised PSM from its FIPS service-data payload.
///
/// Returns `None` when fewer than 2 bytes are present — e.g. a legacy,
/// UUID-only advert that carries no PSM, for which the dialer falls back to
/// [`DEFAULT_PSM`](super::DEFAULT_PSM). Any bytes beyond the first two are
/// ignored, so the encoding can grow without breaking older readers.
pub fn decode_psm(data: &[u8]) -> Option<u16> {
match data {
[lo, hi, ..] => Some(u16::from_le_bytes([*lo, *hi])),
_ => None,
}
}
/// Short-lived map of discovered peer addresses to their advertised listener
/// PSM, populated by the scan loop and consulted by `connect()`.
///
/// Keyed on [`BleAddr`], which **rotates** with MAC randomization, so entries
/// are transient: they are re-learned each scan cycle rather than cached
/// durably. A stale entry merely causes one dial failure and a re-probe on the
/// next scan tick (see "Why MAC randomization is harmless" in ble-interop.md).
#[derive(Debug, Default)]
pub struct PsmMap {
inner: Mutex<HashMap<BleAddr, u16>>,
/// The most-recently-learned PSM across all peers. A node advertises exactly
/// one OS-assigned listener PSM, so when an exact-address lookup misses — the
/// common case, since the peer's RPA rotates between the scan that learned the
/// PSM and the dial — this is a far better guess than the fixed legacy default,
/// which no one listens on. 0 = unset.
last: AtomicU16,
}
impl PsmMap {
/// Create an empty map.
pub fn new() -> Self {
Self::default()
}
/// Record a peer's advertised PSM, learned from a scan advert. A later
/// advert for the same address overwrites the earlier value.
pub fn learn(&self, addr: &BleAddr, psm: u16) {
self.lock().insert(addr.clone(), psm);
if psm != 0 {
self.last.store(psm, Ordering::Relaxed);
}
}
/// Look up a peer's learned PSM, if one has been seen this scan cycle.
pub fn lookup(&self, addr: &BleAddr) -> Option<u16> {
self.lock().get(addr).copied()
}
/// Resolve the PSM to dial for `addr`: the learned per-peer PSM if known,
/// otherwise `fallback` (the configured PSM, else the legacy
/// [`DEFAULT_PSM`](super::DEFAULT_PSM)).
pub fn resolve(&self, addr: &BleAddr, fallback: u16) -> u16 {
if let Some(psm) = self.lookup(addr) {
return psm;
}
// Exact-address miss — almost always because the peer's MAC rotated
// between the scan that learned its PSM and this dial. Use the most
// recently learned PSM rather than the legacy default (never a real
// listener); a wrong guess only costs a dial-retry and a re-probe.
match self.last.load(Ordering::Relaxed) {
0 => fallback,
last => last,
}
}
/// Forget a single learned entry (e.g. after a dial failure).
pub fn forget(&self, addr: &BleAddr) {
self.lock().remove(addr);
}
/// Drop all learned entries. Called at the start of a scan cycle, since
/// addresses rotate and a dropped PSM only costs a dial-retry.
pub fn clear(&self) {
self.lock().clear();
}
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<BleAddr, u16>> {
self.inner.lock().unwrap_or_else(|e| e.into_inner())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::transport::ble::DEFAULT_PSM;
fn addr(n: u8) -> BleAddr {
BleAddr {
adapter: "ble0".to_string(),
device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
}
}
#[test]
fn codec_roundtrips_little_endian() {
// 0x0085 -> [0x85, 0x00]; explicit LE byte order matches the wire spec.
assert_eq!(encode_psm(0x0085), [0x85, 0x00]);
assert_eq!(encode_psm(0x1234), [0x34, 0x12]);
for psm in [0u16, 1, 0x0085, 0x0080, 0x00FF, 0x1234, u16::MAX] {
assert_eq!(decode_psm(&encode_psm(psm)), Some(psm));
}
}
#[test]
fn decode_rejects_short_payload_but_ignores_trailing() {
assert_eq!(decode_psm(&[]), None); // legacy UUID-only advert
assert_eq!(decode_psm(&[0x85]), None); // truncated
// Forward-compatible: trailing bytes beyond the PSM are ignored.
assert_eq!(decode_psm(&[0x85, 0x00, 0xFF, 0xFF]), Some(0x0085));
}
#[test]
fn learn_lookup_and_overwrite() {
let map = PsmMap::new();
assert_eq!(map.lookup(&addr(1)), None);
map.learn(&addr(1), 0x0091);
assert_eq!(map.lookup(&addr(1)), Some(0x0091));
// A later advert for the same address overwrites.
map.learn(&addr(1), 0x00A0);
assert_eq!(map.lookup(&addr(1)), Some(0x00A0));
// Distinct addresses are independent.
map.learn(&addr(2), 0x00B0);
assert_eq!(map.lookup(&addr(1)), Some(0x00A0));
assert_eq!(map.lookup(&addr(2)), Some(0x00B0));
}
#[test]
fn resolve_prefers_learned_then_falls_back() {
let map = PsmMap::new();
// No learned PSM yet -> legacy default.
assert_eq!(map.resolve(&addr(1), DEFAULT_PSM), DEFAULT_PSM);
map.learn(&addr(1), 0x0091);
assert_eq!(map.resolve(&addr(1), DEFAULT_PSM), 0x0091);
// An unseen address (e.g. the peer's MAC rotated since we learned its PSM)
// dials the most-recently-learned PSM, not the legacy default.
assert_eq!(map.resolve(&addr(9), DEFAULT_PSM), 0x0091);
}
#[test]
fn forget_and_clear_drop_entries() {
let map = PsmMap::new();
map.learn(&addr(1), 0x0091);
map.learn(&addr(2), 0x0092);
map.forget(&addr(1));
assert_eq!(map.lookup(&addr(1)), None);
assert_eq!(map.lookup(&addr(2)), Some(0x0092));
map.clear();
assert_eq!(map.lookup(&addr(2)), None);
}
}
+175
View File
@@ -0,0 +1,175 @@
//! `AsyncRead` adapter over a datagram-shaped [`BleStream`].
//!
//! L2CAP delivers different boundary guarantees per platform:
//!
//! - **BlueZ** (`SOCK_SEQPACKET`) preserves SDU boundaries — one `recv` is
//! exactly one FIPS packet.
//! - **Android** (`BluetoothSocket` input stream) and **CoreBluetooth** are
//! byte-stream oriented — a `recv` may return a fragment of a packet, or
//! several packets coalesced.
//!
//! FIPS packets are self-delimiting via the 4-byte FMP common prefix, so the
//! shared framer [`crate::transport::tcp::stream::read_fmp_packet`] can recover
//! boundaries from any byte stream. This adapter turns a [`BleStream`] (whose
//! `recv` fills a `&mut [u8]`) into the [`AsyncRead`] that framer expects. On a
//! SeqPacket backend it's a no-op pass-through; on a stream backend it
//! reassembles. Either way the layer above sees one whole packet per read.
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
use super::io::BleStream;
/// An in-flight `recv`: owns its scratch buffer and yields an owned `Vec`, so
/// the future is `'static` and can live across `poll_read` calls.
type PendingRecv = Pin<Box<dyn Future<Output = std::io::Result<Vec<u8>>> + Send>>;
/// Buffers a [`BleStream`] into an [`AsyncRead`] byte stream.
pub struct BleStreamRead<S: BleStream + 'static> {
stream: Arc<S>,
/// Per-`recv` scratch size; also the framer's MTU bound.
mtu: u16,
/// Bytes from the last `recv` not yet consumed by the framer.
leftover: Vec<u8>,
/// Read cursor into `leftover`.
pos: usize,
/// In-flight `recv`, if one is underway.
pending: Option<PendingRecv>,
}
impl<S: BleStream + 'static> BleStreamRead<S> {
/// Wrap a stream. `mtu` is the per-`recv` scratch size (use the channel's
/// recv MTU).
pub fn new(stream: Arc<S>, mtu: u16) -> Self {
Self {
stream,
mtu: mtu.max(1),
leftover: Vec::new(),
pos: 0,
pending: None,
}
}
/// The wrapped stream, for sending on the same channel (e.g. the pubkey
/// exchange writes here while reads come through the buffer).
pub fn stream(&self) -> &S {
&self.stream
}
}
impl<S: BleStream + 'static> AsyncRead for BleStreamRead<S> {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
dst: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
// BleStreamRead is Unpin (all fields are), so get_mut is sound.
let this = self.get_mut();
loop {
// Serve buffered bytes first.
if this.pos < this.leftover.len() {
let avail = &this.leftover[this.pos..];
let n = avail.len().min(dst.remaining());
dst.put_slice(&avail[..n]);
this.pos += n;
return Poll::Ready(Ok(()));
}
// Buffer drained: pull the next datagram. The future owns its
// scratch and returns it truncated, so it captures only `Arc<S>`.
if this.pending.is_none() {
let stream = Arc::clone(&this.stream);
let mtu = this.mtu as usize;
this.pending = Some(Box::pin(async move {
let mut scratch = vec![0u8; mtu];
let n = stream
.recv(&mut scratch)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
scratch.truncate(n);
Ok(scratch)
}));
}
match this.pending.as_mut().unwrap().as_mut().poll(cx) {
Poll::Ready(Ok(buf)) => {
this.pending = None;
// A zero-length recv is the BleStream peer-closed signal;
// leaving `dst` unfilled surfaces as EOF to the framer.
if buf.is_empty() {
return Poll::Ready(Ok(()));
}
this.leftover = buf;
this.pos = 0;
// loop to copy out
}
Poll::Ready(Err(e)) => {
this.pending = None;
return Poll::Ready(Err(e));
}
Poll::Pending => return Poll::Pending,
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::transport::ble::addr::BleAddr;
use crate::transport::ble::io::MockBleStream;
use tokio::io::AsyncReadExt;
fn addr(n: u8) -> BleAddr {
BleAddr {
adapter: "hci0".to_string(),
device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
}
}
/// Several small recvs reassemble into one read_exact.
#[tokio::test]
async fn reassembles_fragmented_recv() {
let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048);
// Peer sends three fragments that together form one 10-byte message.
a.send(b"abc").await.unwrap();
a.send(b"defg").await.unwrap();
a.send(b"hij").await.unwrap();
let mut reader = BleStreamRead::new(Arc::new(b), 2048);
let mut out = [0u8; 10];
reader.read_exact(&mut out).await.unwrap();
assert_eq!(&out, b"abcdefghij");
}
/// A recv carrying several packets is served across multiple reads without
/// dropping the tail (the coalescing case).
#[tokio::test]
async fn serves_coalesced_recv_in_pieces() {
let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048);
a.send(b"0123456789").await.unwrap();
let mut reader = BleStreamRead::new(Arc::new(b), 2048);
let mut first = [0u8; 4];
reader.read_exact(&mut first).await.unwrap();
assert_eq!(&first, b"0123");
let mut rest = [0u8; 6];
reader.read_exact(&mut rest).await.unwrap();
assert_eq!(&rest, b"456789");
}
/// A closed stream surfaces as EOF (read_exact errors with UnexpectedEof).
#[tokio::test]
async fn closed_stream_is_eof() {
let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048);
drop(a); // peer closes
let mut reader = BleStreamRead::new(Arc::new(b), 2048);
let mut out = [0u8; 4];
let err = reader.read_exact(&mut out).await.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
}
}
+44 -44
View File
@@ -11,15 +11,15 @@ pub mod tcp;
pub mod tor;
pub mod udp;
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub mod ethernet;
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
pub mod ble;
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
use ble::DefaultBleTransport;
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
use ethernet::EthernetTransport;
#[cfg(test)]
use loopback::LoopbackTransport;
@@ -894,7 +894,7 @@ pub enum TransportHandle {
/// UDP/IP transport.
Udp(UdpTransport),
/// Raw Ethernet transport.
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
Ethernet(EthernetTransport),
/// TCP/IP transport.
Tcp(TcpTransport),
@@ -903,7 +903,7 @@ pub enum TransportHandle {
/// Nym mixnet transport (via SOCKS5).
Nym(NymTransport),
/// BLE L2CAP transport.
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
Ble(DefaultBleTransport),
/// In-process loopback transport (test harness only).
#[cfg(test)]
@@ -915,12 +915,12 @@ impl TransportHandle {
pub async fn start(&mut self) -> Result<(), TransportError> {
match self {
TransportHandle::Udp(t) => t.start_async().await,
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.start_async().await,
#[cfg(test)]
TransportHandle::Loopback(t) => t.start_async().await,
@@ -931,12 +931,12 @@ impl TransportHandle {
pub async fn stop(&mut self) -> Result<(), TransportError> {
match self {
TransportHandle::Udp(t) => t.stop_async().await,
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.stop_async().await,
#[cfg(test)]
TransportHandle::Loopback(t) => t.stop_async().await,
@@ -947,12 +947,12 @@ impl TransportHandle {
pub async fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<usize, TransportError> {
match self {
TransportHandle::Udp(t) => t.send_async(addr, data).await,
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.send_async(addr, data).await,
#[cfg(test)]
TransportHandle::Loopback(t) => t.send_async(addr, data).await,
@@ -963,12 +963,12 @@ impl TransportHandle {
pub fn transport_id(&self) -> TransportId {
match self {
TransportHandle::Udp(t) => t.transport_id(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.transport_id(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.transport_id(),
@@ -979,12 +979,12 @@ impl TransportHandle {
pub fn name(&self) -> Option<&str> {
match self {
TransportHandle::Udp(t) => t.name(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(t) => t.name(),
TransportHandle::Tcp(t) => t.name(),
TransportHandle::Tor(t) => t.name(),
TransportHandle::Nym(t) => t.name(),
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.name(),
#[cfg(test)]
TransportHandle::Loopback(_) => None,
@@ -995,12 +995,12 @@ impl TransportHandle {
pub fn transport_type(&self) -> &TransportType {
match self {
TransportHandle::Udp(t) => t.transport_type(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.transport_type(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.transport_type(),
@@ -1011,12 +1011,12 @@ impl TransportHandle {
pub fn state(&self) -> TransportState {
match self {
TransportHandle::Udp(t) => t.state(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(t) => t.state(),
TransportHandle::Tcp(t) => t.state(),
TransportHandle::Tor(t) => t.state(),
TransportHandle::Nym(t) => t.state(),
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.state(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.state(),
@@ -1027,12 +1027,12 @@ impl TransportHandle {
pub fn mtu(&self) -> u16 {
match self {
TransportHandle::Udp(t) => t.mtu(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(t) => t.mtu(),
TransportHandle::Tcp(t) => t.mtu(),
TransportHandle::Tor(t) => t.mtu(),
TransportHandle::Nym(t) => t.mtu(),
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.mtu(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.mtu(),
@@ -1046,12 +1046,12 @@ impl TransportHandle {
pub fn link_mtu(&self, addr: &TransportAddr) -> u16 {
match self {
TransportHandle::Udp(t) => t.link_mtu(addr),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.link_mtu(addr),
#[cfg(test)]
TransportHandle::Loopback(t) => t.link_mtu(addr),
@@ -1062,12 +1062,12 @@ impl TransportHandle {
pub fn local_addr(&self) -> Option<std::net::SocketAddr> {
match self {
TransportHandle::Udp(t) => t.local_addr(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(_) => None,
TransportHandle::Tcp(t) => t.local_addr(),
TransportHandle::Tor(_) => None,
TransportHandle::Nym(_) => None,
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(_) => None,
#[cfg(test)]
TransportHandle::Loopback(_) => None,
@@ -1078,12 +1078,12 @@ impl TransportHandle {
pub fn interface_name(&self) -> Option<&str> {
match self {
TransportHandle::Udp(_) => None,
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(t) => Some(t.interface_name()),
TransportHandle::Tcp(_) => None,
TransportHandle::Tor(_) => None,
TransportHandle::Nym(_) => None,
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(_) => None,
#[cfg(test)]
TransportHandle::Loopback(_) => None,
@@ -1118,12 +1118,12 @@ impl TransportHandle {
pub fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError> {
match self {
TransportHandle::Udp(t) => t.discover(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(t) => t.discover(),
TransportHandle::Tcp(t) => t.discover(),
TransportHandle::Tor(t) => t.discover(),
TransportHandle::Nym(t) => t.discover(),
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.discover(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.discover(),
@@ -1134,12 +1134,12 @@ impl TransportHandle {
pub fn auto_connect(&self) -> bool {
match self {
TransportHandle::Udp(t) => t.auto_connect(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.auto_connect(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.auto_connect(),
@@ -1150,12 +1150,12 @@ impl TransportHandle {
pub fn accept_connections(&self) -> bool {
match self {
TransportHandle::Udp(t) => t.accept_connections(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.accept_connections(),
#[cfg(test)]
TransportHandle::Loopback(t) => t.accept_connections(),
@@ -1172,12 +1172,12 @@ impl TransportHandle {
pub async fn connect(&self, addr: &TransportAddr) -> Result<(), TransportError> {
match self {
TransportHandle::Udp(_) => Ok(()), // connectionless
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.connect_async(addr).await,
#[cfg(test)]
TransportHandle::Loopback(_) => Ok(()), // connectionless
@@ -1192,12 +1192,12 @@ impl TransportHandle {
pub fn connection_state(&self, addr: &TransportAddr) -> ConnectionState {
match self {
TransportHandle::Udp(_) => ConnectionState::Connected,
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.connection_state_sync(addr),
#[cfg(test)]
TransportHandle::Loopback(_) => ConnectionState::Connected,
@@ -1211,12 +1211,12 @@ impl TransportHandle {
pub async fn close_connection(&self, addr: &TransportAddr) {
match self {
TransportHandle::Udp(t) => t.close_connection(addr),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
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")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => t.close_connection_async(addr).await,
#[cfg(test)]
TransportHandle::Loopback(_) => {} // connectionless no-op
@@ -1236,12 +1236,12 @@ impl TransportHandle {
pub fn congestion(&self) -> TransportCongestion {
match self {
TransportHandle::Udp(t) => t.congestion(),
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(_) => TransportCongestion::default(),
TransportHandle::Tcp(_) => TransportCongestion::default(),
TransportHandle::Tor(_) => TransportCongestion::default(),
TransportHandle::Nym(_) => TransportCongestion::default(),
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(_) => TransportCongestion::default(),
#[cfg(test)]
TransportHandle::Loopback(_) => TransportCongestion::default(),
@@ -1256,7 +1256,7 @@ impl TransportHandle {
TransportHandle::Udp(t) => {
serde_json::to_value(t.stats().snapshot()).unwrap_or_default()
}
#[cfg(unix)]
#[cfg(any(target_os = "linux", target_os = "macos"))]
TransportHandle::Ethernet(t) => {
let snap = t.stats().snapshot();
serde_json::json!({
@@ -1281,7 +1281,7 @@ impl TransportHandle {
TransportHandle::Nym(t) => {
serde_json::to_value(t.stats().snapshot()).unwrap_or_default()
}
#[cfg(target_os = "linux")]
#[cfg(ble_available)]
TransportHandle::Ble(t) => {
serde_json::to_value(t.stats().snapshot()).unwrap_or_default()
}
+1 -1
View File
@@ -301,7 +301,7 @@ fn extract_pktinfo_ifindex(msg: &libc::msghdr) -> Option<u32> {
if cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_PKTINFO {
let data_ptr = unsafe { libc::CMSG_DATA(cmsg_ptr) } as *const libc::in6_pktinfo;
let pktinfo: libc::in6_pktinfo = unsafe { std::ptr::read_unaligned(data_ptr) };
return Some(pktinfo.ipi6_ifindex);
return Some(pktinfo.ipi6_ifindex as u32);
}
cmsg_ptr = unsafe { libc::CMSG_NXTHDR(msg, cmsg_ptr) };
}
+26
View File
@@ -1224,6 +1224,32 @@ mod windows_tun {
#[cfg(windows)]
pub use windows_tun::{TunDevice, TunWriter, run_tun_reader, shutdown_tun_interface};
// Android uses an app-owned TUN (the embedder owns the fd, e.g. an Android
// VpnService); FIPS never creates or configures a system TUN here. These no-op
// stubs stand in for the platform ops so the shared TunDevice code compiles.
#[cfg(target_os = "android")]
mod platform {
use super::TunError;
use std::net::Ipv6Addr;
pub fn is_ipv6_disabled() -> bool {
false
}
pub async fn interface_exists(_name: &str) -> bool {
false
}
pub async fn delete_interface(_name: &str) -> Result<(), TunError> {
Ok(())
}
pub async fn configure_interface(
_name: &str,
_addr: Ipv6Addr,
_mtu: u16,
) -> Result<(), TunError> {
Ok(())
}
}
#[cfg(target_os = "linux")]
mod platform {
use super::TunError;
+72
View File
@@ -72,3 +72,75 @@ and optional trace-level RUST_LOG, capturing per-rep diagnostics and a
mechanism-match summary across the run. Used for statistical reliability
characterization of known flake classes under calibrated stress, not as
a per-commit gate; not part of `ci-local.sh`.
## Running CI locally (`ci-local.sh`)
[`ci-local.sh`](ci-local.sh) runs the full local CI pipeline — build,
clippy, unit tests, and the integration suites (including the chaos
scenarios) — mirroring the GitHub `ci.yml` integration matrix. Run
`./ci-local.sh --help` for the full option list and `--list` for the
available suites. `--check-parity` verifies the local suite set matches
the GitHub matrix (see [check-ci-parity.sh](check-ci-parity.sh)).
### Per-run isolation and the `FIPS_CI_RUN_ID` override
Every invocation derives a **run id** and scopes all of its Docker
resources to it, so two simultaneous runs on the same host (for example,
one per git worktree, or an operator testing by hand while CI is in
flight) never collide:
- **Compose projects** are named `fipsci_<run-id>_<suite>`, so
container, network, and volume names are all prefixed per run.
- **Build images** are tagged `fips-test:<run-id>` and
`fips-test-app:<run-id>` (exported as `FIPS_TEST_IMAGE` /
`FIPS_TEST_APP_IMAGE` for the compose consumers).
- Each parallel chaos child gets a unique, non-overlapping `/24` in
`10.30.x` (via the sim `--subnet` override). `10.30.x` sits outside
Docker's default address pool and the fixed-subnet suites' `172.x`
ranges, so neither a sibling chaos child nor an auto-assigned network
can swallow a pinned subnet.
By default the run id is `<short-git-sha>-<random>` — the SHA portion
records *what code* a container is testing, the random suffix keeps
simultaneous runs of the same SHA disjoint. Override it for a
reproducible, attach-by-name debug session:
```sh
FIPS_CI_RUN_ID=mydebug ./ci-local.sh --only static-mesh
# containers are named fipsci_mydebug_static_fips-node-a, etc.
```
### Preemption-safety and exit codes
`ci-local.sh` is safe to cancel mid-run. A signal trap tears down *every*
compose project the run started (not just the current suite) and reaps
any in-flight parallel chaos children, bounded by a `timeout` so a stuck
`compose down` cannot wedge the trap. Exit codes distinguish a cancelled
run from a failing one:
| Code | Meaning |
| ---- | ------- |
| `0` | all stages passed |
| `1` | one or more stages failed |
| `130` | interrupted by SIGINT — cancelled, not a failure |
| `143` | terminated by SIGTERM — cancelled, not a failure |
A preempting CI worker (the push-triggered, CI-gated build pipeline that
kills an in-flight run when a newer same-branch tip arrives) maps
`130`/`143`*cancelled* (discard, do not record a failing commit), `0`
→ green, any other non-zero → red.
### Cleaning up leftover resources
Every CI-created container, network, and volume carries the label
`com.corganlabs.fips-ci=1`. If a run is hard-killed (SIGKILL, OOM, crash)
and leaves resources behind, reap them with:
```sh
./ci-local.sh --reap # or: ./ci-cleanup.sh
```
[`ci-cleanup.sh`](ci-cleanup.sh) force-removes everything bearing the CI
label or a `fipsci_` compose-project prefix; it is safe to run when there
is nothing to reap and safe to run repeatedly. Pass `--project-prefix` to
scope the sweep to a single run.
+2
View File
@@ -1,6 +1,8 @@
networks:
acl-net:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: 172.31.0.0/24
+2
View File
@@ -6,6 +6,8 @@
networks:
bt-net:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: 172.99.0.0/24
+4
View File
@@ -65,6 +65,7 @@ VERBOSE=""
SEED=""
DURATION=""
NODES=""
SUBNET=""
while [ $# -gt 0 ]; do
case "$1" in
@@ -72,6 +73,7 @@ while [ $# -gt 0 ]; do
--seed) SEED="$2"; shift 2 ;;
--duration) DURATION="$2"; shift 2 ;;
--nodes) NODES="$2"; shift 2 ;;
--subnet) SUBNET="$2"; shift 2 ;;
--list) list_scenarios ;;
-*) echo "Error: Unknown option '$1'" >&2; usage ;;
*)
@@ -135,6 +137,7 @@ PYTHON_ARGS=("$SCENARIO_FILE")
[ -n "$VERBOSE" ] && PYTHON_ARGS+=("$VERBOSE")
[ -n "$SEED" ] && PYTHON_ARGS+=("--seed" "$SEED")
[ -n "$DURATION" ] && PYTHON_ARGS+=("--duration" "$DURATION")
[ -n "$SUBNET" ] && PYTHON_ARGS+=("--subnet" "$SUBNET")
echo "=== FIPS Stochastic Simulation ==="
echo ""
@@ -143,6 +146,7 @@ echo " File: $SCENARIO_FILE"
[ -n "$SEED" ] && echo " Seed: $SEED (override)"
[ -n "$DURATION" ] && echo " Duration: ${DURATION}s (override)"
[ -n "$NODES" ] && echo " Nodes: $NODES (override)"
[ -n "$SUBNET" ] && echo " Subnet: $SUBNET (override)"
echo ""
# If --nodes is specified, create a patched copy of the scenario file
+8
View File
@@ -25,6 +25,12 @@ def main():
"--duration", type=int, default=None,
help="Override scenario duration in seconds",
)
parser.add_argument(
"--subnet", type=str, default=None,
help="Override topology subnet CIDR (e.g. 10.30.0.0/24); node IPs "
"derive from it. Used by CI to give each parallel run a "
"non-overlapping network.",
)
args = parser.parse_args()
level = logging.DEBUG if args.verbose else logging.INFO
@@ -48,6 +54,8 @@ def main():
print("Error: --duration must be >= 1", file=sys.stderr)
sys.exit(1)
scenario.duration_secs = args.duration
if args.subnet is not None:
scenario.topology.subnet = args.subnet
runner = SimRunner(scenario)
result = runner.run()
+2
View File
@@ -20,6 +20,8 @@ _COMPOSE_TEMPLATE = Template(
networks:
fips-net:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: {{ subnet }}
+106
View File
@@ -0,0 +1,106 @@
#!/bin/bash
# Reap FIPS CI docker resources: containers, networks, volumes, images.
#
# Force-removes everything created by ci-local.sh that is still around —
# whether a run finished cleanly, was preempted (SIGTERM/SIGKILL), OOM-killed,
# or crashed. Two complementary selectors make this robust no matter how a
# prior run died:
#
# 1. The CI label com.corganlabs.fips-ci=1 (attached to every direct
# `docker run`/network/volume ci-local drives).
# 2. The compose project-name prefix fipsci_ (every compose project ci-local
# starts is named fipsci_<run-id>_<suite>, so its containers/networks/
# volumes all carry com.docker.compose.project=fipsci_... and are named
# with that prefix).
#
# Usage:
# ci-cleanup.sh Reap ALL fips-ci resources (any run)
# ci-cleanup.sh --project-prefix P Restrict the compose-project sweep to
# names starting with P (scopes the reap
# to a single run; the label sweep still
# runs)
# ci-cleanup.sh --label L Override the CI label (default above)
# ci-cleanup.sh --images "a b,c" Also `docker rmi -f` these image tags
# (space- or comma-separated)
#
# Safe to run when there is nothing to reap, and safe to run repeatedly.
# Also reachable as `ci-local.sh --reap`.
set -uo pipefail
LABEL="com.corganlabs.fips-ci=1"
PROJECT_PREFIX="fipsci_" # broad default: every CI run
IMAGES=""
while [[ $# -gt 0 ]]; do
case "$1" in
--label) LABEL="$2"; shift 2 ;;
--project-prefix) PROJECT_PREFIX="$2"; shift 2 ;;
--images) IMAGES="$2"; shift 2 ;;
-h|--help) sed -n '2,/^set /{ /^set /d; s/^# \?//; p }' "$0"; exit 0 ;;
*) echo "Unknown option: $1" >&2; exit 2 ;;
esac
done
if ! command -v docker >/dev/null 2>&1; then
# No docker, nothing to reap.
exit 0
fi
if ! docker info >/dev/null 2>&1; then
# Daemon unreachable; treat as nothing to reap rather than wedging a caller.
exit 0
fi
# Each docker mutation is wrapped in `timeout` so a stuck daemon/resource can
# never wedge a caller (ci-local's signal trap relies on this being bounded).
TMO=30
# Distinct compose project names (read off container labels) that start with
# the configured prefix.
ci_projects() {
docker ps -a --format '{{.Label "com.docker.compose.project"}}' 2>/dev/null \
| grep -E "^${PROJECT_PREFIX}" | sort -u
}
reap_containers() {
# By CI label.
docker ps -aq --filter "label=${LABEL}" 2>/dev/null \
| xargs -r timeout "$TMO" docker rm -f >/dev/null 2>&1 || true
# By compose project (carried even when container_name is explicit).
local p
for p in $(ci_projects); do
docker ps -aq --filter "label=com.docker.compose.project=${p}" 2>/dev/null \
| xargs -r timeout "$TMO" docker rm -f >/dev/null 2>&1 || true
done
}
reap_networks() {
docker network ls -q --filter "label=${LABEL}" 2>/dev/null \
| xargs -r timeout "$TMO" docker network rm >/dev/null 2>&1 || true
# Compose networks are named <project>_<net> → match by name prefix so
# orphaned networks (whose containers are already gone) are still caught.
docker network ls --format '{{.Name}}' 2>/dev/null | grep -E "^${PROJECT_PREFIX}" \
| xargs -r timeout "$TMO" docker network rm >/dev/null 2>&1 || true
}
reap_volumes() {
docker volume ls -q --filter "label=${LABEL}" 2>/dev/null \
| xargs -r timeout "$TMO" docker volume rm >/dev/null 2>&1 || true
docker volume ls --format '{{.Name}}' 2>/dev/null | grep -E "^${PROJECT_PREFIX}" \
| xargs -r timeout "$TMO" docker volume rm >/dev/null 2>&1 || true
}
reap_images() {
[[ -z "$IMAGES" ]] && return 0
local imgs
read -ra imgs <<< "${IMAGES//,/ }"
[[ ${#imgs[@]} -eq 0 ]] && return 0
timeout "$TMO" docker rmi -f "${imgs[@]}" >/dev/null 2>&1 || true
}
# Order matters: containers reference networks/volumes, so drop them first.
reap_containers
reap_networks
reap_volumes
reap_images
exit 0
+154 -6
View File
@@ -14,6 +14,9 @@
# --list List available integration suites
# --check-parity Verify this suite set matches ci.yml's integration
# matrix (see testing/check-ci-parity.sh), then exit
# --reap Force-remove all leftover FIPS CI docker resources
# (containers/networks/volumes carrying the CI label or a
# fipsci_ compose project), then exit. See ci-cleanup.sh.
# -h, --help Show this help
#
# Integration suites (default coverage):
@@ -32,8 +35,13 @@
# tor-socks5, tor-directory
#
# Exit codes:
# 0 — all stages passed
# 1 — one or more stages failed
# 0 — all stages passed
# 1 — one or more stages failed
# 130 — interrupted by SIGINT (128 + 2; run was cancelled, not a failure)
# 143 — terminated by SIGTERM (128 + 15; run was cancelled, not a failure)
#
# A preempting CI worker maps 130/143 → "cancelled" (discard, do not record a
# failing commit), 0 → green, any other non-zero → red.
#
# ── CI parity invariant ─────────────────────────────────────────────────────
# This local default suite set and the GitHub integration matrix
@@ -193,6 +201,7 @@ while [[ $# -gt 0 ]]; do
-j|--jobs) PARALLEL_JOBS="$2"; shift 2 ;;
--list) list_suites ;;
--check-parity) exec "$SCRIPT_DIR/check-ci-parity.sh" ;;
--reap) exec "$SCRIPT_DIR/ci-cleanup.sh" ;;
-h|--help) usage ;;
*) echo "Unknown option: $1"; usage ;;
esac
@@ -214,6 +223,104 @@ record() {
fi
}
# ── Per-run isolation + signal-safe teardown ───────────────────────────────
#
# This script may be preempted (a CI worker sends SIGTERM, waits ~30s, then
# SIGKILL) so it can restart on a newer tip. To make that safe:
# * every docker resource is namespaced to THIS run (compose project prefix
# + per-run image tags) so a restart never collides with a dying run;
# * a trap tears down everything this run created on signal/exit, bounded by
# `timeout` so a stuck `down` cannot wedge the trap (SIGKILL is the backstop).
# Derive a run id: honor the worker's $FIPS_CI_RUN_ID, else <short-sha>-<rand>,
# else $$-<timestamp>. Sanitize to a valid compose project / image-tag token.
if [[ -n "${FIPS_CI_RUN_ID:-}" ]]; then
CI_RUN_ID="$FIPS_CI_RUN_ID"
else
_ci_sha="$(git -C "$PROJECT_ROOT" rev-parse --short HEAD 2>/dev/null || true)"
if [[ -n "$_ci_sha" ]]; then
CI_RUN_ID="${_ci_sha}-${RANDOM}${RANDOM}"
else
CI_RUN_ID="$$-$(date +%s)"
fi
fi
CI_RUN_ID="$(printf '%s' "$CI_RUN_ID" | tr '[:upper:]' '[:lower:]' | tr -c 'a-z0-9_-' '-')"
# A docker image tag and a compose project name must both START with an
# alphanumeric, so strip any leading -/_ left by sanitization.
CI_RUN_ID="$(printf '%s' "$CI_RUN_ID" | sed -E 's/^[^a-z0-9]+//')"
[[ -z "$CI_RUN_ID" ]] && CI_RUN_ID="r$$"
CI_PROJECT_PREFIX="fipsci_${CI_RUN_ID}"
CI_IMAGE_TEST="fips-test:${CI_RUN_ID}"
CI_IMAGE_APP="fips-test-app:${CI_RUN_ID}"
CI_LABEL="com.corganlabs.fips-ci=1"
# Exported so child suite scripts + their compose/`docker run` invocations
# inherit the run identity. Compose files read FIPS_TEST_IMAGE/FIPS_TEST_APP_IMAGE
# (default :latest when unset, preserving manual `docker compose` use).
export FIPS_CI_RUN_ID="$CI_RUN_ID"
export FIPS_TEST_IMAGE="$CI_IMAGE_TEST"
export FIPS_TEST_APP_IMAGE="$CI_IMAGE_APP"
# Per-suite compose project name: ${prefix}_<suite-or-compose-basename>. Keeps
# today's intra-run distinctness (one project per compose file / chaos child)
# while adding the cross-run prefix that scopes the reap.
ci_project() { printf '%s_%s' "$CI_PROJECT_PREFIX" "$1"; }
# PIDs of in-flight parallel chaos children (subshells). The trap signals these.
CI_CHAOS_PIDS=()
CI_CLEANED=0
# Best-effort, BOUNDED teardown of every docker resource THIS run may have
# created. Idempotent (guarded), so the signal and EXIT paths don't double-run.
ci_teardown() {
[[ $CI_CLEANED -eq 1 ]] && return 0
CI_CLEANED=1
# 1. Propagate to parallel chaos children and reap them (bounded).
if [[ ${#CI_CHAOS_PIDS[@]} -gt 0 ]]; then
kill -TERM "${CI_CHAOS_PIDS[@]}" 2>/dev/null || true
local _end=$(( SECONDS + 10 )) _p
for _p in "${CI_CHAOS_PIDS[@]}"; do
while kill -0 "$_p" 2>/dev/null && (( SECONDS < _end )); do
sleep 0.3
done
done
kill -KILL "${CI_CHAOS_PIDS[@]}" 2>/dev/null || true
wait "${CI_CHAOS_PIDS[@]}" 2>/dev/null || true
fi
# 2. Remove all compose projects + direct-run resources + per-run images
# for this run. ci-cleanup.sh wraps each docker op in `timeout`; bound
# the whole sweep too so the trap can never wedge.
timeout 150 bash "$SCRIPT_DIR/ci-cleanup.sh" \
--label "$CI_LABEL" \
--project-prefix "$CI_PROJECT_PREFIX" \
--images "$CI_IMAGE_TEST $CI_IMAGE_APP" >/dev/null 2>&1 || true
}
on_signal() {
local sig="$1"
# Block re-entry and stop the EXIT trap from overriding the signal code.
trap '' TERM INT EXIT
echo "" >&2
fail "Received SIG$sig — cancelling run, tearing down ${CI_PROJECT_PREFIX}"
ci_teardown
# 128 + signal number: distinct from 0 (green) / 1 (stage failed).
if [[ "$sig" == "TERM" ]]; then exit 143; else exit 130; fi
}
on_exit() {
local code=$?
trap '' TERM INT EXIT
ci_teardown
exit "$code"
}
trap 'on_signal TERM' TERM
trap 'on_signal INT' INT
trap 'on_exit' EXIT
# ── Stage 1: Build ─────────────────────────────────────────────────────────
run_build() {
@@ -317,6 +424,7 @@ run_static() {
local topology="$1"
local compose="testing/static/docker-compose.yml"
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project static)"
info "[$topology] Generating configs"
bash testing/static/scripts/generate-configs.sh "$topology" || { record "static-$topology" 1; return; }
@@ -341,6 +449,7 @@ run_static() {
run_rekey() {
local compose="testing/static/docker-compose.yml"
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project static)"
info "[rekey] Generating configs"
bash testing/static/scripts/generate-configs.sh rekey || { record "rekey" 1; return; }
@@ -369,6 +478,7 @@ run_rekey() {
run_admission_cap() {
local compose="testing/static/docker-compose.yml"
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project static)"
info "[admission-cap] Generating configs"
bash testing/static/scripts/generate-configs.sh mesh || { record "admission-cap" 1; return; }
@@ -395,6 +505,9 @@ run_chaos() {
local name="$1"
shift
local rc=0
# Distinct project per scenario (chaos children run in parallel). When
# invoked from a background subshell this export is local to that child.
export COMPOSE_PROJECT_NAME="$(ci_project "chaos-$name")"
info "[chaos/$name] Running simulation"
if bash testing/chaos/scripts/chaos.sh "$@" 2>&1; then
@@ -410,6 +523,7 @@ run_chaos() {
run_gateway() {
local compose="testing/static/docker-compose.yml"
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project static)"
info "[gateway] Generating configs"
bash testing/static/scripts/generate-configs.sh gateway gateway-test || { record "gateway" 1; return; }
@@ -434,6 +548,7 @@ run_gateway() {
# Run sidecar test
run_sidecar() {
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project sidecar)"
info "[sidecar] Running integration test"
if bash testing/sidecar/scripts/test-sidecar.sh --skip-build 2>&1; then
@@ -450,6 +565,7 @@ run_sidecar() {
run_rekey_accept_off() {
local compose="testing/static/docker-compose.yml"
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project static)"
info "[rekey-accept-off] Generating configs"
bash testing/static/scripts/generate-configs.sh rekey-accept-off || \
@@ -484,6 +600,7 @@ run_rekey_accept_off() {
run_rekey_outbound_only() {
local compose="testing/static/docker-compose.yml"
local rc=0
export COMPOSE_PROJECT_NAME="$(ci_project static)"
info "[rekey-outbound-only] Generating configs"
bash testing/static/scripts/generate-configs.sh rekey-outbound-only || \
@@ -512,6 +629,7 @@ run_rekey_outbound_only() {
# Run ACL allowlist integration test
run_acl_allowlist() {
export COMPOSE_PROJECT_NAME="$(ci_project acl)"
info "[acl-allowlist] Running integration test"
if bash testing/acl-allowlist/test.sh --skip-build 2>&1; then
record "acl-allowlist" 0
@@ -522,6 +640,7 @@ run_acl_allowlist() {
# Run firewall baseline integration test
run_firewall() {
export COMPOSE_PROJECT_NAME="$(ci_project firewall)"
info "[firewall] Running integration test"
if bash testing/firewall/test.sh --skip-build 2>&1; then
record "firewall" 0
@@ -533,6 +652,7 @@ run_firewall() {
# Run a NAT scenario (cone, symmetric, lan)
run_nat() {
local scenario="$1"
export COMPOSE_PROJECT_NAME="$(ci_project nat)"
info "[nat-$scenario] Running NAT lab"
if bash testing/nat/scripts/nat-test.sh "$scenario" 2>&1; then
record "nat-$scenario" 0
@@ -546,6 +666,7 @@ run_nat() {
# (A→B publish/consume), Phase 2 (B→A reverse), and Phase 3 (malformed
# advert injected directly to the relay; consumer-liveness assertion).
run_nostr_publish_consume() {
export COMPOSE_PROJECT_NAME="$(ci_project nat)"
info "[nostr-publish-consume] Running Nostr publish/consume test"
if bash testing/nat/scripts/nostr-relay-test.sh 2>&1; then
record "nostr-publish-consume" 0
@@ -560,6 +681,7 @@ run_nostr_publish_consume() {
# kill. Asserts the daemon detects each fault, recovers from delay, and
# never panics.
run_stun_faults() {
export COMPOSE_PROJECT_NAME="$(ci_project nat)"
info "[stun-faults] Running STUN fault-injection test"
if bash testing/nat/scripts/stun-faults-test.sh 2>&1; then
record "stun-faults" 0
@@ -590,6 +712,7 @@ run_deb_install() {
# Run Tor SOCKS5 outbound test (live Tor network)
run_tor_socks5() {
export COMPOSE_PROJECT_NAME="$(ci_project tor-socks5)"
info "[tor-socks5] Running Tor SOCKS5 outbound test (live Tor)"
if bash testing/tor/socks5-outbound/scripts/tor-test.sh 2>&1; then
record "tor-socks5" 0
@@ -600,6 +723,7 @@ run_tor_socks5() {
# Run Tor directory-mode test (live Tor network)
run_tor_directory() {
export COMPOSE_PROJECT_NAME="$(ci_project tor-directory)"
info "[tor-directory] Running Tor directory-mode test (live Tor)"
if bash testing/tor/directory-mode/scripts/directory-test.sh 2>&1; then
record "tor-directory" 0
@@ -616,10 +740,20 @@ run_integration() {
info "Installing release binaries"
install_binaries testing/docker
# Build unified test image once (used by all harnesses)
info "Building fips-test Docker image"
docker build -t fips-test:latest testing/docker --quiet || { record "docker-build" 1; return; }
docker build -t fips-test-app:latest -f testing/docker/Dockerfile.app testing/docker --quiet || { record "docker-build-app" 1; return; }
# Build unified test image once (used by all harnesses). Tag per-run
# (fips-test:${run}) so a build killed mid-flight never wedges the next
# run's rebuild, and concurrent runs never clobber each other's image.
# Then retag :latest for the compose files / harness scripts that still
# reference fips-test:latest directly; the retag happens only after BOTH
# builds succeed, so :latest never points at a half-built image.
info "Building $CI_IMAGE_TEST Docker image"
docker build -t "$CI_IMAGE_TEST" --label "$CI_LABEL" testing/docker --quiet \
|| { record "docker-build" 1; return; }
docker build -t "$CI_IMAGE_APP" --label "$CI_LABEL" \
-f testing/docker/Dockerfile.app testing/docker --quiet \
|| { record "docker-build-app" 1; return; }
docker tag "$CI_IMAGE_TEST" fips-test:latest
docker tag "$CI_IMAGE_APP" fips-test-app:latest
# Single suite mode
if [[ -n "$ONLY_SUITE" ]]; then
@@ -673,6 +807,7 @@ run_integration() {
local pids=()
local suite_names=()
local running=0
local chaos_idx=0
for entry in "${CHAOS_SUITES[@]}"; do
# Parse: "display-name scenario [flags...]"
@@ -680,6 +815,15 @@ run_integration() {
local name="${parts[0]}"
local args=("${parts[@]:1}")
# Give each chaos child a unique, non-overlapping /24 in 10.30.x so
# parallel children never collide with each other, and so a chaos
# net can never swallow a fixed-subnet suite (sidecar/static/nat in
# 172.x). 10.30.x sits outside docker's default-address-pool range
# (172.17-31 / 192.168), so auto-assigned nets can't land on it
# either. Node IPs derive from this subnet inside the sim.
args+=("--subnet" "10.30.${chaos_idx}.0/24")
chaos_idx=$((chaos_idx + 1))
# Throttle: wait for a slot
while [[ $running -ge $PARALLEL_JOBS ]]; do
wait -n -p done_pid 2>/dev/null || true
@@ -693,6 +837,7 @@ run_integration() {
run_chaos "$name" "${args[@]}" >"$logfile" 2>&1
) &
pids+=($!)
CI_CHAOS_PIDS+=("$!")
suite_names+=("$name:$logfile")
running=$((running + 1))
done
@@ -715,6 +860,9 @@ run_integration() {
fi
rm -f "$logfile"
done
# All chaos children have been waited on; clear so a later signal does
# not try to kill already-reaped PIDs.
CI_CHAOS_PIDS=()
fi
# Sidecar
+1
View File
@@ -65,6 +65,7 @@ start_systemd_container_with_tun() {
local name="$1" image="$2"
cleanup_container "$name"
docker run -d --name "$name" \
--label com.corganlabs.fips-ci=1 \
--privileged \
--cgroupns=host \
--device /dev/net/tun \
+2
View File
@@ -67,6 +67,7 @@ start_systemd_container() {
local name="$1" image="$2"
cleanup_container "$name"
docker run -d --name "$name" \
--label com.corganlabs.fips-ci=1 \
--privileged \
--cgroupns=host \
-v /sys/fs/cgroup:/sys/fs/cgroup:rw \
@@ -79,6 +80,7 @@ start_systemd_container_with_tun() {
local name="$1" image="$2"
cleanup_container "$name"
docker run -d --name "$name" \
--label com.corganlabs.fips-ci=1 \
--privileged \
--cgroupns=host \
--device /dev/net/tun \
+2
View File
@@ -1,6 +1,8 @@
networks:
fw-net:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: 172.32.0.0/24
+30 -14
View File
@@ -162,6 +162,13 @@ REKEY_SETTLE=12 # FSP-cutover settle budget (Phase 6)
# strict assertion sweep runs. A genuinely stuck pair still fails — the
# poll times out and the recording sweep captures it.
POST_REKEY_TIMEOUT=45
# Progress-aware stall budget for the convergence detector
# (wait_for_full_baseline → wait_until_connected). If no additional pair
# becomes reachable for this long while more than the near-converged
# slack of pairs is still down, the detector gives up early instead of
# burning the whole convergence/post-rekey window; any progress resets
# the clock, so a slow-but-converging mesh under netem keeps polling.
RECONVERGE_STALL=15
LOG_POLL_INTERVAL=2
# Data-plane continuity stream (control-differential). Streams run a
@@ -434,25 +441,27 @@ ping_all_pairs() {
done
}
# Convergence detector probe: one full all-pairs ping sweep in the
# "convergence" context (which ping_all_pairs deliberately does NOT
# record as a failure), setting PASSED/FAILED for wait_until_connected.
_baseline_probe() {
ping_all_pairs quiet 1 "convergence"
}
# Poll until every directed pair pings clean, or until timeout. This is a
# convergence DETECTOR — used at establishment (Phase 1) and after each
# rekey (Phases 3/5). It pings with the "convergence" context, which
# ping_all_pairs deliberately does not record as a failure; the caller
# runs a separate strict assertion sweep afterwards.
#
# Delegates to the shared progress-aware wait_until_connected so the
# deadline extends while more pairs are still coming up and gives up fast
# on a genuine stall, instead of the prior fixed-deadline poll that could
# false-time-out under heavy CI contention even while still converging.
# Returns 0 once every pair is reachable, 1 on stall/timeout.
wait_for_full_baseline() {
local timeout="$1"
local start=$SECONDS
local best_passed=0 best_failed="$NUM_DIRECTED"
while (( SECONDS - start < timeout )); do
ping_all_pairs quiet 1 "convergence"
if [ "$PASSED" -gt "$best_passed" ]; then
best_passed="$PASSED"; best_failed="$FAILED"
fi
[ "$FAILED" -eq 0 ] && return 0
sleep 1
done
PASSED="$best_passed"; FAILED="$best_failed"
return 1
wait_until_connected _baseline_probe "$timeout" "$RECONVERGE_STALL"
}
phase_result() {
@@ -759,8 +768,15 @@ echo ""
# ── Phase 4: second rekey cycle ──────────────────────────────────────
echo "Phase 4: Second rekey cycle (waiting ${SECOND_REKEY_WAIT}s)"
sleep "$SECOND_REKEY_WAIT"
echo "Phase 4: Second rekey cycle (waiting up to ${SECOND_REKEY_WAIT}s for the next cutover)"
# Poll for the next FMP cutover beyond what Phases 2/3 already saw, using
# the same pre/post cutover-count delta convention as the control window
# (Phase 1b), instead of a blind sleep. Bounded by SECOND_REKEY_WAIT so a
# stalled rekey falls through to the strict Phase 5/6 assertions.
fmp_cutovers_before="$(count_log_pattern 'Rekey cutover complete \(initiator\), K-bit flipped')"
wait_for_log_pattern_count \
"Rekey cutover complete \(initiator\), K-bit flipped" \
"$((fmp_cutovers_before + 1))" "$SECOND_REKEY_WAIT" || true
echo ""
echo "Phase 5: Post-second-rekey connectivity (reconverge within ${POST_REKEY_TIMEOUT}s)"
+4
View File
@@ -1,11 +1,15 @@
networks:
wan:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: 172.31.254.0/24
shared-lan:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: 172.31.10.0/24
+1 -1
View File
@@ -202,7 +202,7 @@ dump_stun_udp_probe() {
local capture_file
capture_file="$(mktemp)"
docker run --rm --net=container:fips-nat-stun --cap-add NET_ADMIN --cap-add NET_RAW \
docker run --rm --label com.corganlabs.fips-ci=1 --net=container:fips-nat-stun --cap-add NET_ADMIN --cap-add NET_RAW \
--entrypoint sh "$helper_image" \
-lc "timeout 8 tcpdump -ni any 'udp and not port 53' -c 80" \
>"$capture_file" 2>&1 &
+1
View File
@@ -57,6 +57,7 @@ run_host_ip() {
local image="$1"
shift
docker run --rm \
--label com.corganlabs.fips-ci=1 \
--privileged \
--net=host \
--pid=host \
+10
View File
@@ -2,6 +2,12 @@ networks:
fips-net:
name: ${FIPS_NETWORK:-fips-sidecar-net}
driver: bridge
# Reap label: this harness uses fixed -p sidecar-{a,b,c} project names
# (the test execs containers by those derived names), so its resources are
# NOT covered by ci-local's fipsci_ project prefix. Label them directly so
# ci-cleanup.sh can still reap them after a preemption/SIGKILL.
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: ${FIPS_SUBNET:-172.20.1.0/24}
@@ -10,6 +16,8 @@ services:
fips:
image: fips-test:latest
hostname: fips-sidecar
labels:
- "com.corganlabs.fips-ci=1"
cap_add:
- NET_ADMIN
devices:
@@ -33,6 +41,8 @@ services:
app:
image: fips-test-app:latest
labels:
- "com.corganlabs.fips-ci=1"
network_mode: "service:fips"
depends_on:
- fips
+6 -2
View File
@@ -1,11 +1,15 @@
networks:
fips-net:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
ipam:
config:
- subnet: 172.20.0.0/24
gateway-lan:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
enable_ipv6: true
ipam:
config:
@@ -534,7 +538,7 @@ services:
ipv4_address: 172.20.0.12
gw-client:
image: fips-test-app:latest
image: ${FIPS_TEST_APP_IMAGE:-fips-test-app:latest}
profiles: ["gateway"]
container_name: fips-gw-client
hostname: gw-client
@@ -556,7 +560,7 @@ services:
# Same image and gateway-lan attachment as
# gw-client; the gateway must allocate a distinct virtual IP for it.
gw-client-2:
image: fips-test-app:latest
image: ${FIPS_TEST_APP_IMAGE:-fips-test-app:latest}
profiles: ["gateway"]
container_name: fips-gw-client-2
hostname: gw-client-2
+1 -1
View File
@@ -149,7 +149,7 @@ HELPER_IMAGE=$(docker inspect -f '{{.Config.Image}}' fips-node-$CAP_NODE 2>/dev/
LOAD_PID=$!
# Foreground: tcpdump capture for CAPTURE_SECS
docker run --rm --net=container:fips-node-$CAP_NODE \
docker run --rm --label com.corganlabs.fips-ci=1 --net=container:fips-node-$CAP_NODE \
--cap-add NET_ADMIN --cap-add NET_RAW \
--entrypoint sh "$HELPER_IMAGE" \
-c "timeout $CAPTURE_SECS tcpdump -nn -i any 'udp port 2121' -l 2>&1 || true" \
+38 -9
View File
@@ -158,9 +158,20 @@ REKEY_SETTLE=12 # > DRAIN_WINDOW_SECS (10) so post-rekey samples are off
# fully converged. Keep this bounded to preserve a meaningful scheduling check
# while still allowing for log visibility at the timeout edge.
FIRST_REKEY_TIMEOUT=$((REKEY_AFTER_SECS + 15))
SECOND_REKEY_WAIT=40 # wait for second cycle
SECOND_REKEY_WAIT=40 # upper bound for observing the second rekey cutover
LOG_EVENT_POLL_INTERVAL=1
# Post-rekey data-plane reconvergence is polled, not fixed-slept.
# wait_until_connected drives the same all-pairs ping sweep the strict
# per-phase assert depends on: it fails fast when reachability stops
# improving (no new reachable pair for RECONVERGE_STALL seconds while
# more than the near-converged slack of pairs is still down) and extends
# its deadline up to RECONVERGE_TIMEOUT only while pairs keep coming up,
# so a slow-but-converging post-rekey mesh under CI load passes instead
# of false-failing on a fixed settle window.
RECONVERGE_TIMEOUT=45
RECONVERGE_STALL=10
TIMEOUT=5
CONVERGENCE_PING_TIMEOUT=1
# Strict-ping retry policy for the per-phase ping_all asserts. Under 1%
@@ -392,20 +403,38 @@ assert_min_count "Rekey cutover complete (initiator), K-bit flipped" 1 \
phase_result "FMP rekey events"
echo ""
# Verify connectivity after first rekey (strict — no failures allowed)
echo "Phase 3: Post-rekey connectivity (settling ${REKEY_SETTLE}s)"
sleep "$REKEY_SETTLE"
# Verify connectivity after first rekey (strict — no failures allowed).
# Wait for the data plane to reconverge with a progress-aware deadline
# (the same all-pairs ping sweep the strict assert below depends on)
# instead of a fixed settle: fail fast on a genuinely stuck pair, extend
# only while reachability is still climbing. The strict ping_all is the
# actual assertion, run only after the reconverge poll.
echo "Phase 3: Post-rekey connectivity (reconverging up to ${RECONVERGE_TIMEOUT}s)"
wait_until_connected _baseline_ping "$RECONVERGE_TIMEOUT" "$RECONVERGE_STALL" || true
ping_all "" "$TIMEOUT" "$MAX_PING_ATTEMPTS"
phase_result "Post-first-rekey (all 20 pairs)"
echo ""
# ── Phase 4: Wait for second rekey cycle ──────────────────────────────
echo "Phase 4: Second rekey cycle (waiting ${SECOND_REKEY_WAIT}s)"
sleep "$SECOND_REKEY_WAIT"
# Poll for the next FMP rekey cutover instead of blind-sleeping: capture
# the cutover count reached so far, then wait until at least one more
# cutover lands — that increment is the second rekey cycle firing.
# Bounded by SECOND_REKEY_WAIT so a stalled rekey still falls through to
# the strict Phase 5/6 assertions rather than hanging.
echo "Phase 4: Second rekey cycle (waiting up to ${SECOND_REKEY_WAIT}s for the next cutover)"
fmp_cutovers_before=$(count_log_pattern "Rekey cutover complete (initiator), K-bit flipped")
wait_for_log_pattern_count \
"Rekey cutover complete (initiator), K-bit flipped" \
"$((fmp_cutovers_before + 1))" "$SECOND_REKEY_WAIT" || true
# Verify connectivity after second rekey (back-to-back)
echo "Phase 5: Post-second-rekey connectivity (settling ${REKEY_SETTLE}s)"
sleep "$REKEY_SETTLE"
# Verify connectivity after second rekey (back-to-back). This is the
# site of the recurring post-second-rekey straggler-pair flake: wait for
# the data plane to reconverge with a progress-aware deadline (the same
# all-pairs ping sweep the strict assert below depends on) rather than a
# fixed settle window — fail fast if reachability stops improving, extend
# only while pairs are still coming up.
echo "Phase 5: Post-second-rekey connectivity (reconverging up to ${RECONVERGE_TIMEOUT}s)"
wait_until_connected _baseline_ping "$RECONVERGE_TIMEOUT" "$RECONVERGE_STALL" || true
ping_all "" "$TIMEOUT" "$MAX_PING_ATTEMPTS"
phase_result "Post-second-rekey (all 20 pairs)"
echo ""
@@ -11,6 +11,8 @@
networks:
dir-test:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
services:
fips-a:
@@ -10,6 +10,8 @@
networks:
tor-test:
driver: bridge
labels:
- "com.corganlabs.fips-ci=1"
services:
tor-daemon: