#![cfg(all(not(target_arch = "wasm32"), openrtc_unpublished_ble))]
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use openrtc::application_crypto::APPLICATION_KEY_BYTES;
use openrtc::application_crypto_streams::{wrap_peer_streams, PeerRecvStream, PeerSendStream};
use openrtc::client::{
BleConfig, BleHardwareHarnessDebugSnapshot, Client, IrohPathKind, TransportConfig,
};
use openrtc::native_node::{IncomingStream, IncomingStreamType};
const DEFAULT_PROJECT_ID: &str = "openrtc-ble-hardware-harness";
const DEFAULT_API_KEY: &str = "pk_test_openrtc_ble_hardware_harness";
#[derive(Parser, Debug)]
#[command(
name = "ble-hardware-harness",
about = "Private OpenRTC BLE hardware E2E harness"
)]
struct Args {
#[command(subcommand)]
command: Command,
#[arg(long, global = true)]
app_key_hex: Option<String>,
}
#[derive(Subcommand, Debug)]
enum Command {
Listen {
#[arg(long, default_value = ".openrtc-ble-harness-listener.key")]
secret_file: PathBuf,
},
Dial {
#[arg(long)]
peer: String,
#[arg(long, default_value = ".openrtc-ble-harness-dialer.key")]
secret_file: PathBuf,
#[arg(long, default_value_t = 1)]
reconnects: u32,
#[arg(long, default_value_t = 256 * 1024)]
payload_bytes: usize,
#[arg(long, default_value_t = 180_000)]
connect_timeout_ms: u64,
#[arg(long, default_value_t = 2_000)]
reconnect_pause_ms: u64,
#[arg(long, default_value_t = false)]
wait_between_cycles: bool,
},
}
#[tokio::main]
async fn main() -> Result<()> {
openrtc::ensure_default_rustls_provider();
let args = Args::parse();
let app_key = parse_app_key(args.app_key_hex.as_deref())?;
match args.command {
Command::Listen { secret_file } => run_listener(secret_file, app_key).await,
Command::Dial {
peer,
secret_file,
reconnects,
payload_bytes,
connect_timeout_ms,
reconnect_pause_ms,
wait_between_cycles,
} => {
run_dialer(
peer,
secret_file,
app_key,
reconnects,
payload_bytes,
Duration::from_millis(connect_timeout_ms),
Duration::from_millis(reconnect_pause_ms),
wait_between_cycles,
)
.await
}
}
}
async fn run_listener(secret_file: PathBuf, app_key: [u8; APPLICATION_KEY_BYTES]) -> Result<()> {
let secret = load_or_create_secret(&secret_file)?;
let client = build_client();
let node_id = client
.init_iroh_ble_hardware_harness(Some(secret.to_bytes().to_vec()), Vec::new())
.await?;
let incoming = client.incoming_streams().await?;
println!("BLE listener ready.");
println!("Endpoint ID:");
println!(" {node_id}");
println!(
"BLE listener debug: {}",
format_ble_debug(&client, Some(node_id.as_str())).await
);
println!();
println!("On the nearby device run:");
println!(
" cargo run --bin ble-hardware-harness --features transport-ble -- dial --peer {node_id}"
);
println!();
println!("Waiting for encrypted BLE streams. Press Ctrl-C to stop.");
let listener = tokio::spawn(async move {
loop {
match incoming.recv().await {
Ok(stream) => {
let key = app_key;
tokio::spawn(async move {
if let Err(error) = handle_incoming_stream(stream, key).await {
eprintln!("[ble-hardware-harness] incoming stream failed: {error:#}");
}
});
}
Err(error) => {
eprintln!("[ble-hardware-harness] incoming stream receiver closed: {error}");
break;
}
}
}
});
tokio::signal::ctrl_c().await?;
listener.abort();
client.get_endpoint().await?.close().await;
Ok(())
}
async fn handle_incoming_stream(
stream: IncomingStream,
app_key: [u8; APPLICATION_KEY_BYTES],
) -> Result<()> {
let IncomingStreamType::Bi(send, recv) = stream.stream else {
println!(
"[ble-hardware-harness] ignoring non-bi stream from {}",
stream.endpoint_id
);
return Ok(());
};
let (mut send, mut recv) = wrap_peer_streams(Some(app_key), send, recv)
.context("failed to wrap incoming stream with application crypto")?;
ensure_encrypted_streams(&send, &recv)?;
let payload = read_peer_stream_to_end(&mut recv, 32 * 1024 * 1024).await?;
println!(
"[ble-hardware-harness] received encrypted payload from {}: {} bytes",
stream.endpoint_id,
payload.len()
);
send.write_all(&payload).await?;
send.finish()?;
Ok(())
}
async fn run_dialer(
peer: String,
secret_file: PathBuf,
app_key: [u8; APPLICATION_KEY_BYTES],
reconnects: u32,
payload_bytes: usize,
connect_timeout: Duration,
reconnect_pause: Duration,
wait_between_cycles: bool,
) -> Result<()> {
let remote = peer
.parse::<iroh::EndpointId>()
.with_context(|| format!("invalid peer endpoint id: {peer}"))?;
let secret = load_or_create_secret(&secret_file)?;
let cycles = reconnects.saturating_add(1);
for cycle in 0..cycles {
println!();
println!(
"=== BLE hardware cycle {}/{}: starting OpenRTC client ===",
cycle + 1,
cycles
);
let client = build_client();
let local_node_id = client
.init_iroh_ble_hardware_harness(Some(secret.to_bytes().to_vec()), Vec::new())
.await?;
println!("Local endpoint id: {local_node_id}");
println!(
"Initial BLE debug: {}",
format_ble_debug(&client, Some(&remote.to_string())).await
);
println!("Dialing {remote} with relays and IP transports disabled...");
connect_ble_only(&client, remote, connect_timeout).await?;
wait_for_ble_path(&client, &remote.to_string(), Duration::from_secs(20)).await?;
let connection_id = client
.register_ble_hardware_harness_peer(&remote.to_string(), app_key)
.await?;
println!("Registered encrypted harness peer: {connection_id}");
run_encrypted_round_trip(
&client,
&remote.to_string(),
b"openrtc ble hardware harness message".to_vec(),
"message",
)
.await?;
run_encrypted_round_trip(
&client,
&remote.to_string(),
generated_payload(payload_bytes),
"file-payload",
)
.await?;
let kind = client.iroh_path_kind(&remote.to_string()).await;
anyhow::ensure!(
kind == IrohPathKind::Ble,
"selected path changed away from BLE after transfer: {kind:?}"
);
println!(
"Cycle {} passed: selected path is BLE and encrypted transfers echoed.",
cycle + 1
);
let _ = client.disconnect(remote).await;
client.get_endpoint().await?.close().await;
if cycle + 1 < cycles {
if wait_between_cycles {
println!(
"Cycle paused. Restart Bluetooth or the listener app now, then press Enter to reconnect..."
);
let mut line = String::new();
std::io::stdin().read_line(&mut line)?;
} else {
println!(
"Simulating app restart; waiting {} ms before reconnect...",
reconnect_pause.as_millis()
);
tokio::time::sleep(reconnect_pause).await;
}
}
}
println!();
println!("BLE hardware harness passed all {cycles} cycle(s).");
Ok(())
}
async fn connect_ble_only(
client: &Client,
remote: iroh::EndpointId,
timeout: Duration,
) -> Result<()> {
let started = Instant::now();
let remote_label = remote.to_string();
let service_uuid = client
.prime_ble_hardware_harness_target_scan(&remote_label)
.await?;
println!("Targeted BLE scan started for service UUID {service_uuid}.");
wait_for_ble_discovery(client, &remote_label, timeout.min(Duration::from_secs(60))).await?;
let progress_client = client.clone();
let progress_remote_label = remote_label.clone();
let progress = tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(5)).await;
println!(
"BLE discovery/dial waiting elapsed_ms={} debug {}",
started.elapsed().as_millis(),
format_ble_debug(&progress_client, Some(&progress_remote_label)).await
);
}
});
let mut attempt = 0u32;
let mut last_error = None;
loop {
let elapsed = started.elapsed();
if elapsed >= timeout {
progress.abort();
let debug = format_ble_debug(client, Some(&remote_label)).await;
let last_error = last_error
.as_deref()
.unwrap_or("BLE iroh connect did not complete before the overall timeout");
println!(
"BLE discovery/dial failed after {} ms: {last_error}; debug {debug}",
elapsed.as_millis()
);
anyhow::bail!(
"timed out after {} ms waiting for BLE discovery/dial to {}; last_error={}; debug {}",
timeout.as_millis(),
remote_label,
last_error,
debug
);
}
attempt += 1;
let remaining = timeout.saturating_sub(elapsed);
let attempt_timeout = remaining.min(Duration::from_secs(30));
println!(
"BLE iroh connect attempt {attempt} starting window_ms={}.",
attempt_timeout.as_millis()
);
match client
.ensure_connected_with_timeout(remote, attempt_timeout)
.await
{
Ok(()) => {
progress.abort();
println!(
"Connected over BLE-only iroh path after {} ms.",
started.elapsed().as_millis()
);
return Ok(());
}
Err(error) => {
let snapshot = client.ble_hardware_harness_debug(Some(&remote_label)).await;
let debug = format_ble_debug_snapshot(&snapshot);
let peer_state = summarize_ble_peer_state(&snapshot);
last_error = Some(error.to_string());
if !snapshot.target_seen {
progress.abort();
println!(
"BLE discovery/dial lost target after {} ms: {error}; debug {debug}",
started.elapsed().as_millis()
);
anyhow::bail!(
"BLE target disappeared while connecting to {}; last_error={}; debug {}",
remote_label,
error,
debug
);
}
println!(
"BLE iroh connect attempt {attempt} did not finish yet after {} ms: {error}; peer_state={peer_state}; continuing",
started.elapsed().as_millis()
);
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
}
}
fn summarize_ble_peer_state(snapshot: &BleHardwareHarnessDebugSnapshot) -> String {
if snapshot.peers.is_empty() {
return format!(
"none target_seen={} scan_hints={} reservations={} routable={} pipes={}",
snapshot.target_seen,
snapshot.route_scan_hints,
snapshot.route_reservations,
snapshot.route_routable,
snapshot.route_pipes
);
}
snapshot
.peers
.iter()
.map(|peer| {
let detail = peer.phase_detail.as_deref().unwrap_or("no-detail");
format!(
"{}:{} failures={} path={}",
peer.phase,
detail,
peer.consecutive_failures,
peer.connect_path.as_deref().unwrap_or("none")
)
})
.collect::<Vec<_>>()
.join("; ")
}
fn format_ble_debug_snapshot(snapshot: &BleHardwareHarnessDebugSnapshot) -> String {
serde_json::to_string(snapshot).unwrap_or_else(|error| format!("{{\"error\":\"{error}\"}}"))
}
async fn wait_for_ble_discovery(client: &Client, remote: &str, timeout: Duration) -> Result<()> {
let started = Instant::now();
loop {
let snapshot = client.ble_hardware_harness_debug(Some(remote)).await;
let debug = format_ble_debug_snapshot(&snapshot);
if snapshot.target_seen {
println!(
"BLE advertisement discovered after {} ms; debug {debug}",
started.elapsed().as_millis()
);
return Ok(());
}
if started.elapsed() >= timeout {
println!(
"BLE advertisement discovery failed after {} ms; debug {debug}",
started.elapsed().as_millis()
);
anyhow::bail!(
"BLE advertisement for {remote} was not discovered within {} ms; debug {debug}",
timeout.as_millis(),
)
}
println!(
"BLE advertisement waiting elapsed_ms={} debug {debug}",
started.elapsed().as_millis()
);
tokio::time::sleep(Duration::from_secs(5)).await;
}
}
async fn wait_for_ble_path(client: &Client, remote: &str, timeout: Duration) -> Result<()> {
let started = Instant::now();
loop {
let kind = client.iroh_path_kind(remote).await;
match kind {
IrohPathKind::Ble => {
println!("Selected iroh path reports BLE.");
return Ok(());
}
IrohPathKind::DirectLan | IrohPathKind::DirectQuic | IrohPathKind::Relay => {
anyhow::bail!("expected BLE selected path, got {kind:?}");
}
IrohPathKind::Unknown => {
if started.elapsed() >= timeout {
anyhow::bail!("timed out waiting for BLE selected path");
}
tokio::time::sleep(Duration::from_millis(250)).await;
}
}
}
}
async fn run_encrypted_round_trip(
client: &Client,
remote: &str,
payload: Vec<u8>,
label: &str,
) -> Result<()> {
let (_connection_id, _remote_node_id, mut send, mut recv) =
client.open_peer_bi(remote, Some(10_000)).await?;
ensure_encrypted_streams(&send, &recv)?;
send.write_all(&payload).await?;
send.finish()?;
let echoed = read_peer_stream_to_end(&mut recv, payload.len().saturating_add(4096)).await?;
anyhow::ensure!(
echoed == payload,
"{label} echo mismatch: sent {} bytes, received {} bytes",
payload.len(),
echoed.len()
);
println!(
"Encrypted {label} round-trip passed: {} bytes.",
payload.len()
);
Ok(())
}
fn ensure_encrypted_streams(send: &PeerSendStream, recv: &PeerRecvStream) -> Result<()> {
anyhow::ensure!(
send.is_encrypted(),
"send stream is not application-encrypted"
);
anyhow::ensure!(
recv.is_encrypted(),
"recv stream is not application-encrypted"
);
Ok(())
}
async fn read_peer_stream_to_end(stream: &mut PeerRecvStream, max_bytes: usize) -> Result<Vec<u8>> {
let mut out = Vec::new();
let mut chunk = vec![0u8; 16 * 1024];
loop {
let read = stream.read(&mut chunk).await?;
if read == 0 {
break;
}
out.extend_from_slice(&chunk[..read]);
anyhow::ensure!(
out.len() <= max_bytes,
"peer stream exceeded max size: {} > {}",
out.len(),
max_bytes
);
}
Ok(out)
}
fn generated_payload(bytes: usize) -> Vec<u8> {
(0..bytes)
.map(|index| {
let x = index as u64;
(x.wrapping_mul(31).wrapping_add(x >> 3) & 0xff) as u8
})
.collect()
}
async fn format_ble_debug(client: &Client, target_node_id: Option<&str>) -> String {
let snapshot = client.ble_hardware_harness_debug(target_node_id).await;
serde_json::to_string(&snapshot).unwrap_or_else(|error| {
format!("{{\"error\":\"failed to encode debug snapshot: {error}\"}}")
})
}
fn build_client() -> Client {
Client::builder(
DEFAULT_PROJECT_ID.to_string(),
DEFAULT_API_KEY.to_string(),
Box::new(|| None),
)
.transport_config(TransportConfig {
ble: Some(BleConfig {
enabled: true,
..BleConfig::default()
}),
..TransportConfig::default()
})
.build()
}
fn parse_app_key(raw: Option<&str>) -> Result<[u8; APPLICATION_KEY_BYTES]> {
let Some(raw) = raw else {
println!(
"Using deterministic lab application key. Pass --app-key-hex for a unique private test key."
);
return Ok([0x42; APPLICATION_KEY_BYTES]);
};
let bytes = hex::decode(raw.trim()).context("app key must be hex")?;
anyhow::ensure!(
bytes.len() == APPLICATION_KEY_BYTES,
"app key must be {APPLICATION_KEY_BYTES} bytes, got {}",
bytes.len()
);
let mut key = [0u8; APPLICATION_KEY_BYTES];
key.copy_from_slice(&bytes);
Ok(key)
}
fn load_or_create_secret(path: &Path) -> Result<iroh::SecretKey> {
if path.exists() {
let raw = std::fs::read_to_string(path)
.with_context(|| format!("failed to read secret file {}", path.display()))?;
return iroh::SecretKey::from_str(raw.trim())
.with_context(|| format!("failed to parse secret file {}", path.display()));
}
let secret = iroh::SecretKey::generate();
if let Some(parent) = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
std::fs::write(path, hex::encode(secret.to_bytes()))
.with_context(|| format!("failed to write secret file {}", path.display()))?;
println!("Generated persistent endpoint secret at {}", path.display());
Ok(secret)
}