use anyhow::{anyhow, bail, Context, Result};
use base64::Engine as _;
use openrtc::client::{StateSnapshot, TransportConfig, WebRTCConfig};
use openrtc::native::{load_or_create_endpoint_key, ControlPlane, DeviceSigner, Features, FileSigner, RoomAuthorityOptions};
use openrtc::Client;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet};
#[cfg(test)]
use std::{fs, path::Path};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader};
use zeroize::Zeroize;
#[cfg(feature = "testing-endpoints")]
#[path = "openrtc_node_host/test_lane.rs"]
mod test_lane;
#[cfg(all(test, unix))]
use std::os::unix::fs::{DirBuilderExt as _, PermissionsExt as _};
const PROTOCOL_VERSION: u8 = 1;
const MAX_CONFIG_BYTES: usize = 24 * 1024;
const MAX_COMMAND_BYTES: usize = 96 * 1024;
const MAX_APPLICATION_PAYLOAD_BYTES: usize = 64 * 1024;
const MAX_IDENTIFIER_BYTES: usize = 160;
const MAX_ERROR_BYTES: usize = 512;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct HostConfig {
protocol_version: u8,
api_key: String,
secret_key: String,
service_id: String,
generation: u64,
shard_ids: Vec<String>,
device_id: String,
room_id: String,
device_name: String,
state_directory: PathBuf,
#[serde(default = "default_max_peers")]
max_peers: u32,
#[serde(default = "default_true")]
iroh_relay: bool,
}
impl HostConfig {
fn parse(line: &str) -> Result<Self> {
let mut config: Self = serde_json::from_str(line).context("decode node host config")?;
if config.protocol_version != PROTOCOL_VERSION {
bail!("unsupported node host protocol version");
}
config.api_key = openrtc::validate_api_key(&config.api_key)?.to_string();
let secret_suffix = config
.secret_key
.strip_prefix("sk_live_")
.or_else(|| config.secret_key.strip_prefix("sk_test_"));
if config.secret_key.len() > 128
|| !secret_suffix.is_some_and(|suffix| {
!suffix.is_empty()
&& suffix
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
})
{
bail!("node host secret key is invalid");
}
config.service_id = bounded_id("serviceId", config.service_id, 80)?;
config.device_id = bounded_id("deviceId", config.device_id, MAX_IDENTIFIER_BYTES)?;
config.room_id = bounded_id("roomId", config.room_id, MAX_IDENTIFIER_BYTES)?;
config.device_name = bounded_text("deviceName", config.device_name, 80)?;
if config.generation == 0 {
bail!("node host generation is invalid");
}
if config.shard_ids.is_empty() || config.shard_ids.len() > 64 {
bail!("node host shardIds must contain 1..64 identifiers");
}
let mut shards = config
.shard_ids
.into_iter()
.map(|value| bounded_id("shardId", value, 80))
.collect::<Result<Vec<_>>>()?;
let original_len = shards.len();
shards.sort();
shards.dedup();
if shards.len() != original_len {
bail!("node host shardIds must be unique");
}
config.shard_ids = shards;
if !(1..=50).contains(&config.max_peers) {
bail!("node host maxPeers must be within 1..50 room members");
}
if !config.state_directory.is_absolute() {
bail!("node host stateDirectory must be absolute");
}
Ok(config)
}
}
fn default_max_peers() -> u32 {
8
}
fn default_true() -> bool {
true
}
fn bounded_id(name: &str, value: String, maximum: usize) -> Result<String> {
let value = value.trim().to_string();
if value.is_empty()
|| value.len() > maximum
|| !value.bytes().all(|byte| {
byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'.' | b':' | b'@' | b'-')
})
{
bail!("node host {name} is invalid");
}
Ok(value)
}
fn bounded_text(name: &str, value: String, maximum: usize) -> Result<String> {
let value = value.trim().to_string();
if value.is_empty()
|| value.len() > maximum
|| value.chars().any(|character| character.is_control())
{
bail!("node host {name} is invalid");
}
Ok(value)
}
#[derive(Debug, Deserialize)]
#[serde(
tag = "type",
rename_all = "kebab-case",
rename_all_fields = "camelCase",
deny_unknown_fields
)]
enum Command {
PublishAssignment {
id: String,
revision: u64,
subject_device_id: String,
shard_id: String,
#[serde(default)]
priority_device_ids: Vec<String>,
#[serde(default)]
relevant_entity_ids: Vec<String>,
#[serde(default = "default_assignment_ttl_ms")]
ttl_ms: u64,
},
Send {
id: String,
connection_id: String,
payload_base64: String,
},
Shutdown {
id: String,
},
}
impl Command {
fn id(&self) -> &str {
match self {
Self::PublishAssignment { id, .. } | Self::Send { id, .. } | Self::Shutdown { id } => {
id
}
}
}
fn validate(self) -> Result<Self> {
bounded_id("request id", self.id().to_string(), 80)?;
match &self {
Self::PublishAssignment {
revision,
subject_device_id,
shard_id,
priority_device_ids,
relevant_entity_ids,
ttl_ms,
..
} => {
if *revision == 0 || *ttl_ms == 0 || *ttl_ms > 120_000 {
bail!("node host authority assignment revision or TTL is invalid");
}
bounded_id(
"subjectDeviceId",
subject_device_id.clone(),
MAX_IDENTIFIER_BYTES,
)?;
bounded_id("shardId", shard_id.clone(), 80)?;
if priority_device_ids.len() > 8 || relevant_entity_ids.len() > 32 {
bail!("node host authority assignment exceeds its identifier bound");
}
for value in priority_device_ids {
bounded_id("priorityDeviceId", value.clone(), MAX_IDENTIFIER_BYTES)?;
}
for value in relevant_entity_ids {
bounded_id("relevantEntityId", value.clone(), MAX_IDENTIFIER_BYTES)?;
}
}
Self::Send {
connection_id,
payload_base64,
..
} => {
bounded_id("connectionId", connection_id.clone(), MAX_IDENTIFIER_BYTES)?;
if payload_base64.len() > MAX_APPLICATION_PAYLOAD_BYTES.saturating_mul(2) {
bail!("node host application payload exceeds its byte limit");
}
}
Self::Shutdown { .. } => {}
}
Ok(self)
}
}
fn default_assignment_ttl_ms() -> u64 {
60_000
}
#[derive(Serialize)]
#[serde(
tag = "type",
rename_all = "kebab-case",
rename_all_fields = "camelCase"
)]
enum OutputEvent<'a> {
Failure {
error: String,
#[serde(skip_serializing_if = "Option::is_none")]
service_error: Option<openrtc::service_errors::ServiceError>,
},
Ready {
protocol_version: u8,
device_id: &'a str,
room_id: &'a str,
node_id: &'a str,
#[cfg(feature = "testing-endpoints")]
#[serde(skip_serializing_if = "Option::is_none")]
test_lane: Option<&'a test_lane::TestLane>,
},
Response {
id: &'a str,
ok: bool,
#[serde(skip_serializing_if = "Option::is_none")]
value: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
service_error: Option<openrtc::service_errors::ServiceError>,
},
ConnectionState {
snapshot: &'a StateSnapshot,
transport_stable_id: Option<&'a str>,
transport_generation: &'a str,
route_generation: &'a str,
},
PeerMessage {
connection_id: &'a str,
payload_base64: &'a str,
transport_stable_id: &'a str,
transport_generation: &'a str,
route_generation: &'a str,
},
ServiceError {
avenue: OutputServiceAvenue<'a>,
runtime_instance_id: &'a str,
service_error: &'a openrtc::service_errors::ServiceError,
},
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct OutputServiceAvenue<'a> {
kind: &'a str,
id: &'a str,
}
async fn emit<W: AsyncWrite + Unpin>(writer: &mut W, event: &OutputEvent<'_>) -> Result<()> {
let mut line = serde_json::to_vec(event)?;
line.push(b'\n');
writer.write_all(&line).await?;
writer.flush().await?;
Ok(())
}
async fn read_bounded_line<R: AsyncBufRead + Unpin>(
reader: &mut R,
maximum: usize,
) -> Result<Option<String>> {
let mut line = Vec::new();
loop {
let available = reader.fill_buf().await?;
if available.is_empty() {
if line.is_empty() {
return Ok(None);
}
break;
}
let consumed = available
.iter()
.position(|byte| *byte == b'\n')
.map(|position| position + 1)
.unwrap_or(available.len());
let payload_length = if available.get(consumed.saturating_sub(1)) == Some(&b'\n') {
consumed - 1
} else {
consumed
};
if line.len().saturating_add(payload_length) > maximum {
bail!("node host input line exceeds its byte limit");
}
line.extend_from_slice(&available[..payload_length]);
let reached_newline = payload_length < consumed;
reader.consume(consumed);
if reached_newline {
break;
}
}
if line.last() == Some(&b'\r') {
line.pop();
}
String::from_utf8(line)
.map(Some)
.context("node host input must be UTF-8")
}
async fn read_command<R: AsyncBufRead + Unpin>(reader: &mut R) -> Result<Option<Command>> {
loop {
let Some(line) = read_bounded_line(reader, MAX_COMMAND_BYTES).await? else {
return Ok(None);
};
if line.trim().is_empty() {
continue;
}
return serde_json::from_str::<Command>(&line)
.context("decode node host command")?
.validate()
.map(Some);
}
}
fn bounded_error(error: &anyhow::Error) -> String {
let raw = error.to_string();
let mut redacted = String::with_capacity(raw.len());
let mut offset = 0;
while offset < raw.len() {
let remaining = &raw[offset..];
if remaining.starts_with("sk_live_") || remaining.starts_with("sk_test_") {
redacted.push_str("[REDACTED]");
offset += if remaining.starts_with("sk_live_") {
"sk_live_".len()
} else {
"sk_test_".len()
};
while let Some(character) = raw[offset..].chars().next() {
if !character.is_ascii_alphanumeric() && !matches!(character, '_' | '-') {
break;
}
offset += character.len_utf8();
}
} else {
let character = remaining.chars().next().expect("nonempty error suffix");
redacted.push(character);
offset += character.len_utf8();
}
}
redacted
.chars()
.filter(|character| !character.is_control())
.take(MAX_ERROR_BYTES)
.collect()
}
#[cfg(unix)]
async fn wait_for_signal() -> Result<()> {
let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
tokio::select! {
result = tokio::signal::ctrl_c() => result.context("listen for interrupt signal"),
_ = terminate.recv() => Ok(()),
}
}
#[cfg(not(unix))]
async fn wait_for_signal() -> Result<()> {
tokio::signal::ctrl_c()
.await
.context("listen for interrupt signal")
}
async fn run(
#[cfg(feature = "testing-endpoints")] lane: Option<test_lane::TestLane>,
) -> Result<()> {
let mut stdin = BufReader::new(tokio::io::stdin());
let config_line = read_bounded_line(&mut stdin, MAX_CONFIG_BYTES)
.await?
.ok_or_else(|| anyhow!("node host config is required on stdin"))?;
let config = HostConfig::parse(&config_line)?;
let device_signer = Arc::new(FileSigner::load_or_create(
&config.state_directory.join("device-ed25519.key"),
)?);
let assignment_signer = Arc::new(FileSigner::load_or_create(
&config.state_directory.join("assignment-ed25519.key"),
)?);
let mut iroh_secret = load_or_create_endpoint_key(config.state_directory.join("iroh.key"))?.to_bytes();
let control_plane = ControlPlane::anonymous(&config.api_key, device_signer)?;
#[cfg(feature = "testing-endpoints")]
let control_plane = match &lane {
Some(lane) => lane.configure(control_plane, &config)?,
None => control_plane,
};
let authority = control_plane
.join_authority_room(
config.secret_key,
config.room_id.clone(),
config.device_id.clone(),
"node-service",
RoomAuthorityOptions {
service_id: config.service_id.clone(),
generation: config.generation,
shard_ids: config.shard_ids.clone(),
max_peers: Some(config.max_peers),
features: Features {
iroh_relay: config.iroh_relay,
..Features::default()
},
assignment_signer,
},
)
.await?;
let mut transport = TransportConfig::default();
transport.relay = config.iroh_relay;
transport.webrtc = Some(WebRTCConfig::default());
#[cfg(feature = "testing-endpoints")]
if lane.is_some() {
transport.webrtc = None;
}
let client = authority
.compose_client(
Client::builder(config.api_key.clone(), Box::new(|| None))?.transport_config(transport),
)
.await?;
let mut service_errors = authority.subscribe_service_errors()?;
let mut service_errors_open = true;
let mut state_updates = client.connection_state_updates();
let mut peer_messages = client.subscribe_native_peer_data();
#[cfg(feature = "testing-endpoints")]
let node_id = match &lane {
Some(lane) => lane.endpoint(&client, &iroh_secret).await?,
None => {
client
.init_iroh(Some(iroh_secret.to_vec()), Vec::new())
.await?
}
};
#[cfg(not(feature = "testing-endpoints"))]
let node_id = client
.init_iroh(Some(iroh_secret.to_vec()), Vec::new())
.await?;
iroh_secret.zeroize();
let ticket = client
.endpoint_ticket_with_token(&format!("v2:room:{}", config.room_id), config.max_peers)
.await?;
#[cfg(feature = "testing-endpoints")]
let ticket = match lane.as_ref() {
Some(lane) => lane.prepare_ticket(&client, ticket)?,
None => ticket,
};
client
.update_presence(
&config.room_id,
&config.device_name,
&ticket,
Some("{\"openrtcNodeHost\":true}"),
)
.await?;
let mut stdout = tokio::io::stdout();
emit(
&mut stdout,
&OutputEvent::Ready {
protocol_version: PROTOCOL_VERSION,
device_id: &config.device_id,
room_id: &config.room_id,
node_id: &node_id,
#[cfg(feature = "testing-endpoints")]
test_lane: lane.as_ref(),
},
)
.await?;
let mut emitted_state = HashSet::<String>::new();
let mut projected_generations = HashMap::<String, (u64, u64, u64)>::new();
let mut command = Box::pin(read_command(&mut stdin));
let mut signal = Box::pin(wait_for_signal());
loop {
tokio::select! {
next = &mut command => {
let next = next?;
drop(command);
let Some(next) = next else { break };
let id = next.id().to_string();
let result = match next {
Command::PublishAssignment {
revision,
subject_device_id,
shard_id,
priority_device_ids,
relevant_entity_ids,
ttl_ms,
..
} => authority.publish_assignment(
revision,
subject_device_id,
shard_id,
priority_device_ids,
relevant_entity_ids,
ttl_ms,
).await.and_then(|assignment| {
serde_json::to_value(assignment).context("encode authority assignment")
}),
Command::Send { connection_id, payload_base64, .. } => {
let decoded = base64::engine::general_purpose::STANDARD
.decode(payload_base64)
.context("decode application payload");
match decoded {
Ok(payload) if payload.len() <= MAX_APPLICATION_PAYLOAD_BYTES => {
client.send_peer(&connection_id, &payload).await.map(|_| Value::Null)
}
Ok(_) => Err(anyhow!("application payload exceeds its byte limit")),
Err(error) => Err(error),
}
}
Command::Shutdown { .. } => {
emit(&mut stdout, &OutputEvent::Response {
id: &id,
ok: true,
value: Some(Value::Null),
error: None,
service_error: None,
}).await?;
break;
}
};
match result {
Ok(value) => emit(&mut stdout, &OutputEvent::Response {
id: &id,
ok: true,
value: Some(value),
error: None,
service_error: None,
}).await?,
Err(error) => emit(&mut stdout, &OutputEvent::Response {
id: &id,
ok: false,
value: None,
error: Some(bounded_error(&error)),
service_error: openrtc::service_errors::ServiceError::from_error(&error).cloned(),
}).await?,
}
command = Box::pin(read_command(&mut stdin));
}
update = state_updates.recv() => match update {
Ok(snapshot) => {
let fingerprint = serde_json::to_string(&snapshot)?;
if emitted_state.insert(fingerprint) {
if emitted_state.len() > 256 {
emitted_state.clear();
}
if snapshot.routable {
if let Some(transport_stable_id) = snapshot.active_transport_stable_id {
projected_generations.insert(
snapshot.connection_id.clone(),
(
transport_stable_id,
snapshot.transport_generation,
snapshot.route_generation,
),
);
}
}
let transport_stable_id = snapshot
.active_transport_stable_id
.or_else(|| projected_generations
.get(&snapshot.connection_id)
.filter(|(_, transport_generation, route_generation)| {
*transport_generation == snapshot.transport_generation
&& *route_generation == snapshot.route_generation
})
.map(|(transport_stable_id, _, _)| *transport_stable_id))
.map(|value| value.to_string());
let transport_generation = snapshot.transport_generation.to_string();
let route_generation = snapshot.route_generation.to_string();
emit(&mut stdout, &OutputEvent::ConnectionState {
snapshot: &snapshot,
transport_stable_id: transport_stable_id.as_deref(),
transport_generation: &transport_generation,
route_generation: &route_generation,
}).await?;
if matches!(snapshot.state.to_ascii_lowercase().as_str(), "closed" | "failed")
&& projected_generations
.get(&snapshot.connection_id)
.is_some_and(|(_, transport_generation, route_generation)| {
*transport_generation == snapshot.transport_generation
&& *route_generation == snapshot.route_generation
})
{
projected_generations.remove(&snapshot.connection_id);
}
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
bail!("node host connection-state projection lagged");
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
bail!("node host connection-state projection closed");
}
},
message = peer_messages.recv() => match message {
Ok(message) => {
let encoded = base64::engine::general_purpose::STANDARD.encode(&message.payload);
let transport_stable_id = message.transport_stable_id.to_string();
let transport_generation = message.transport_generation.to_string();
let route_generation = message.route_generation.to_string();
emit(&mut stdout, &OutputEvent::PeerMessage {
connection_id: &message.connection_id,
payload_base64: &encoded,
transport_stable_id: &transport_stable_id,
transport_generation: &transport_generation,
route_generation: &route_generation,
}).await?;
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
bail!("node host peer-message projection lagged");
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
bail!("node host peer-message projection closed");
}
},
observation = service_errors.recv(), if service_errors_open => match observation {
Ok(observation) => {
emit(&mut stdout, &OutputEvent::ServiceError {
avenue: OutputServiceAvenue {
kind: &observation.avenue.kind,
id: &observation.avenue.id,
},
runtime_instance_id: &observation.runtime_instance_id,
service_error: &observation.service_error,
}).await?;
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
continue;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
service_errors_open = false;
}
},
result = &mut signal => {
result?;
break;
}
}
}
let _ = client.set_offline(&config.room_id).await;
authority.close().await;
Ok(())
}
#[tokio::main]
async fn main() -> Result<()> {
let mut arguments = std::env::args().skip(1).collect::<Vec<_>>();
let protocol_info = arguments
.last()
.is_some_and(|value| value == "--protocol-info");
if protocol_info {
arguments.pop();
}
#[cfg(feature = "testing-endpoints")]
let lane = test_lane::TestLane::parse(&arguments)?;
#[cfg(not(feature = "testing-endpoints"))]
if !arguments.is_empty() {
bail!("openrtc-node-host accepts only --protocol-info");
}
if protocol_info {
let info = json!({
"protocolVersion": PROTOCOL_VERSION,
"maxConfigBytes": MAX_CONFIG_BYTES,
"maxCommandBytes": MAX_COMMAND_BYTES,
"maxApplicationPayloadBytes": MAX_APPLICATION_PAYLOAD_BYTES,
"lifecycleOwner": "rust-peer-session",
});
#[cfg(feature = "testing-endpoints")]
let info = {
let mut info = info;
if let Some(lane) = &lane {
info["testLane"] = serde_json::to_value(lane)?;
}
info
};
println!("{}", info);
return Ok(());
}
let result = run(
#[cfg(feature = "testing-endpoints")]
lane,
)
.await;
if let Err(error) = result {
let message = bounded_error(&error);
let _ = emit(&mut tokio::io::stdout(), &OutputEvent::Failure {
error: message.clone(),
service_error: openrtc::service_errors::ServiceError::from_error(&error).cloned(),
}).await;
return Err(anyhow!("openrtc-node-host failed: {message}"));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
struct TestDirectory(PathBuf);
impl TestDirectory {
fn create() -> Self {
let path = std::env::temp_dir().join(format!(
"openrtc-node-host-test-{}-{}",
std::process::id(),
uuid::Uuid::new_v4()
));
#[cfg(unix)]
fs::DirBuilder::new().mode(0o700).create(&path).unwrap();
#[cfg(not(unix))]
fs::create_dir(&path).unwrap();
Self(path)
}
fn path(&self) -> &Path {
&self.0
}
}
impl Drop for TestDirectory {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.0);
}
}
fn config(state_directory: &Path) -> String {
json!({
"protocolVersion": 1,
"apiKey": "pk_test_0000000000000000000000000000000000000000",
"secretKey": "sk_test_node_host",
"serviceId": "authority-1",
"generation": 1,
"shardIds": ["zone-a"],
"deviceId": "node-host-1",
"roomId": "room-1",
"deviceName": "Node host",
"stateDirectory": state_directory,
})
.to_string()
}
#[test]
fn response_service_metadata_is_additive_and_allowlisted() {
for code in openrtc::service_errors::SERVICE_ERROR_CODES {
let service_error = openrtc::service_errors::ServiceError::from_value(&json!({
"code": code, "retryable": true, "scope": "app", "operation": "session.put",
"requestId": "request-1", "retryAfterMs": 120000, "resetAt": 1800000000000_u64,
"providerCost": 99, "secretKey": "must-not-escape",
}));
let event = OutputEvent::Response { id: "command-1", ok: false, value: None,
error: Some("denied".into()), service_error: service_error.clone() };
let value = serde_json::to_value(event).unwrap();
assert_eq!(value["error"], "denied");
assert_eq!(value["serviceError"]["code"], *code);
assert_eq!(value["serviceError"]["retryAfterMs"], 120000);
assert!(value["serviceError"].get("providerCost").is_none());
assert!(value["serviceError"].get("secretKey").is_none());
let event = OutputEvent::Failure { error: "denied".into(), service_error };
let value = serde_json::to_value(event).unwrap();
assert_eq!(value["type"], "failure");
assert_eq!(value["serviceError"]["code"], *code);
assert!(value.get("id").is_none());
}
let event = OutputEvent::Response { id: "legacy", ok: false, value: None,
error: Some("unknown failure".into()), service_error: None };
assert!(serde_json::to_value(event).unwrap().get("serviceError").is_none());
}
#[test]
fn unsolicited_service_error_event_serializes_nested_camel_case_and_sanitized_fields() {
let service_error = openrtc::service_errors::ServiceError::from_value(&json!({
"code": "app-rate-limited", "retryable": false, "scope": "app",
"operation": "gateway.grant.issue", "requestId": "obs-1",
"providerCost": 99, "message": "must-not-escape",
})).unwrap();
let event = OutputEvent::ServiceError {
avenue: OutputServiceAvenue { kind: "room", id: "room-1" },
runtime_instance_id: "runtime-1",
service_error: &service_error,
};
let value = serde_json::to_value(event).unwrap();
assert_eq!(value["type"], "service-error");
assert_eq!(value["avenue"]["kind"], "room");
assert_eq!(value["avenue"]["id"], "room-1");
assert_eq!(value["runtimeInstanceId"], "runtime-1");
assert_eq!(value["serviceError"]["retryable"], false);
assert!(value["serviceError"].get("providerCost").is_none());
assert!(value["serviceError"].get("message").is_none());
}
#[test]
fn config_is_bounded_and_secret_safe() {
let directory = TestDirectory::create();
let parsed = HostConfig::parse(&config(directory.path())).unwrap();
assert_eq!(parsed.room_id, "room-1");
assert_eq!(parsed.max_peers, 8);
let mut body: serde_json::Value = serde_json::from_str(&config(directory.path())).unwrap();
for max_peers in [0, 51, 5_000] {
body["maxPeers"] = json!(max_peers);
assert!(HostConfig::parse(&body.to_string()).is_err());
}
for max_peers in [1, 8, 50] {
body["maxPeers"] = json!(max_peers);
assert_eq!(
HostConfig::parse(&body.to_string()).unwrap().max_peers,
max_peers
);
}
assert!(HostConfig::parse(&config(directory.path()).replace(
"\"secretKey\":\"sk_test_node_host\"",
"\"secretKey\":\"pk_test_not_a_secret\"",
))
.is_err());
assert_eq!(
bounded_error(&anyhow!("server returned sk_test_node_host")),
"server returned [REDACTED]"
);
}
#[test]
#[cfg(unix)]
fn key_files_are_stable_and_private() {
let directory = TestDirectory::create();
let path = directory.path().join("device.key");
let first = load_or_create_endpoint_key(&path).unwrap();
let second = load_or_create_endpoint_key(&path).unwrap();
assert_eq!(first.public(), second.public());
#[cfg(unix)]
assert_eq!(fs::metadata(path).unwrap().permissions().mode() & 0o077, 0);
}
#[tokio::test]
async fn input_reader_rejects_an_oversized_line() {
let input = vec![b'a'; MAX_COMMAND_BYTES + 1];
let mut reader = BufReader::new(input.as_slice());
assert!(read_bounded_line(&mut reader, MAX_COMMAND_BYTES)
.await
.is_err());
}
#[test]
fn command_decoder_denies_unknown_or_oversized_input() {
let unknown = json!({ "type": "dial", "id": "request-1" }).to_string();
assert!(serde_json::from_str::<Command>(&unknown).is_err());
let oversized = json!({
"type": "send",
"id": "request-1",
"connectionId": "connection-1",
"payloadBase64": "A".repeat(MAX_APPLICATION_PAYLOAD_BYTES * 2 + 1),
})
.to_string();
assert!(serde_json::from_str::<Command>(&oversized)
.unwrap()
.validate()
.is_err());
}
#[test]
fn process_protocol_exposes_raw_payloads_not_transport_frames() {
let encoded = serde_json::to_value(OutputEvent::PeerMessage {
connection_id: "connection-1",
payload_base64: "aGVsbG8=",
transport_stable_id: "17",
transport_generation: "3",
route_generation: "5",
})
.unwrap();
assert_eq!(encoded["payloadBase64"], "aGVsbG8=");
assert_eq!(encoded["transportStableId"], "17");
assert_eq!(encoded["transportGeneration"], "3");
assert_eq!(encoded["routeGeneration"], "5");
assert!(encoded.get("frameBase64").is_none());
}
}