use anyhow::{anyhow, bail, Context, Result};
use async_trait::async_trait;
use base64::Engine as _;
use ed25519_dalek::{Signer as _, SigningKey};
use futures::StreamExt;
use openrtc::client::TransportConfig;
#[cfg(test)]
use openrtc::native::RoomArchitecturePhase;
use openrtc::native::{
adaptive_room_sentinel_owner_limits, DeviceSigner, InMemoryNativeManagedGroupStateStore,
NativeCapabilities, NativeGatewayGrant, NativeGatewayGrantProvider, NativeGatewayGrantRequest,
NativeManagedRoomMessage, NativeManagedRoomPublish, RoomArchitectureMode,
RoomArchitectureSnapshot, RoomDelivery,
};
use openrtc::Client;
use reqwest::header::{HeaderValue, AUTHORIZATION, CONTENT_TYPE};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader};
use zeroize::Zeroize;
const INPUT_PROTOCOL_VERSION: u8 = 1;
const SIGNER_PROTOCOL_VERSION: u8 = 1;
const MAX_CONFIG_BYTES: usize = 16 * 1024;
const MAX_COMMAND_BYTES: usize = 48 * 1024;
const MAX_SIGNER_BODY_BYTES: usize = 64 * 1024;
const MAX_MANAGED_PAYLOAD_BYTES: usize = 32 * 1024;
const SIGNER_TIMEOUT: Duration = Duration::from_secs(10);
const BUILD_SOURCE_COMMIT: &str = match option_env!("OPENRTC_SENTINEL_BUILD_SOURCE_COMMIT") {
Some(value) => value,
None => "unbound",
};
const BUILD_SOURCE_MANIFEST_FINGERPRINT: &str =
match option_env!("OPENRTC_SENTINEL_BUILD_SOURCE_MANIFEST_FINGERPRINT") {
Some(value) => value,
None => "unbound",
};
const BUILD_ID: &str = match option_env!("OPENRTC_SENTINEL_BUILD_ID") {
Some(value) => value,
None => "unbound",
};
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct SentinelConfig {
protocol_version: u8,
api_key: String,
gateway_url: String,
signer_url: String,
signer_token: String,
expected_build_id: String,
expected_source_manifest_fingerprint: String,
device_id: String,
room_id: String,
device_name: String,
#[serde(default = "default_platform_type")]
platform_type: String,
#[serde(default)]
architecture: RoomArchitectureMode,
#[serde(default)]
delivery: RoomDelivery,
#[serde(default)]
relay_only: bool,
}
impl SentinelConfig {
fn parse(line: &str) -> Result<Self> {
let mut config: Self = serde_json::from_str(line).context("decode sentinel config")?;
if config.protocol_version != INPUT_PROTOCOL_VERSION {
bail!("unsupported sentinel input protocol");
}
config.api_key = openrtc::validate_api_key(&config.api_key)?.to_string();
config.gateway_url = bounded_url("gatewayUrl", config.gateway_url, false)?;
config.signer_url = bounded_url("signerUrl", config.signer_url, true)?;
config.signer_token = bounded_secret("signerToken", config.signer_token, 4_096)?;
config.expected_build_id = bounded_sha256("expectedBuildId", config.expected_build_id)?;
config.expected_source_manifest_fingerprint = bounded_sha256(
"expectedSourceManifestFingerprint",
config.expected_source_manifest_fingerprint,
)?;
config.device_id = bounded_id("deviceId", config.device_id, 160)?;
config.room_id = bounded_id("roomId", config.room_id, 160)?;
config.device_name = bounded_text("deviceName", config.device_name, 80)?;
config.platform_type = bounded_id("platformType", config.platform_type, 64)?;
Ok(config)
}
fn validate_build_identity(&self) -> Result<()> {
if BUILD_ID == "unbound"
|| BUILD_SOURCE_COMMIT == "unbound"
|| BUILD_SOURCE_MANIFEST_FINGERPRINT == "unbound"
|| self.expected_build_id != BUILD_ID
|| self.expected_source_manifest_fingerprint != BUILD_SOURCE_MANIFEST_FINGERPRINT
{
bail!("native sentinel build identity is stale or unbound");
}
Ok(())
}
}
fn default_platform_type() -> String {
"adaptive-room-sentinel".to_string()
}
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!("sentinel {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!("sentinel {name} is invalid");
}
Ok(value)
}
fn bounded_secret(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_graphic())
{
bail!("sentinel {name} is invalid");
}
Ok(value)
}
fn bounded_sha256(name: &str, value: String) -> Result<String> {
let value = value.trim().to_ascii_lowercase();
if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
bail!("sentinel {name} is invalid");
}
Ok(value)
}
fn bounded_url(name: &str, value: String, loopback_only: bool) -> Result<String> {
let value = bounded_text(name, value, 2_048)?;
let url = reqwest::Url::parse(&value).with_context(|| format!("parse sentinel {name}"))?;
let loopback = matches!(
url.host_str(),
Some("127.0.0.1") | Some("localhost") | Some("::1")
);
let allowed_scheme = if loopback_only {
url.scheme() == "http"
} else {
matches!(url.scheme(), "https" | "wss")
|| (loopback && matches!(url.scheme(), "http" | "ws"))
};
if !allowed_scheme
|| (loopback_only && !loopback)
|| !url.username().is_empty()
|| url.password().is_some()
|| url.query().is_some()
|| url.fragment().is_some()
{
bail!("sentinel {name} is invalid");
}
Ok(value.trim_end_matches('/').to_string())
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct SignerAvenue<'a> {
kind: &'a str,
id: &'a str,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct SignerGrantRequest<'a> {
protocol_version: u8,
app_tag: &'a str,
avenue: SignerAvenue<'a>,
device_id: &'a str,
runtime_instance_id: &'a str,
ticket_fingerprint: &'a str,
purpose: &'a str,
architecture: Option<RoomArchitectureMode>,
room_delivery: Option<RoomDelivery>,
#[serde(skip_serializing_if = "Option::is_none")]
refresh_grant: Option<&'a str>,
}
struct LoopbackGrantProvider {
client: reqwest::Client,
signer_url: String,
authorization: HeaderValue,
app_tag: String,
}
struct EphemeralDeviceSigner(SigningKey);
impl EphemeralDeviceSigner {
fn generate() -> Result<Self> {
let mut secret = [0_u8; 32];
getrandom::getrandom(&mut secret).context("generate sentinel device key")?;
let signer = Self(SigningKey::from_bytes(&secret));
secret.zeroize();
Ok(signer)
}
}
impl DeviceSigner for EphemeralDeviceSigner {
fn public_jwk(&self, _app_tag: &str) -> Result<serde_json::Value> {
Ok(serde_json::json!({
"kty": "OKP",
"crv": "Ed25519",
"x": base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(self.0.verifying_key().to_bytes()),
}))
}
fn sign(&self, _app_tag: &str, challenge: &[u8]) -> Result<Vec<u8>> {
Ok(self.0.sign(challenge).to_bytes().to_vec())
}
}
impl LoopbackGrantProvider {
fn new(signer_url: String, signer_token: String, app_tag: String) -> Result<Self> {
let client = reqwest::Client::builder()
.timeout(SIGNER_TIMEOUT)
.redirect(reqwest::redirect::Policy::none())
.no_proxy()
.build()?;
let mut authorization = HeaderValue::from_str(&format!("Bearer {signer_token}"))
.context("encode sentinel signer authorization")?;
authorization.set_sensitive(true);
Ok(Self {
client,
signer_url,
authorization,
app_tag,
})
}
}
#[async_trait]
impl NativeGatewayGrantProvider for LoopbackGrantProvider {
async fn grant(&self, request: NativeGatewayGrantRequest) -> Result<NativeGatewayGrant> {
let body = serde_json::to_vec(&SignerGrantRequest {
protocol_version: SIGNER_PROTOCOL_VERSION,
app_tag: &self.app_tag,
avenue: SignerAvenue {
kind: &request.avenue.kind,
id: &request.avenue.id,
},
device_id: &request.device_id,
runtime_instance_id: &request.runtime_instance_id,
ticket_fingerprint: &request.ticket_fingerprint,
purpose: &request.purpose,
architecture: request.architecture,
room_delivery: request.room_delivery,
refresh_grant: request.refresh_grant.as_deref(),
})?;
if body.len() > MAX_SIGNER_BODY_BYTES {
bail!("sentinel signer request exceeds its byte limit");
}
let response = self
.client
.post(&self.signer_url)
.header(AUTHORIZATION, self.authorization.clone())
.header(CONTENT_TYPE, "application/json")
.body(body)
.send()
.await
.context("request a fresh gateway grant from the loopback signer")?;
if !response.status().is_success() {
bail!(
"loopback signer rejected the grant request with status {}",
response.status()
);
}
if response
.content_length()
.is_some_and(|length| length > MAX_SIGNER_BODY_BYTES as u64)
{
bail!("sentinel signer response exceeds its byte limit");
}
let mut bytes = Vec::new();
let mut stream = response.bytes_stream();
while let Some(chunk) = stream.next().await {
let chunk = chunk.context("read loopback signer response")?;
if bytes.len().saturating_add(chunk.len()) > MAX_SIGNER_BODY_BYTES {
bail!("sentinel signer response exceeds its byte limit");
}
bytes.extend_from_slice(&chunk);
}
decode_grant_response(&bytes)
}
}
fn decode_grant_response(bytes: &[u8]) -> Result<NativeGatewayGrant> {
if bytes.is_empty() || bytes.len() > MAX_SIGNER_BODY_BYTES {
bail!("sentinel signer response has an invalid size");
}
serde_json::from_slice(bytes).context("decode loopback signer grant")
}
#[derive(Debug, Deserialize)]
#[serde(
tag = "type",
rename_all = "kebab-case",
rename_all_fields = "camelCase",
deny_unknown_fields
)]
enum InputCommand {
Shutdown,
ObserveArchitecture,
ManagedPublish {
message_id: String,
channel: String,
#[serde(default)]
priority: u8,
#[serde(default)]
zone_id: Option<String>,
payload_base64: String,
},
}
impl InputCommand {
fn into_managed_publish(self) -> Result<Option<NativeManagedRoomPublish>> {
let Self::ManagedPublish {
message_id,
channel,
priority,
zone_id,
payload_base64,
} = self
else {
return Ok(None);
};
let message_id = bounded_id("messageId", message_id, 160)?;
let channel = bounded_id("channel", channel, 80)?;
if priority > 3 {
bail!("sentinel managed priority is invalid");
}
let zone_id = zone_id
.map(|value| bounded_id("zoneId", value, 160))
.transpose()?;
if payload_base64.len() > MAX_MANAGED_PAYLOAD_BYTES.saturating_mul(2) {
bail!("sentinel managed payload exceeds its byte limit");
}
let payload = base64::engine::general_purpose::STANDARD
.decode(payload_base64)
.context("decode sentinel managed payload")?;
if payload.is_empty() || payload.len() > MAX_MANAGED_PAYLOAD_BYTES {
bail!("sentinel managed payload exceeds its byte limit");
}
Ok(Some(NativeManagedRoomPublish {
message_id,
channel,
priority,
zone_id,
payload,
}))
}
}
#[derive(Serialize)]
#[serde(
tag = "type",
rename_all = "kebab-case",
rename_all_fields = "camelCase"
)]
enum OutputEvent<'a> {
InitialReady {
protocol_version: u8,
device_id: &'a str,
room_id: &'a str,
node_id: &'a str,
build_id: &'a str,
source_manifest_fingerprint: &'a str,
},
RoomArchitecture {
protocol_version: u8,
device_id: &'a str,
room_id: &'a str,
snapshot: &'a RoomArchitectureSnapshot,
},
ManagedPublished {
protocol_version: u8,
device_id: &'a str,
room_id: &'a str,
message_id: &'a str,
},
ManagedReceived {
protocol_version: u8,
device_id: &'a str,
room_id: &'a str,
sender_device_id: &'a str,
message_id: &'a str,
channel: &'a str,
priority: u8,
zone_id: Option<&'a str>,
payload_base64: &'a str,
},
}
async fn emit_json_line<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!("sentinel 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("sentinel input must be UTF-8")
}
async fn read_input_command<R: AsyncBufRead + Unpin>(
reader: &mut R,
) -> Result<Option<InputCommand>> {
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::<InputCommand>(&line)
.map(Some)
.context("decode sentinel command");
}
}
#[cfg(unix)]
async fn wait_for_process_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_process_signal() -> Result<()> {
tokio::signal::ctrl_c()
.await
.context("listen for interrupt signal")
}
async fn run() -> 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!("sentinel config is required on stdin"))?;
let config = SentinelConfig::parse(&config_line)?;
config.validate_build_identity()?;
let app_tag = openrtc::app_tag_from_api_key(&config.api_key);
let grant_provider: Arc<dyn NativeGatewayGrantProvider> = Arc::new(LoopbackGrantProvider::new(
config.signer_url,
config.signer_token,
app_tag.clone(),
)?);
let capabilities = NativeCapabilities::new(
&config.api_key,
config.device_id.clone(),
config.platform_type,
grant_provider,
)?
.with_managed_group_storage(
Arc::new(EphemeralDeviceSigner::generate()?),
Arc::new(InMemoryNativeManagedGroupStateStore::default()),
)
.with_testing_endpoint(config.gateway_url)?;
let handle = capabilities.join_room_with_options(
config.room_id.clone(),
config.architecture,
config.delivery,
)?;
let mut architecture = handle.subscribe_room_architecture()?;
let mut managed_messages = handle.subscribe_managed_room()?;
let mut transport = TransportConfig::default();
transport.iroh_relay_only = config.relay_only;
let client = Arc::new(
Client::builder(config.api_key, Box::new(|| None))?
.transport_config(transport)
.signaling_backend(handle.signaling())
.build(),
);
let node_id = client.init_iroh(None, Vec::new()).await?;
let ticket = client.endpoint_ticket().await?;
client
.update_presence(
&config.room_id,
&config.device_name,
&ticket,
Some("{\"adaptiveRoomSentinel\":true}"),
)
.await?;
let mut stdout = tokio::io::stdout();
emit_json_line(
&mut stdout,
&OutputEvent::InitialReady {
protocol_version: INPUT_PROTOCOL_VERSION,
device_id: &config.device_id,
room_id: &config.room_id,
node_id: &node_id,
build_id: BUILD_ID,
source_manifest_fingerprint: BUILD_SOURCE_MANIFEST_FINGERPRINT,
},
)
.await?;
let mut last_architecture = None;
let mut input_command = Box::pin(read_input_command(&mut stdin));
let mut process_signal = Box::pin(wait_for_process_signal());
loop {
let current = architecture.borrow_and_update().clone();
if let Some(snapshot) = current {
if last_architecture.as_ref() != Some(&snapshot) {
emit_json_line(
&mut stdout,
&OutputEvent::RoomArchitecture {
protocol_version: INPUT_PROTOCOL_VERSION,
device_id: &config.device_id,
room_id: &config.room_id,
snapshot: &snapshot,
},
)
.await?;
last_architecture = Some(snapshot);
}
}
tokio::select! {
changed = architecture.changed() => {
if changed.is_err() {
bail!("native room architecture observer stopped");
}
}
result = &mut input_command => {
let command = result?;
drop(input_command);
match command {
None | Some(InputCommand::Shutdown) => break,
Some(InputCommand::ObserveArchitecture) => {
let snapshot = handle.room_architecture()?
.ok_or_else(|| anyhow!("native room architecture is unavailable"))?;
emit_json_line(
&mut stdout,
&OutputEvent::RoomArchitecture {
protocol_version: INPUT_PROTOCOL_VERSION,
device_id: &config.device_id,
room_id: &config.room_id,
snapshot: &snapshot,
},
).await?;
input_command = Box::pin(read_input_command(&mut stdin));
}
Some(command) => {
let publish = command
.into_managed_publish()?
.expect("non-shutdown command is a managed publish");
let message_id = publish.message_id.clone();
handle.publish_managed_room(vec![publish]).await?;
emit_json_line(
&mut stdout,
&OutputEvent::ManagedPublished {
protocol_version: INPUT_PROTOCOL_VERSION,
device_id: &config.device_id,
room_id: &config.room_id,
message_id: &message_id,
},
).await?;
input_command = Box::pin(read_input_command(&mut stdin));
}
}
}
received = managed_messages.recv() => {
let message = received.context("native managed room observer lagged or stopped")?;
emit_managed_received(
&mut stdout,
&config.device_id,
&config.room_id,
&message,
).await?;
}
result = &mut process_signal => {
result?;
break;
}
}
}
let _ = client.set_offline(&config.room_id).await;
handle.close().await;
Ok(())
}
async fn emit_managed_received<W: AsyncWrite + Unpin>(
writer: &mut W,
device_id: &str,
room_id: &str,
message: &NativeManagedRoomMessage,
) -> Result<()> {
let payload_base64 =
base64::engine::general_purpose::STANDARD.encode(message.payload.as_slice());
emit_json_line(
writer,
&OutputEvent::ManagedReceived {
protocol_version: INPUT_PROTOCOL_VERSION,
device_id,
room_id,
sender_device_id: &message.sender_device_id,
message_id: &message.message_id,
channel: &message.channel,
priority: message.priority,
zone_id: message.zone_id.as_deref(),
payload_base64: &payload_base64,
},
)
.await
}
#[tokio::main]
async fn main() -> Result<()> {
let arguments = std::env::args().skip(1).collect::<Vec<_>>();
if arguments == ["--evidence-metadata"] {
println!(
"{}",
serde_json::json!({
"schemaVersion": 1,
"sourceCommit": BUILD_SOURCE_COMMIT,
"sourceManifestFingerprint": BUILD_SOURCE_MANIFEST_FINGERPRINT,
"buildId": BUILD_ID,
"ownerLimits": adaptive_room_sentinel_owner_limits(),
})
);
return Ok(());
}
if !arguments.is_empty() {
bail!("adaptive-room sentinel accepts only --evidence-metadata");
}
run().await
}
#[cfg(test)]
mod tests {
use super::*;
fn valid_config() -> String {
serde_json::json!({
"protocolVersion": 1,
"apiKey": "pk_test_0000000000000000000000000000000000000000",
"gatewayUrl": "https://gateway-staging.openrtc.app",
"signerUrl": "http://127.0.0.1:43123/grants",
"signerToken": "sentinel-auth-token",
"expectedBuildId": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"expectedSourceManifestFingerprint": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"deviceId": "native-sentinel-1",
"roomId": "room-1",
"deviceName": "Native sentinel"
})
.to_string()
}
#[test]
fn config_accepts_only_bounded_loopback_signer_inputs() {
let config = SentinelConfig::parse(&valid_config()).expect("valid config");
assert_eq!(config.architecture, RoomArchitectureMode::Auto);
assert_eq!(config.delivery, RoomDelivery::Reliable);
assert!(config.validate_build_identity().is_err());
let remote = valid_config().replace(
"http://127.0.0.1:43123/grants",
"https://example.com/grants",
);
assert!(SentinelConfig::parse(&remote).is_err());
let signing_key = valid_config().replace(
"\"deviceName\":\"Native sentinel\"",
"\"deviceName\":\"Native sentinel\",\"signingKey\":\"private\"",
);
assert!(SentinelConfig::parse(&signing_key).is_err());
}
#[tokio::test]
async fn input_reader_enforces_the_limit_before_returning_a_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 managed_publish_command_is_bounded_and_decoded_once() {
let command: InputCommand = serde_json::from_value(serde_json::json!({
"type": "managed-publish",
"messageId": "message-1",
"channel": "state",
"priority": 2,
"payloadBase64": "aGVsbG8="
}))
.expect("command");
let publish = command
.into_managed_publish()
.expect("valid publish")
.expect("publish command");
assert_eq!(publish.payload, b"hello");
let oversized: InputCommand = serde_json::from_value(serde_json::json!({
"type": "managed-publish",
"messageId": "message-2",
"channel": "state",
"payloadBase64": "A".repeat(MAX_MANAGED_PAYLOAD_BYTES * 2 + 1)
}))
.expect("command shape");
assert!(oversized.into_managed_publish().is_err());
let observe: InputCommand = serde_json::from_value(serde_json::json!({
"type": "observe-architecture"
}))
.expect("observation command");
assert!(matches!(observe, InputCommand::ObserveArchitecture));
}
#[test]
fn grant_response_is_bounded() {
assert!(decode_grant_response(&vec![b'x'; MAX_SIGNER_BODY_BYTES + 1]).is_err());
let grant = serde_json::json!({
"protocolVersion": 2,
"gatewayUrl": "https://gateway-staging.openrtc.app",
"routeKey": "room:room-1",
"token": "opaque-token",
"expiresAtMs": 4_000_000_000_000_u64
});
let decoded = decode_grant_response(grant.to_string().as_bytes()).expect("grant");
assert_eq!(decoded.route_key, "room:room-1");
}
#[tokio::test]
async fn output_contains_only_customer_safe_observation_fields() {
let snapshot = RoomArchitectureSnapshot {
requested: RoomArchitectureMode::Auto,
effective: openrtc::native::EffectiveRoomArchitecture::Managed,
epoch: 3,
phase: RoomArchitecturePhase::Settled,
reason: openrtc::native::RoomArchitectureReason::Size,
held_credits_usd: 0.25,
quote_expires_at_ms: 4_000_000_000_000,
};
let mut output = Vec::new();
emit_json_line(
&mut output,
&OutputEvent::RoomArchitecture {
protocol_version: 1,
device_id: "device-1",
room_id: "room-1",
snapshot: &snapshot,
},
)
.await
.expect("output");
let value: serde_json::Value = serde_json::from_slice(&output).expect("json line");
assert_eq!(value["type"], "room-architecture");
assert!(value.get("signerToken").is_none());
assert!(value.get("gatewayUrl").is_none());
assert!(value.get("apiKey").is_none());
}
#[tokio::test]
async fn managed_receive_output_is_bounded_to_plain_application_fields() {
let message = NativeManagedRoomMessage {
sender_device_id: "sender-1".to_string(),
message_id: "message-1".to_string(),
channel: "state".to_string(),
priority: 1,
zone_id: None,
payload: b"hello".to_vec(),
};
let mut output = Vec::new();
emit_managed_received(&mut output, "device-1", "room-1", &message)
.await
.expect("output");
let value: serde_json::Value = serde_json::from_slice(&output).expect("json line");
assert_eq!(value["type"], "managed-received");
assert_eq!(value["payloadBase64"], "aGVsbG8=");
assert!(value.get("ciphertext").is_none());
assert!(value.get("encryptionEpoch").is_none());
}
}