diff --git a/.gitignore b/.gitignore index 9b2b3d5..a4885f1 100644 --- a/.gitignore +++ b/.gitignore @@ -123,7 +123,7 @@ dist .tern-port # Local working notes (RTCP implementation handover, etc.) -handover.md +handover*.md # Rust rust/target/ diff --git a/index.js b/index.js index 6012b9b..091e966 100644 --- a/index.js +++ b/index.js @@ -204,6 +204,7 @@ class proxy { * @param { Object } [ properties.remote.dtls ] * @param { string } properties.remote.dtls.fingerprint - the fingerprint we verify the remote against * @param { string } properties.remote.dtls.setup - "active" or "passive" + * @param { boolean } [ properties.remote.rtcpmux = false ] - RFC 5761 rtcp-mux: carry RTCP over the RTP port/5-tuple instead of the separate P+1 control port. Set from the SDP `a=rtcp-mux` attribute. * @param { Object } [ properties.direction ] - direction from our perspective * @param { boolean } [ properties.direction.send = true ] * @param { boolean } [ properties.direction.recv = true ] diff --git a/rust/src/channel/actor.rs b/rust/src/channel/actor.rs index f5fa016..4b577c4 100644 --- a/rust/src/channel/actor.rs +++ b/rust/src/channel/actor.rs @@ -224,17 +224,30 @@ pub fn spawn_with_sockets( state.port_reservation = cfg.port_reservation; *state.local_icepwd.lock() = cfg.local_icepwd; - // Spawn the recv_loop — reads the socket continuously, classifies - // STUN/DTLS/RTP and feeds jitter/DTLS-mpsc immediately. let cancel = CancellationToken::new(); + + // DTLS keying material is published here once the handshake completes so + // the inbound RTCP readers can build their SRTCP decrypt context (Tier 2). + // Both the dedicated P+1 loop and — under rtcp-mux — the RTP recv_loop + // subscribe, so create the watch pair before spawning either. + let (srtp_key_tx, srtp_key_rx) = tokio::sync::watch::channel(None); + state.srtp_key_tx = Some(srtp_key_tx); + + // Spawn the recv_loop — reads the RTP socket continuously, classifies + // STUN/DTLS/RTP and feeds jitter/DTLS-mpsc immediately. It also demuxes + // rtcp-mux'd RTCP (RFC 5761) off the RTP port, hence the RTCP accounting + // fields and the SRTCP key subscription. super::recv_loop::spawn(super::recv_loop::RecvLoopConfig { sock: state.rtp_sock.clone(), jitter: state.jitter.clone(), remote_addr: state.remote_addr.clone(), in_count: state.in_count.clone(), rx_stats: state.rx_stats.clone(), + remote_report: state.remote_report.clone(), + local_ssrc: state.ssrc, local_icepwd: state.local_icepwd.clone(), dtls_tx: state.dtls_inbound_tx.clone(), + key_rx: srtp_key_rx.clone(), cancel: cancel.clone(), }); @@ -245,6 +258,7 @@ pub fn spawn_with_sockets( rx_stats: state.rx_stats.clone(), remote_report: state.remote_report.clone(), local_ssrc: state.ssrc, + key_rx: srtp_key_rx, cancel: cancel.clone(), }); @@ -597,6 +611,10 @@ async fn run( }); } subs.prebuffer.clear(); + // RTCP BYE (Tier 2): tell the peer the stream is ending now, before we + // cancel the loops and drop the sockets. Best-effort; encrypted as SRTCP + // on a secure channel. + super::rtcp_tx::send_bye(state).await; // Abort any in-flight DTLS handshake task. Without this, a handshake // that hasn't completed (peer disappeared, no response, etc.) outlives // the channel and busy-spins the runtime: the task's mpsc senders are @@ -736,6 +754,7 @@ async fn handle_command_local( state.set_remote_addr(cfg.addr); state.ticks_without_rtp = 0; state.remote_pt = cfg.payload_type; + state.rtcpmux = cfg.rtcpmux; state.codecx.set_negotiated_pt(cfg.payload_type); if let Some(pt) = cfg.rfc2833_payload_type { state.rfc2833_pt = pt; diff --git a/rust/src/channel/commands.rs b/rust/src/channel/commands.rs index 2768f47..e9c251b 100644 --- a/rust/src/channel/commands.rs +++ b/rust/src/channel/commands.rs @@ -15,6 +15,10 @@ pub struct RemoteConfig { pub ilbc_payload_type: Option, pub rfc2833_payload_type: Option, pub dtls: Option, + /// RFC 5761 rtcp-mux: when true the peer carries RTCP over the RTP port / + /// 5-tuple, so we send/receive RTCP there instead of on the separate P+1 + /// control port. Default false = classic split ports (SIP softphones). + pub rtcpmux: bool, /// ICE password of the *remote* agent. Per RFC 8445 §7.1.1, the remote /// uses our local icepwd to sign STUN Binding Requests it sends us; our /// Binding Responses are signed with the same key. This field is kept diff --git a/rust/src/channel/dtls_session.rs b/rust/src/channel/dtls_session.rs index b41aa4b..e0f9fcd 100644 --- a/rust/src/channel/dtls_session.rs +++ b/rust/src/channel/dtls_session.rs @@ -7,8 +7,10 @@ // Outbound DTLS frames come back via a second mpsc and are sent on the // real socket by the tick. // -// After the handshake completes, keying material is exported and used to -// create SRTP encrypt/decrypt contexts (see srtp_ctx.rs). +// After the handshake completes, keying material is exported and split +// (`split_keying_material` / `local_srtp_params` / `remote_srtp_params`) to +// build the SRTP/SRTCP encrypt/decrypt contexts — RTP contexts in the tick +// (`poll_dtls_handshake`), the inbound SRTCP context in `rtcp_loop`. use std::net::SocketAddr; use std::sync::Arc; @@ -250,6 +252,29 @@ pub fn split_keying_material( } } +/// Key + salt + profile for the **local** (outbound) SRTP/SRTCP direction. +/// The local side writes with the server key when we are the DTLS server, +/// otherwise the client key. +pub fn local_srtp_params(km: &SrtpKeyingMaterial) -> (&[u8], &[u8], ProtectionProfile) { + if km.local_is_server { + (&km.server_write_key, &km.server_write_salt, km.profile) + } else { + (&km.client_write_key, &km.client_write_salt, km.profile) + } +} + +/// Key + salt + profile for the **remote** (inbound) SRTP/SRTCP direction — +/// the peer writes with the opposite key to us. Used to build the SRTCP +/// decrypt context in `rtcp_loop`, mirroring the RTP decrypt context the tick +/// builds in `poll_dtls_handshake`. +pub fn remote_srtp_params(km: &SrtpKeyingMaterial) -> (&[u8], &[u8], ProtectionProfile) { + if km.local_is_server { + (&km.client_write_key, &km.client_write_salt, km.profile) + } else { + (&km.server_write_key, &km.server_write_salt, km.profile) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/rust/src/channel/facade.rs b/rust/src/channel/facade.rs index 8952255..61c397e 100644 --- a/rust/src/channel/facade.rs +++ b/rust/src/channel/facade.rs @@ -322,6 +322,10 @@ impl ChannelObject { let port = params.get_named_property::("port").ok(); let codec = params.get_named_property::("codec").ok().unwrap_or(0); let icepwd = params.get_named_property::("icepwd").ok(); + let rtcpmux = params + .get_named_property::("rtcpmux") + .ok() + .unwrap_or(false); let dtls = parse_remote_dtls(¶ms); let Some(addr_s) = addr else { return false; @@ -343,6 +347,7 @@ impl ChannelObject { ilbc_payload_type: None, rfc2833_payload_type: None, dtls, + rtcpmux, icepwd, }, ack, @@ -1031,6 +1036,16 @@ fn extract_remote_icepwd(params: &Object) -> Option { .filter(|s| !s.is_empty()) } +/// RFC 5761 rtcp-mux flag from `openchannel({ remote: { rtcpmux: true } })`. +/// Absent/false = classic split RTP/RTCP ports. +fn extract_rtcpmux(params: &Object) -> bool { + params + .get_named_property::("remote") + .ok() + .and_then(|r| r.get_named_property::("rtcpmux").ok()) + .unwrap_or(false) +} + /// Parse an optional `dtls` block from a remote config object. JS shape: /// `{ fingerprint: { hash: "sha-256 ..." }, mode: "active"|"passive" }` /// — matches `projectrtpdtls.js` test fixtures. Returns None when missing @@ -1065,6 +1080,7 @@ pub fn open_channel(env: Env, params: Object, callback: JsFunction) -> Result = callback .create_threadsafe_function( @@ -1245,6 +1261,7 @@ pub fn open_channel(env: Env, params: Object, callback: JsFunction) -> Result>) { } run_post_mix_phase(members, n_alive).await; - close_idle_members(members); + close_idle_members(members).await; } /// Per-member inbound work — delegates to `Member::process_inbound` for @@ -993,14 +993,16 @@ fn broadcast_dtmf_to_peer_relays( } /// Remove idle members from the mix and emit each one's Close event. -fn close_idle_members(members: &mut HashMap>) { +async fn close_idle_members(members: &mut HashMap>) { let idle_ids: Vec = members .iter() .filter(|(_, m)| m.is_idle()) .map(|(&id, _)| id) .collect(); for id in idle_ids { - if let Some(m) = members.remove(&id) { + if let Some(mut m) = members.remove(&id) { + // RTCP BYE (Tier 2) before the member (and its sockets) drop. + super::rtcp_tx::send_bye(&mut m.state).await; m.emit_close_event("idle"); } } diff --git a/rust/src/channel/mod.rs b/rust/src/channel/mod.rs index 8397043..f0cb74d 100644 --- a/rust/src/channel/mod.rs +++ b/rust/src/channel/mod.rs @@ -8,9 +8,8 @@ // 2. jitter — reorder buffer // 3. state — ChannelState struct // 4. commands — Command enum + Handle -// 5. dtls_session — gnutls wrapper -// 6. srtp_ctx — libsrtp2 wrapper -// 7. tick — the per-tick pipeline +// 5. dtls_session — DTLS handshake + SRTP keying (webrtc-dtls/-srtp) +// 6. tick — the per-tick pipeline // 8. player / recorder / dtmf — media subsystems // 9. actor — the tokio task // 10. mixer — mix group actor @@ -33,6 +32,5 @@ pub mod rtcp_loop; pub mod rtcp_stats; pub mod rtcp_tx; pub mod rtp; -pub mod srtp_ctx; pub mod state; pub mod tick; diff --git a/rust/src/channel/recv_loop.rs b/rust/src/channel/recv_loop.rs index 11ec426..fbe997d 100644 --- a/rust/src/channel/recv_loop.rs +++ b/rust/src/channel/recv_loop.rs @@ -15,11 +15,13 @@ use std::time::Instant; use parking_lot::Mutex as PLMutex; use tokio::net::UdpSocket; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, watch}; use tokio_util::sync::CancellationToken; +use super::dtls_session::SrtpKeyingMaterial; use super::jitter::JitterBuffer; -use super::rtcp_stats::RxStats; +use super::rtcp_loop; +use super::rtcp_stats::{RemoteReport, RxStats}; use super::rtp::{self, RtpPacket}; use crate::stun; @@ -30,8 +32,16 @@ pub struct RecvLoopConfig { pub in_count: Arc, /// RFC 3550 receiver accounting — fed on every inbound RTP packet. pub rx_stats: Arc>, + /// Peer's view of the stream we send — folded from rtcp-mux'd RTCP + /// (RFC 5761) that arrives on the RTP port. Shared with `rtcp_loop`. + pub remote_report: Arc>, + /// Our SSRC — selects the muxed report block that is about our stream. + pub local_ssrc: u32, pub local_icepwd: Arc>, pub dtls_tx: Arc>>>>, + /// DTLS keying material for decrypting muxed SRTCP; `None` until the + /// handshake completes (and always, for non-secure channels). + pub key_rx: watch::Receiver>, pub cancel: CancellationToken, } @@ -41,6 +51,9 @@ pub fn spawn(cfg: RecvLoopConfig) -> tokio::task::JoinHandle<()> { async fn run(cfg: RecvLoopConfig) { let mut buf = [0u8; rtp::RTP_MAX_LENGTH]; + // Built lazily on the first muxed SRTCP packet after keys arrive; a plain + // (non-secure) channel never populates it and muxed RTCP stays cleartext. + let mut srtcp_decrypt: Option = None; loop { tokio::select! { biased; @@ -49,7 +62,7 @@ async fn run(cfg: RecvLoopConfig) { match result { Ok((n, peer)) => { *cfg.remote_addr.lock() = Some(peer); - handle_packet(&cfg, &buf[..n], peer).await; + handle_packet(&cfg, &buf[..n], peer, &mut srtcp_decrypt).await; } Err(_) => break, } @@ -58,7 +71,12 @@ async fn run(cfg: RecvLoopConfig) { } } -async fn handle_packet(cfg: &RecvLoopConfig, pkt: &[u8], peer: SocketAddr) { +async fn handle_packet( + cfg: &RecvLoopConfig, + pkt: &[u8], + peer: SocketAddr, + srtcp_decrypt: &mut Option, +) { if pkt.is_empty() { return; } @@ -88,6 +106,22 @@ async fn handle_packet(cfg: &RecvLoopConfig, pkt: &[u8], peer: SocketAddr) { return; } + // rtcp-mux (RFC 5761): RTCP carried on the RTP port. See `is_muxed_rtcp` + // for the demux rule. SRTP/SRTCP leave the header in cleartext, so this + // classifies before any decrypt. On a non-mux channel the peer targets P+1 + // and `rtcp_loop` handles it, so this branch simply never fires. + if is_muxed_rtcp(pkt) { + rtcp_loop::maybe_build_decrypt(&cfg.key_rx, srtcp_decrypt); + rtcp_loop::handle_rtcp( + pkt, + srtcp_decrypt.as_mut(), + &cfg.rx_stats, + &cfg.remote_report, + cfg.local_ssrc, + ); + return; + } + // RTP / DTMF — push to jitter. DTMF (rfc2833) classification happens // at pop time in the tick, since it needs access to Subsystems. if pkt.len() >= rtp::RTP_FIXED_HEADER_LEN { @@ -106,3 +140,50 @@ async fn handle_packet(cfg: &RecvLoopConfig, pkt: &[u8], peer: SocketAddr) { cfg.jitter.lock().push(rp); } } + +/// RFC 5761 §4 demux: on the shared RTP port, is this datagram RTCP rather +/// than RTP? True when the first byte is in the RTP/RTCP version range +/// (128..=191) and the second byte is an RTCP packet type — SR/RR/SDES/BYE/APP +/// (200..=204). Those values map to an RTP marker+payload-type of 72..=76, +/// which RTP deliberately never assigns, so the classification is unambiguous +/// and works on SRTP/SRTCP too (the header stays in cleartext). +fn is_muxed_rtcp(pkt: &[u8]) -> bool { + pkt.len() >= 2 && (128..=191).contains(&pkt[0]) && (200..=204).contains(&pkt[1]) +} + +#[cfg(test)] +mod tests { + use super::is_muxed_rtcp; + + #[test] + fn classifies_rtcp_packet_types() { + // Version-2 RTCP header (0x80) + each RTCP packet type. + for pt in 200u8..=204 { + assert!(is_muxed_rtcp(&[0x80, pt]), "PT {pt} should demux as RTCP"); + } + } + + #[test] + fn rtp_audio_is_not_mistaken_for_rtcp() { + // PCMU (pt 0), PCMA (8), G722 (9), rfc2833 (101) — with and without + // the marker bit. None of the second bytes fall in 200..=204. + for pt in [0u8, 8, 9, 101] { + assert!(!is_muxed_rtcp(&[0x80, pt]), "RTP pt {pt} misread as RTCP"); + let marked = 0x80 | pt; // marker bit set + assert!( + !is_muxed_rtcp(&[0x80, marked]), + "marked RTP pt {pt} misread as RTCP" + ); + } + } + + #[test] + fn rejects_out_of_range_and_short() { + assert!(!is_muxed_rtcp(&[0x80, 199]), "205- boundary below"); + assert!(!is_muxed_rtcp(&[0x80, 205]), "205 is above BYE/APP window"); + assert!(!is_muxed_rtcp(&[0x00, 200]), "STUN-range first byte"); + assert!(!is_muxed_rtcp(&[0x30, 200]), "DTLS-range first byte"); + assert!(!is_muxed_rtcp(&[0x80]), "too short"); + assert!(!is_muxed_rtcp(&[]), "empty"); + } +} diff --git a/rust/src/channel/rtcp_loop.rs b/rust/src/channel/rtcp_loop.rs index ba084ef..6e01e2f 100644 --- a/rust/src/channel/rtcp_loop.rs +++ b/rust/src/channel/rtcp_loop.rs @@ -17,8 +17,10 @@ use std::time::Instant; use parking_lot::Mutex as PLMutex; use tokio::net::UdpSocket; +use tokio::sync::watch; use tokio_util::sync::CancellationToken; +use super::dtls_session::{remote_srtp_params, SrtpKeyingMaterial}; use super::rtcp::{self, ReportBlock, RtcpItem}; use super::rtcp_stats::{RemoteReport, RxStats}; use super::rtp; @@ -30,6 +32,10 @@ pub struct RtcpLoopConfig { /// Our SSRC — a report block whose SSRC matches describes the stream *we* /// send, so it is the one the peer's loss/jitter/RTT figures are about. pub local_ssrc: u32, + /// DTLS keying material, published by the tick once the handshake + /// completes. `None` until then (and for non-secure channels), in which + /// case inbound RTCP is treated as cleartext. + pub key_rx: watch::Receiver>, pub cancel: CancellationToken, } @@ -39,13 +45,25 @@ pub fn spawn(cfg: RtcpLoopConfig) -> tokio::task::JoinHandle<()> { async fn run(cfg: RtcpLoopConfig) { let mut buf = [0u8; rtp::RTP_MAX_LENGTH]; + // Built lazily on the first packet after keying material arrives. Once + // present, all inbound RTCP on a secure channel is SRTCP. + let mut srtcp_decrypt: Option = None; loop { tokio::select! { biased; _ = cfg.cancel.cancelled() => break, result = cfg.sock.recv_from(&mut buf) => { match result { - Ok((n, _peer)) => handle_packet(&cfg, &buf[..n]), + Ok((n, _peer)) => { + maybe_build_decrypt(&cfg.key_rx, &mut srtcp_decrypt); + handle_rtcp( + &buf[..n], + srtcp_decrypt.as_mut(), + &cfg.rx_stats, + &cfg.remote_report, + cfg.local_ssrc, + ); + } Err(_) => break, } } @@ -53,8 +71,52 @@ async fn run(cfg: RtcpLoopConfig) { } } -fn handle_packet(cfg: &RtcpLoopConfig, pkt: &[u8]) { - // Malformed / non-RTCP compound — ignore (Tier 1 is best-effort). +/// Build the SRTCP decrypt context once the handshake has published keys. +/// No-op if already built or no keys yet. Shared with `recv_loop`, which +/// builds its own context for rtcp-mux'd inbound RTCP. +pub fn maybe_build_decrypt( + key_rx: &watch::Receiver>, + slot: &mut Option, +) { + if slot.is_some() { + return; + } + if let Some(keys) = key_rx.borrow().as_ref() { + let (key, salt, profile) = remote_srtp_params(keys); + if let Ok(ctx) = webrtc_srtp::context::Context::new(key, salt, profile, None, None) { + *slot = Some(ctx); + } + } +} + +/// Decrypt (if a context is present) then parse and fold an inbound RTCP +/// datagram into the shared accounting. Shared by this dedicated P+1 loop and, +/// under rtcp-mux (RFC 5761), by `recv_loop` reading off the RTP port. +pub fn handle_rtcp( + pkt: &[u8], + decrypt: Option<&mut webrtc_srtp::context::Context>, + rx_stats: &PLMutex, + remote_report: &PLMutex, + local_ssrc: u32, +) { + match decrypt { + // Auth failure / malformed SRTCP → decrypt errors, packet dropped. + Some(ctx) => { + if let Ok(plain) = ctx.decrypt_rtcp(pkt) { + parse_and_fold(&plain, rx_stats, remote_report, local_ssrc); + } + } + None => parse_and_fold(pkt, rx_stats, remote_report, local_ssrc), + } +} + +fn parse_and_fold( + pkt: &[u8], + rx_stats: &PLMutex, + remote_report: &PLMutex, + local_ssrc: u32, +) { + // Malformed / non-RTCP compound — ignore (best-effort). let items = match rtcp::parse(pkt) { Ok(items) => items, Err(_) => return, @@ -67,13 +129,13 @@ fn handle_packet(cfg: &RtcpLoopConfig, pkt: &[u8]) { for item in items { match item { RtcpItem::SenderReport { info, reports, .. } => { - cfg.rx_stats + rx_stats .lock() .note_sender_report(info.ntp_sec, info.ntp_frac, now); - fold_reports(cfg, &reports, now_ntp_mid); + fold_reports(remote_report, local_ssrc, &reports, now_ntp_mid); } RtcpItem::ReceiverReport { reports, .. } => { - fold_reports(cfg, &reports, now_ntp_mid); + fold_reports(remote_report, local_ssrc, &reports, now_ntp_mid); } RtcpItem::Bye { .. } => { // Peer signalled end of stream; teardown is driven elsewhere. @@ -82,10 +144,78 @@ fn handle_packet(cfg: &RtcpLoopConfig, pkt: &[u8]) { } } -fn fold_reports(cfg: &RtcpLoopConfig, reports: &[ReportBlock], now_ntp_mid: u32) { +fn fold_reports( + remote_report: &PLMutex, + local_ssrc: u32, + reports: &[ReportBlock], + now_ntp_mid: u32, +) { for rb in reports { - if rb.ssrc == cfg.local_ssrc { - cfg.remote_report.lock().update_from(rb, now_ntp_mid); + if rb.ssrc == local_ssrc { + remote_report.lock().update_from(rb, now_ntp_mid); } } } + +#[cfg(test)] +mod tests { + use super::*; + use webrtc_srtp::context::Context; + use webrtc_srtp::protection_profile::ProtectionProfile; + + // A paired encrypt/decrypt context sharing one key — the decrypt side + // recovers what the encrypt side protected. Mirrors the SRTP round-trip + // in `dtls_session::tests`, but for the RTCP (SRTCP) methods. + fn srtcp_pair() -> (Context, Context) { + let key = [0x11u8; 16]; + let salt = [0x22u8; 14]; + let profile = ProtectionProfile::Aes128CmHmacSha1_80; + let enc = Context::new(&key, &salt, profile, None, None).unwrap(); + let dec = Context::new(&key, &salt, profile, None, None).unwrap(); + (enc, dec) + } + + #[test] + fn srtcp_roundtrip_preserves_compound() { + // A plain RR + SDES compound (what `maybe_send_rtcp` emits pre-audio). + let compound = + rtcp::build_compound(0xDEAD_BEEF, None, &[], Some("abcd1234@127.0.0.1"), false); + let (mut enc, mut dec) = srtcp_pair(); + + let protected = enc.encrypt_rtcp(&compound).expect("encrypt_rtcp"); + assert_ne!(&protected[..], &compound[..], "compound must be encrypted"); + assert!( + protected.len() > compound.len(), + "SRTCP appends a 4-byte index + auth tag" + ); + + let recovered = dec.decrypt_rtcp(&protected).expect("decrypt_rtcp"); + assert_eq!( + &recovered[..], + &compound[..], + "round-trip must match byte-for-byte" + ); + + // ...and the recovered bytes still parse as the original RR. + let items = rtcp::parse(&recovered).expect("parse"); + assert!(matches!( + items.first(), + Some(RtcpItem::ReceiverReport { .. }) + )); + } + + #[test] + fn srtcp_rejects_tampered_packet() { + let compound = rtcp::build_compound(0x1234_5678, None, &[], Some("x@127.0.0.1"), false); + let (mut enc, mut dec) = srtcp_pair(); + + let mut protected = enc.encrypt_rtcp(&compound).expect("encrypt_rtcp").to_vec(); + let last = protected.len() - 1; + protected[last] ^= 0xFF; // corrupt the trailing auth tag + + assert!( + dec.decrypt_rtcp(&protected).is_err(), + "a tampered SRTCP packet must fail authentication" + ); + } +} diff --git a/rust/src/channel/rtcp_tx.rs b/rust/src/channel/rtcp_tx.rs index 043df0e..dce046b 100644 --- a/rust/src/channel/rtcp_tx.rs +++ b/rust/src/channel/rtcp_tx.rs @@ -4,7 +4,15 @@ // (Local `tick::run` and Mixed `Member` post-mix housekeeping). On the report // interval it assembles a compound datagram — an SR when we've sent audio, // otherwise an RR, always with an SDES/CNAME — and sends it on the P+1 control -// socket to the derived RTCP remote. +// socket to the derived RTCP remote. On a secure (DTLS-SRTP) channel the +// compound is protected as SRTCP first. +// +// Report timing is randomised per RFC 3550 §6.3.1 so reports from many +// channels don't synchronise. Full §6.2 multiparty reconsideration and +// bandwidth scaling are deliberately out of scope: projectrtp is strictly +// point-to-point (one local SSRC, one latched remote), so the member/sender +// and bandwidth terms stay below Tmin and Tmin is the effective interval — the +// part that matters here is the randomisation, which we do implement. use std::net::SocketAddr; use std::time::Instant; @@ -12,44 +20,74 @@ use std::time::Instant; use super::rtcp::{self, ReportBlock, SenderInfo}; use super::state::ChannelState; -/// Ticks between periodic reports: 250 × 20 ms ≈ 5 s. This is a fixed interval; -/// full RFC 3550 §6.2 randomised/bandwidth-scaled report timing is a deliberate -/// Tier 1 simplification. -pub const RTCP_INTERVAL_TICKS: u64 = 250; +/// Minimum report interval (RFC 3550 Tmin): 250 × 20 ms ≈ 5 s. +pub const RTCP_MIN_INTERVAL_TICKS: u64 = 250; + +/// RFC 3550 §6.3.1 interval-compensation divisor (e/2 - ... ≈ 1.21828). +const RTCP_COMPENSATION: f64 = 1.21828; -/// Emit a periodic SR/RR + SDES on the report interval. No-op off-interval or -/// until a remote address is known. +/// Emit a periodic SR/RR + SDES when the (randomised) report interval elapses. +/// No-op off-interval or until a remote address is known. pub async fn maybe_send_rtcp(state: &mut ChannelState) { - if !state.tick_count.is_multiple_of(RTCP_INTERVAL_TICKS) { + // First call for the channel: schedule the initial report at *half* the + // interval (§6.3.1 initial reconsideration) and return without sending. + if state.rtcp_next_tick == 0 { + let half = (next_interval_ticks(&mut state.rtcp_rng) / 2).max(1); + state.rtcp_next_tick = state.tick_count + half; return; } - - // Tier 1 emits plain RTP/AVP RTCP only. On an SRTP/DTLS channel that would - // be unencrypted (non-compliant SRTCP) and would leak SSRC/CNAME/counts in - // the clear, so stay silent until the Tier 2 SRTCP path lands. - if state.srtp_encrypt.is_some() { + if state.tick_count < state.rtcp_next_tick { return; } + // Reschedule now, with fresh jitter, whether or not this report actually + // goes out — otherwise, while we wait for a remote to be latched, the gate + // would pass every tick and busy-build reports. + state.rtcp_next_tick = state.tick_count + next_interval_ticks(&mut state.rtcp_rng); + + let Some(rtcp_remote) = rtcp_remote_addr(state) else { + return; + }; + let compound = build_report(state, false); + send_compound(state, &compound, rtcp_remote).await; +} - // Derive the RTCP remote: the RTP peer's IP with port + 1 (symmetric RTCP - // without mux). Skip entirely until the RTP peer is known. - let Some(rtp_remote) = state.get_remote_addr() else { +/// Send a single BYE compound on channel close so the peer learns the stream +/// has ended now rather than via RTP timeout. Best-effort — a dropped BYE just +/// falls back to the peer's own idle teardown. No-op if no remote is known. +pub async fn send_bye(state: &mut ChannelState) { + let Some(rtcp_remote) = rtcp_remote_addr(state) else { return; }; - let rtcp_remote = SocketAddr::new(rtp_remote.ip(), rtp_remote.port().wrapping_add(1)); + let compound = build_report(state, true); + send_compound(state, &compound, rtcp_remote).await; +} +/// The RTCP destination. Under rtcp-mux (RFC 5761) RTCP shares the RTP +/// 5-tuple, so it is the RTP peer itself; otherwise it is the RTP peer's IP +/// with port + 1 (symmetric RTCP on the separate control port). +fn rtcp_remote_addr(state: &ChannelState) -> Option { + let r = state.get_remote_addr()?; + if state.rtcpmux { + Some(r) + } else { + Some(SocketAddr::new(r.ip(), r.port().wrapping_add(1))) + } +} + +/// Build a compound report: SR when we've sent audio, otherwise RR; always +/// with an SDES/CNAME and a reception report about the peer if we've latched +/// its stream. `bye` appends a BYE sub-packet. +fn build_report(state: &ChannelState, bye: bool) -> Vec { let now = Instant::now(); let (ntp_sec, ntp_frac) = rtcp::ntp_now(); - // Reception report about the peer, if we've latched its stream yet. let report = state.rx_stats.lock().report_block(now); let reports: &[ReportBlock] = match &report { Some(rb) => std::slice::from_ref(rb), None => &[], }; - // SR once we've sent any audio, otherwise RR. - let compound = if state.out_count > 0 { + if state.out_count > 0 { let info = SenderInfo { ntp_sec, ntp_frac, @@ -57,10 +95,75 @@ pub async fn maybe_send_rtcp(state: &mut ChannelState) { packet_count: state.out_count as u32, octet_count: state.out_octets as u32, }; - rtcp::build_compound(state.ssrc, Some(&info), reports, Some(&state.cname), false) + rtcp::build_compound(state.ssrc, Some(&info), reports, Some(&state.cname), bye) } else { - rtcp::build_compound(state.ssrc, None, reports, Some(&state.cname), false) + rtcp::build_compound(state.ssrc, None, reports, Some(&state.cname), bye) + } +} + +/// Send a compound to `remote`, protecting it as SRTCP first on a secure +/// channel (same gate/context as the RTP send path). Under rtcp-mux the +/// datagram goes out on the RTP socket (shared 5-tuple), otherwise on the P+1 +/// control socket. +async fn send_compound(state: &mut ChannelState, compound: &[u8], remote: SocketAddr) { + // Clone the Arc up front so the socket borrow doesn't collide with the + // mutable `state.srtp_encrypt` borrow below. + let sock = if state.rtcpmux { + state.rtp_sock.clone() + } else { + state.rtcp_sock.clone() }; + if let Some(ref mut ctx) = state.srtp_encrypt { + if let Ok(protected) = ctx.encrypt_rtcp(compound) { + let _ = sock.send_to(&protected, remote).await; + } + } else { + let _ = sock.send_to(compound, remote).await; + } +} + +/// A randomised report interval in ticks: Tmin scaled uniformly over +/// [0.5, 1.5) then divided by the §6.3.1 compensation factor. +fn next_interval_ticks(rng: &mut u64) -> u64 { + let scale = 0.5 + next_unit(rng); // [0.5, 1.5) + let ticks = (RTCP_MIN_INTERVAL_TICKS as f64 * scale / RTCP_COMPENSATION) as u64; + ticks.max(1) +} - let _ = state.rtcp_sock.send_to(&compound, rtcp_remote).await; +/// xorshift64* → uniform f64 in [0, 1). Same RNG family as +/// `facade::rand_icepwd`; no `rand` crate dependency. `rng` is seeded non-zero +/// at channel construction. +fn next_unit(rng: &mut u64) -> f64 { + let mut x = *rng; + x ^= x >> 12; + x ^= x << 25; + x ^= x >> 27; + *rng = x; + let v = x.wrapping_mul(0x2545_F491_4F6C_DD1D); + (v >> 11) as f64 / (1u64 << 53) as f64 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn interval_stays_within_rfc_bounds() { + let mut rng = 0x1234_5678_9abc_def0u64; + let lo = (RTCP_MIN_INTERVAL_TICKS as f64 * 0.5 / RTCP_COMPENSATION) as u64; + let hi = (RTCP_MIN_INTERVAL_TICKS as f64 * 1.5 / RTCP_COMPENSATION) as u64; + for _ in 0..100_000 { + let t = next_interval_ticks(&mut rng); + assert!(t >= lo && t <= hi, "interval {t} out of [{lo},{hi}]"); + } + } + + #[test] + fn unit_is_in_unit_interval() { + let mut rng = 0x0fed_cba9_8765_4321u64; + for _ in 0..100_000 { + let u = next_unit(&mut rng); + assert!((0.0..1.0).contains(&u), "unit {u} out of [0,1)"); + } + } } diff --git a/rust/src/channel/srtp_ctx.rs b/rust/src/channel/srtp_ctx.rs deleted file mode 100644 index b45a0d5..0000000 --- a/rust/src/channel/srtp_ctx.rs +++ /dev/null @@ -1,56 +0,0 @@ -// SRTP context — pure-Rust via webrtc-srtp. -// -// Two streams per channel: one for outbound (protect) and one for inbound -// (unprotect). The webrtc_srtp::Context type is async, so protect/unprotect -// are async methods here — they take a mutable buffer and encrypt/decrypt in -// place (or return a new buffer; final shape chosen alongside tick.rs). -// -// Drop semantics are automatic: webrtc_srtp::Context holds only Rust-owned -// state (no libsrtp C context), so going out of scope cleans everything up. -// That's the whole point of preferring pure Rust here. - -use crate::channel::dtls_session::SrtpKeyingMaterial; - -#[allow(dead_code)] -pub struct SrtpContext { - _km: SrtpKeyingMaterial, - // Real webrtc_srtp::Context instances are created here once tick.rs wires - // them in. The Context API requires an async runtime to drive internal - // replay detection, so construction is deferred until the actor task is - // running. -} - -#[allow(dead_code)] -impl SrtpContext { - pub fn new(km: SrtpKeyingMaterial) -> Self { - Self { _km: km } - } - - // Protect an outbound RTP packet in place. Returns the new post-encryption - // length. TODO: wire to webrtc_srtp::Context::protect_rtp. - pub async fn protect_rtp(&mut self, _buf: &mut Vec) -> Result<(), SrtpError> { - Err(SrtpError::NotYetImplemented) - } - - pub async fn unprotect_rtp(&mut self, _buf: &mut Vec) -> Result<(), SrtpError> { - Err(SrtpError::NotYetImplemented) - } -} - -#[derive(Debug)] -#[allow(dead_code)] -pub enum SrtpError { - NotYetImplemented, - Crypto(String), -} - -impl std::fmt::Display for SrtpError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::NotYetImplemented => write!(f, "SRTP: not yet implemented"), - Self::Crypto(s) => write!(f, "SRTP crypto: {s}"), - } - } -} - -impl std::error::Error for SrtpError {} diff --git a/rust/src/channel/state.rs b/rust/src/channel/state.rs index f035484..a5ae8ad 100644 --- a/rust/src/channel/state.rs +++ b/rust/src/channel/state.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use parking_lot::Mutex as PLMutex; use tokio::net::UdpSocket; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, watch}; use tokio_util::sync::CancellationToken; use super::actor::Event; @@ -42,6 +42,15 @@ pub struct ChannelState { pub remote_report: Arc>, /// Canonical name emitted in SDES; stable for the channel's lifetime. pub cname: String, + /// Tick at which the next periodic RTCP report is due. 0 = not yet + /// scheduled (the first report is planned on the first `maybe_send_rtcp`). + pub rtcp_next_tick: u64, + /// Per-channel xorshift state for randomising the report interval + /// (RFC 3550 §6.3.1). Seeded non-zero at construction. + pub rtcp_rng: u64, + /// RFC 5761 rtcp-mux: RTCP rides the RTP port/5-tuple rather than P+1. + /// Latched from the `remote()` config; false = classic split ports. + pub rtcpmux: bool, #[allow(dead_code)] pub out_pool: Vec, @@ -87,6 +96,9 @@ pub struct ChannelState { pub srtp_keys: Option, pub srtp_encrypt: Option, pub srtp_decrypt: Option, + /// Publishes DTLS keying material to the inbound RTCP loop once the + /// handshake completes, so it can build its SRTCP decrypt context. + pub srtp_key_tx: Option>>, } impl ChannelState { @@ -109,6 +121,16 @@ impl ChannelState { rx_stats: Arc::new(PLMutex::new(RxStats::new(DEFAULT_CLOCK_RATE))), remote_report: Arc::new(PLMutex::new(RemoteReport::default())), cname: format!("{ssrc:08x}@{}", local_addr.ip()), + rtcp_next_tick: 0, + // Seed from wall-clock nanos XOR ssrc; force non-zero (xorshift + // stays stuck at 0). Mirrors facade::rand_ssrc's time source. + rtcp_rng: (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0) + ^ ((ssrc as u64) << 32)) + | 1, + rtcpmux: false, out_pool: Vec::new(), out_sn: 0, out_ts: 0, @@ -136,6 +158,7 @@ impl ChannelState { srtp_keys: None, srtp_encrypt: None, srtp_decrypt: None, + srtp_key_tx: None, } } diff --git a/rust/src/channel/tick.rs b/rust/src/channel/tick.rs index 3ab0d81..c5fcf96 100644 --- a/rust/src/channel/tick.rs +++ b/rust/src/channel/tick.rs @@ -524,30 +524,24 @@ pub(crate) fn poll_dtls_handshake(state: &mut ChannelState) { result.profile, result.is_client, ); - let (our_key, our_salt) = if keys.local_is_server { - (&keys.server_write_key, &keys.server_write_salt) - } else { - (&keys.client_write_key, &keys.client_write_salt) - }; - let (their_key, their_salt) = if keys.local_is_server { - (&keys.client_write_key, &keys.client_write_salt) - } else { - (&keys.server_write_key, &keys.server_write_salt) - }; + let (our_key, our_salt, profile) = super::dtls_session::local_srtp_params(&keys); if let Ok(enc) = - webrtc_srtp::context::Context::new(our_key, our_salt, keys.profile, None, None) + webrtc_srtp::context::Context::new(our_key, our_salt, profile, None, None) { state.srtp_encrypt = Some(enc); } - if let Ok(dec) = webrtc_srtp::context::Context::new( - their_key, - their_salt, - keys.profile, - None, - None, - ) { + let (their_key, their_salt, _) = super::dtls_session::remote_srtp_params(&keys); + if let Ok(dec) = + webrtc_srtp::context::Context::new(their_key, their_salt, profile, None, None) + { state.srtp_decrypt = Some(dec); } + // Hand the keying material to the inbound RTCP loop so it can + // build its own SRTCP decrypt context (SRTP/SRTCP replay state + // is independent, so a dedicated context is correct here). + if let Some(tx) = &state.srtp_key_tx { + let _ = tx.send(Some(keys.clone())); + } state.srtp_keys = Some(keys); state.dtls_result_rx = None; state.dtls_handshake_abort = None; diff --git a/test/interface/projectrtprtcp.js b/test/interface/projectrtprtcp.js index 1728b6e..6f74211 100644 --- a/test/interface/projectrtprtcp.js +++ b/test/interface/projectrtprtcp.js @@ -66,7 +66,7 @@ describe( "rtcp", function() { it( "emits SR + SDES on P+1 and reflects a peer RR in the close stats", async function() { - /* RTCP first fires at tick 250 (~5s); allow headroom. */ + /* Randomised first report fires ~1–3 s (half a randomised interval); headroom. */ this.timeout( 9000 ) this.slow( 8000 ) @@ -144,6 +144,121 @@ describe( "rtcp", function() { expect( r.rttms ).to.equal( null ) } ) + it( "carries RTCP over the RTP port when rtcp-mux is negotiated (RFC 5761)", async function() { + + /* Randomised first report fires ~1-3s; keep the P+1-test headroom. */ + this.timeout( 9000 ) + this.slow( 8000 ) + + const rtp = dgram.createSocket( "udp4" ) + const rtcp = dgram.createSocket( "udp4" ) /* the P+1 port — must stay silent under mux */ + + /* Resolve on the first RTCP compound seen *on the RTP port* — demux by the + RTCP packet-type byte (200..=204), ignoring the echoed PCMU audio. */ + let resolvesr + const gotsr = new Promise( ( res ) => { resolvesr = res } ) + rtp.on( "message", ( m ) => { + if( 2 <= m.length && 200 <= m[ 1 ] && 204 >= m[ 1 ] ) resolvesr( m ) + } ) + + let p1count = 0 + rtcp.on( "message", () => { p1count++ } ) + + await new Promise( ( res ) => rtp.bind( res ) ) + const peerport = rtp.address().port + await new Promise( ( res, rej ) => + rtcp.bind( peerport + 1, ( e ) => ( e ? rej( e ) : res() ) ) ) + + let closestats + let resolveclose + const closed = new Promise( ( res ) => { resolveclose = res } ) + + const channel = await projectrtp.openchannel( + { "remote": { "address": "127.0.0.1", "port": peerport, "codec": 0, "rtcpmux": true } }, + function( d ) { + if( "close" === d.action ) { + closestats = d.stats + resolveclose() + } + } ) + + expect( channel.echo() ).to.be.true + for( let i = 0; 50 > i; i++ ) sendpk( i, channel.local.port, rtp ) + + /* The channel's first RTCP compound — arriving on the RTP port, not P+1. */ + const sr = await gotsr + + expect( sr[ 0 ] >> 6 ).to.equal( 2 ) /* RTP version 2 */ + const items = walkrtcp( sr ) + expect( items[ 0 ].pt ).to.equal( 200 ) /* first sub-packet is an SR */ + expect( items.some( ( it ) => 202 === it.pt ) ).to.be.true /* SDES present */ + expect( sr.readUInt32BE( 4 ) ).to.equal( channel.local.ssrc >>> 0 ) + + /* Feed a crafted RR back over the *same* RTP port (mux) about our stream; + the recv_loop must demux it to the RTCP path and fold it. */ + rtp.send( + buildrr( 25, channel.local.ssrc, 25, 12, 40 ), + channel.local.port, "127.0.0.1" ) + + await new Promise( ( r ) => setTimeout( r, 200 ) ) + channel.close() + await closed + + rtp.close() + rtcp.close() + + /* The muxed RR was folded and surfaced in the close stats. */ + expect( closestats ).to.have.property( "rtcp" ) + expect( closestats.rtcp.out.valid ).to.equal( true ) + expect( closestats.rtcp.out.fractionlost ).to.equal( 25 ) + expect( closestats.rtcp.out.cumulativelost ).to.equal( 12 ) + expect( closestats.rtcp.out.jitter ).to.equal( 40 ) + + /* Nothing should ever land on the separate P+1 control port under mux. */ + expect( p1count ).to.equal( 0 ) + } ) + + it( "sends an RTCP BYE (PT 203) on channel close", async function() { + + this.timeout( 4000 ) + this.slow( 3000 ) + + const rtp = dgram.createSocket( "udp4" ) + const rtcp = dgram.createSocket( "udp4" ) + rtp.on( "message", () => {} ) /* drain echoed audio */ + + await new Promise( ( res ) => rtp.bind( res ) ) + const peerport = rtp.address().port + await new Promise( ( res, rej ) => + rtcp.bind( peerport + 1, ( e ) => ( e ? rej( e ) : res() ) ) ) + + /* Resolve on the first compound that carries a BYE sub-packet. Closing + early (before the first periodic report) means the BYE is the only + datagram, but the filter is robust either way. */ + let resolvebye + const gotbye = new Promise( ( res ) => { resolvebye = res } ) + rtcp.on( "message", ( m ) => { + if( walkrtcp( m ).some( ( it ) => 203 === it.pt ) ) resolvebye( m ) + } ) + + const channel = await projectrtp.openchannel( + { "remote": { "address": "127.0.0.1", "port": peerport, "codec": 0 } }, + function() {} ) + + /* Feed a few packets so the channel latches the remote address, then close + — the BYE is emitted on the close path. */ + expect( channel.echo() ).to.be.true + for( let i = 0; 10 > i; i++ ) sendpk( i, channel.local.port, rtp ) + await new Promise( ( r ) => setTimeout( r, 300 ) ) + channel.close() + + const bye = await gotbye + expect( walkrtcp( bye ).some( ( it ) => 203 === it.pt ) ).to.be.true + + rtp.close() + rtcp.close() + } ) + it( "populates in.skip and lowers MOS when inbound packets are lost", function( done ) { const peer = dgram.createSocket( "udp4" ) diff --git a/test/interface/projectrtprtcpsecure.js b/test/interface/projectrtprtcpsecure.js new file mode 100644 index 0000000..38bde27 --- /dev/null +++ b/test/interface/projectrtprtcpsecure.js @@ -0,0 +1,109 @@ + + +const expect = require( "chai" ).expect +const fs = require( "fs" ) + +const projectrtp = require( "../../index.js" ).projectrtp + +/* Standalone safety net, mirroring projectrtprtcp.js: run() is idempotent so a + second call in the full suite is a no-op. projectrtpserver.js owns teardown. */ +before( () => { projectrtp.run() } ) + +/* Minimal PCM16 mono WAV writer — a self-contained sine source so this file + doesn't depend on codecchain.js's helpers. */ +function writetonewav( path, freqHz = 400, durationSec = 2.0, sampleRate = 8000, amplitude = 0.5 ) { + const total = Math.floor( sampleRate * durationSec ) + const peak = Math.round( 32767 * amplitude ) + const data = Buffer.alloc( total * 2 ) + const w = 2 * Math.PI * freqHz / sampleRate + for( let i = 0; i < total; i++ ) data.writeInt16LE( Math.round( Math.sin( i * w ) * peak ), i * 2 ) + + const header = Buffer.alloc( 44 ) + header.write( "RIFF", 0 ) + header.writeUInt32LE( 36 + data.length, 4 ) + header.write( "WAVE", 8 ) + header.write( "fmt ", 12 ) + header.writeUInt32LE( 16, 16 ) + header.writeUInt16LE( 1, 20 ) /* PCM */ + header.writeUInt16LE( 1, 22 ) /* mono */ + header.writeUInt32LE( sampleRate, 24 ) + header.writeUInt32LE( sampleRate * 2, 28 ) + header.writeUInt16LE( 2, 32 ) /* block align */ + header.writeUInt16LE( 16, 34 ) + header.write( "data", 36 ) + header.writeUInt32LE( data.length, 40 ) + + fs.writeFileSync( path, Buffer.concat( [ header, data ] ) ) +} + +describe( "rtcp secure (SRTCP)", function() { + + const wavpath = "/tmp/rtcp_srtcp_tone.wav" + before( () => { writetonewav( wavpath ) } ) + after( () => { try { fs.unlinkSync( wavpath ) } catch( _ ) { /* ignore */ } } ) + + it( "protects RTCP as SRTCP over DTLS and folds the peer's decrypted reports", async function() { + + /* Randomised reports: first fires ~1–3 s, RTT needs a second exchange; + 6.5 s below leaves headroom for both directions. */ + this.timeout( 12000 ) + this.slow( 10000 ) + + let statsA, statsB + let resolveA, resolveB + const closedA = new Promise( ( r ) => { resolveA = r } ) + const closedB = new Promise( ( r ) => { resolveB = r } ) + + const chanA = await projectrtp.openchannel( {}, ( d ) => { + if( "close" === d.action ) { statsA = d.stats; resolveA() } + } ) + const chanB = await projectrtp.openchannel( {}, ( d ) => { + if( "close" === d.action ) { statsB = d.stats; resolveB() } + } ) + + /* chanA is the DTLS client (active), chanB the server (passive). Each is + given the other's fingerprint. This is the same secure topology as the + codecchain.js DTLS-SRTP test, but here we assert the RTCP path. */ + expect( chanA.remote( { + address: "127.0.0.1", + port: chanB.local.port, + codec: 0, + dtls: { fingerprint: { hash: chanB.local.dtls.fingerprint }, mode: "active" }, + } ) ).to.be.true + + expect( chanB.remote( { + address: "127.0.0.1", + port: chanA.local.port, + codec: 0, + dtls: { fingerprint: { hash: chanA.local.dtls.fingerprint }, mode: "passive" }, + } ) ).to.be.true + + /* chanA plays a looping tone (RTP A→B); chanB echoes it back (RTP B→A). + Both directions carry SRTP media, so both sides latch the other's SSRC + and both accumulate out_count > 0 — so each emits an SR that carries a + reception report about the other. */ + expect( chanB.echo() ).to.be.true + await new Promise( ( r ) => setTimeout( r, 300 ) ) /* settle the handshake */ + expect( chanA.play( { loop: true, files: [ { wav: wavpath } ] } ) ).to.be.true + + /* Wait past the first RTCP interval so each side has sent an SR and the + other has received, authenticated, decrypted and folded it. */ + await new Promise( ( r ) => setTimeout( r, 6500 ) ) + + chanA.close() + chanB.close() + await Promise.all( [ closedA, closedB ] ) + + /* out.valid === true means the peer's SR — SRTCP-encrypted on the wire — + was received, passed auth, decrypted, and its report block about *us* + folded into RemoteReport. That is the end-to-end proof the SRTCP path + works both ways. (rttms stays null until the peer echoes one of our SRs, + which needs two intervals, so we allow null-or-number here.) */ + for( const [ name, s ] of [ [ "A", statsA ], [ "B", statsB ] ] ) { + expect( s, `${name} close stats` ).to.have.property( "rtcp" ) + expect( s.rtcp.out.valid, `${name} peer report not valid — SRTCP decrypt failed?` ).to.equal( true ) + expect( s.rtcp.in.cumulativelost, `${name} in.cumulativelost` ).to.be.a( "number" ) + expect( s.rtcp.rttms, `${name} rttms` ).to.satisfy( ( v ) => null === v || "number" === typeof v ) + } + } ) +} )