feat(rtcp): Tier 2 — SRTCP + randomised timing + BYE + rtcp-mux

.gitignore

@@ -123,7 +123,7 @@ dist
 .tern-port
 
 # Local working notes (RTCP implementation handover, etc.)
-handover.md
+handover*.md
 
 # Rust
 rust/target/

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 ]

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;

rust/src/channel/commands.rs

@@ -15,6 +15,10 @@ pub struct RemoteConfig {
     pub ilbc_payload_type: Option<u8>,
     pub rfc2833_payload_type: Option<u8>,
     pub dtls: Option<RemoteDtls>,
+    /// 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

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::*;

rust/src/channel/facade.rs

@@ -322,6 +322,10 @@ impl ChannelObject {
         let port = params.get_named_property::<u32>("port").ok();
         let codec = params.get_named_property::<u32>("codec").ok().unwrap_or(0);
         let icepwd = params.get_named_property::<String>("icepwd").ok();
+        let rtcpmux = params
+            .get_named_property::<bool>("rtcpmux")
+            .ok()
+            .unwrap_or(false);
         let dtls = parse_remote_dtls(&params);
         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<String> {
         .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::<Object>("remote")
+        .ok()
+        .and_then(|r| r.get_named_property::<bool>("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<Ch
     let override_local_icepwd = extract_local_icepwd(&params);
     let initial_remote_icepwd = extract_remote_icepwd(&params);
     let initial_remote_dtls = extract_remote_dtls(&params);
+    let initial_rtcpmux = extract_rtcpmux(&params);
     let initial_direction = extract_direction(&params);
     let mut tsfn: ThreadsafeFunction<EventPayload, ErrorStrategy::Fatal> = callback
         .create_threadsafe_function(
@@ -1245,6 +1261,7 @@ pub fn open_channel(env: Env, params: Object, callback: JsFunction) -> Result<Ch
             ilbc_payload_type: Some(ilbc_pt),
             rfc2833_payload_type: rfc2833_pt,
             dtls: initial_remote_dtls.clone(),
+            rtcpmux: initial_rtcpmux,
             icepwd: initial_remote_icepwd.clone(),
         };
         let _ = handle.cmd.try_send(Command::Remote { cfg, ack });

rust/src/channel/mixer.rs

@@ -817,7 +817,7 @@ async fn mix_tick(members: &mut HashMap<ChannelId, Box<Member>>) {
     }
 
     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<ChannelId, Box<Member>>) {
+async fn close_idle_members(members: &mut HashMap<ChannelId, Box<Member>>) {
     let idle_ids: Vec<ChannelId> = 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");
         }
     }

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;

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<AtomicU64>,
     /// RFC 3550 receiver accounting — fed on every inbound RTP packet.
     pub rx_stats: Arc<PLMutex<RxStats>>,
+    /// 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<PLMutex<RemoteReport>>,
+    /// Our SSRC — selects the muxed report block that is about our stream.
+    pub local_ssrc: u32,
     pub local_icepwd: Arc<PLMutex<String>>,
     pub dtls_tx: Arc<PLMutex<Option<mpsc::Sender<Vec<u8>>>>>,
+    /// DTLS keying material for decrypting muxed SRTCP; `None` until the
+    /// handshake completes (and always, for non-secure channels).
+    pub key_rx: watch::Receiver<Option<SrtpKeyingMaterial>>,
     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<webrtc_srtp::context::Context> = 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<webrtc_srtp::context::Context>,
+) {
     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");
+    }
+}