1use std::ffi::OsString;2use std::net::{IpAddr, Ipv4Addr, SocketAddr};3use std::process::Stdio;4use std::sync::Arc;56use anyhow::{Context, Result, ensure};7use tokio::net::UdpSocket;8use tokio::process::Command;9use tokio::sync::mpsc;10use tracing::debug;1112use crate::clock::{self, SessionClock};13use crate::config::HostConfig;14use crate::source::{15 log_stderr_chunks, preflight_gstreamer_element, resolve_pipewire_node_by_properties,16 validate_executable,17};18use sigil_protocol::{MAX_AUDIO_PAYLOAD_LEN, OPUS_SAMPLE_RATE};1920const RTP_PAYLOAD_TYPE: u8 = 111;21const AUDIO_QUEUE_CAPACITY: usize = 2;22const RTP_DATAGRAM_CAPACITY: usize = 2_048;232425pub struct EncodedAudioPacket {26 pub sequence: u64,27 /// Post-encode observation in the shared media-session clock.28 pub capture_timestamp_us: u64,29 /// First-packet common-clock anchor plus elapsed RTP samples.30 pub pts_us: i64,31 pub discontinuity: bool,32 pub payload: Arc<[u8]>,33}3435pub type EncodedAudioSource = (36 mpsc::Receiver<EncodedAudioPacket>,37 tokio::task::JoinHandle<Result<()>>,38);3940pub async fn preflight_audio_static(config: &HostConfig) -> Result<()> {41 config.validate()?;42 let audio = config43 .audio44 .as_ref()45 .context("audio configuration is missing")?;46 let pipewire = config47 .gamescope_pipewire48 .as_ref()49 .context("gamescope_pipewire configuration is missing")?;50 validate_executable("pw_dump_path", &pipewire.pw_dump_path)?;51 validate_executable("gst_launch_path", &pipewire.gst_launch_path)?;52 validate_executable("gst_inspect_path", &pipewire.gst_inspect_path)?;53 for element in [54 "pipewiresrc",55 "queue",56 "audioconvert",57 "audioresample",58 "opusenc",59 "rtpopuspay",60 "udpsink",61 ] {62 preflight_gstreamer_element(&pipewire.gst_inspect_path, element).await?;63 }64 ensure!(65 audio.bitrate_bps == 96_000,66 "audio config changed after validation"67 );68 Ok(())69}7071pub async fn preflight_audio(config: &HostConfig) -> Result<String> {72 preflight_audio_static(config).await?;73 let audio = config74 .audio75 .as_ref()76 .context("audio configuration is missing")?;77 let pipewire = config78 .gamescope_pipewire79 .as_ref()80 .context("gamescope_pipewire configuration is missing")?;81 resolve_pipewire_node_by_properties(82 &pipewire.pw_dump_path,83 &audio.node_name,84 &audio.media_class,85 &audio.match_properties,86 )87 .await88}8990pub async fn spawn_pipewire_audio(91 config: HostConfig,92 session_clock: SessionClock,93) -> Result<EncodedAudioSource> {94 let audio = config95 .audio96 .as_ref()97 .context("audio configuration is missing")?;98 let pipewire = config99 .gamescope_pipewire100 .as_ref()101 .context("gamescope_pipewire configuration is missing")?;102 let target_object = resolve_pipewire_node_by_properties(103 &pipewire.pw_dump_path,104 &audio.node_name,105 &audio.media_class,106 &audio.match_properties,107 )108 .await?;109 let socket = UdpSocket::bind(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0))110 .await111 .context("binding bounded audio RTP receiver")?;112 let port = socket113 .local_addr()114 .context("reading audio RTP address")?115 .port();116 let mut ssrc_bytes = [0_u8; 4];117 getrandom::fill(&mut ssrc_bytes).context("generating audio RTP SSRC")?;118 let ssrc = u32::from_be_bytes(ssrc_bytes);119 let args = audio_pipeline_args(&config, &target_object, port, ssrc)?;120 let (sender, receiver) = mpsc::channel(AUDIO_QUEUE_CAPACITY);121 let task = tokio::spawn(async move {122 run_pipewire_audio(config, session_clock, args, socket, ssrc, sender).await123 });124 Ok((receiver, task))125}126127async fn run_pipewire_audio(128 config: HostConfig,129 session_clock: SessionClock,130 args: Vec<OsString>,131 socket: UdpSocket,132 expected_ssrc: u32,133 sender: mpsc::Sender<EncodedAudioPacket>,134) -> Result<()> {135 let pipewire = config136 .gamescope_pipewire137 .as_ref()138 .context("gamescope_pipewire configuration is missing")?;139 let mut command = Command::new(&pipewire.gst_launch_path);140 command141 .args(args)142 .env("LC_ALL", "C")143 .stdin(Stdio::null())144 .stdout(Stdio::null())145 .stderr(Stdio::piped())146 .kill_on_drop(true);147 let mut child = command.spawn().with_context(|| {148 format!(149 "starting audio pipeline with {}",150 pipewire.gst_launch_path.display()151 )152 })?;153 let stderr = child154 .stderr155 .take()156 .context("audio GStreamer stderr was not piped")?;157 let stderr_task = tokio::spawn(log_stderr_chunks(stderr, "gstreamer-audio"));158159 let result = receive_rtp(socket, expected_ssrc, session_clock, sender).await;160 let _ = child.kill().await;161 let status = child.wait().await.ok();162 stderr_task.abort();163 match (result, status) {164 (Err(error), Some(status)) => {165 Err(error).with_context(|| format!("audio GStreamer pipeline exited with {status}"))166 }167 (result, _) => result,168 }169}170171async fn receive_rtp(172 socket: UdpSocket,173 expected_ssrc: u32,174 session_clock: SessionClock,175 sender: mpsc::Sender<EncodedAudioPacket>,176) -> Result<()> {177 let mut datagram = [0_u8; RTP_DATAGRAM_CAPACITY];178 let mut sequence = 0_u64;179 let mut previous_rtp_sequence: Option<u16> = None;180 let mut rtp_clock = RtpClock::default();181 let mut pending_discontinuity = false;182183 loop {184 let (length, source) = socket185 .recv_from(&mut datagram)186 .await187 .context("receiving audio RTP")?;188 ensure!(189 source.ip().is_loopback(),190 "audio RTP arrived from a non-loopback address"191 );192 let packet = parse_rtp_opus(&datagram[..length], expected_ssrc)?;193 let rtp_discontinuity = previous_rtp_sequence194 .is_some_and(|previous| previous.wrapping_add(1) != packet.sequence);195 previous_rtp_sequence = Some(packet.sequence);196 let observed_us = session_clock.now_micros();197 let pts_us = rtp_clock.pts_us(packet.timestamp, observed_us)?;198 let encoded = EncodedAudioPacket {199 sequence,200 capture_timestamp_us: observed_us,201 pts_us,202 discontinuity: pending_discontinuity || rtp_discontinuity,203 payload: Arc::from(packet.payload),204 };205 sequence = sequence.saturating_add(1);206 match sender.try_send(encoded) {207 Ok(()) => pending_discontinuity = false,208 Err(mpsc::error::TrySendError::Full(_)) => {209 pending_discontinuity = true;210 debug!("dropping encoded audio packet because bounded queue is full");211 }212 Err(mpsc::error::TrySendError::Closed(_)) => return Ok(()),213 }214 }215}216217fn audio_pipeline_args(218 config: &HostConfig,219 target_object: &str,220 port: u16,221 ssrc: u32,222) -> Result<Vec<OsString>> {223 let audio = config224 .audio225 .as_ref()226 .context("audio configuration is missing")?;227 ensure!(228 !target_object.is_empty()229 && target_object.len() <= 32230 && target_object.bytes().all(|byte| byte.is_ascii_digit()),231 "resolved PipeWire audio object.serial is invalid"232 );233 let args = [234 "--quiet".to_owned(),235 "pipewiresrc".to_owned(),236 "do-timestamp=true".to_owned(),237 "min-buffers=1".to_owned(),238 "max-buffers=3".to_owned(),239 format!("target-object={target_object}"),240 "!".to_owned(),241 "queue".to_owned(),242 "max-size-buffers=3".to_owned(),243 "max-size-bytes=0".to_owned(),244 "max-size-time=0".to_owned(),245 "leaky=downstream".to_owned(),246 "!".to_owned(),247 "audioconvert".to_owned(),248 "!".to_owned(),249 "audioresample".to_owned(),250 "!".to_owned(),251 "audio/x-raw,format=S16LE,rate=48000,channels=2".to_owned(),252 "!".to_owned(),253 "opusenc".to_owned(),254 "audio-type=restricted-lowdelay".to_owned(),255 format!("bitrate={}", audio.bitrate_bps),256 "bitrate-type=cbr".to_owned(),257 "frame-size=20".to_owned(),258 "dtx=false".to_owned(),259 "inband-fec=false".to_owned(),260 format!("max-payload-size={MAX_AUDIO_PAYLOAD_LEN}"),261 "!".to_owned(),262 "rtpopuspay".to_owned(),263 format!("pt={RTP_PAYLOAD_TYPE}"),264 "mtu=600".to_owned(),265 format!("ssrc={ssrc}"),266 "!".to_owned(),267 "udpsink".to_owned(),268 "host=127.0.0.1".to_owned(),269 format!("port={port}"),270 "sync=false".to_owned(),271 "async=false".to_owned(),272 ];273 Ok(args.into_iter().map(OsString::from).collect())274}275276277struct RtpOpusPacket<'a> {278 sequence: u16,279 timestamp: u32,280 payload: &'a [u8],281}282283fn parse_rtp_opus(datagram: &[u8], expected_ssrc: u32) -> Result<RtpOpusPacket<'_>> {284 ensure!(285 datagram.len() >= 12,286 "RTP datagram is shorter than its fixed header"287 );288 ensure!(datagram[0] >> 6 == 2, "unsupported RTP version");289 let padding = datagram[0] & 0x20 != 0;290 let extension = datagram[0] & 0x10 != 0;291 let csrc_count = usize::from(datagram[0] & 0x0f);292 ensure!(293 datagram[1] & 0x7f == RTP_PAYLOAD_TYPE,294 "unexpected RTP payload type"295 );296 ensure!(297 u32::from_be_bytes(datagram[8..12].try_into().expect("fixed slice length"))298 == expected_ssrc,299 "unexpected RTP SSRC"300 );301 let mut payload_start = 12_usize302 .checked_add(303 csrc_count304 .checked_mul(4)305 .context("RTP CSRC length overflow")?,306 )307 .context("RTP header length overflow")?;308 ensure!(payload_start <= datagram.len(), "truncated RTP CSRC list");309 if extension {310 ensure!(311 payload_start + 4 <= datagram.len(),312 "truncated RTP extension header"313 );314 let words = usize::from(u16::from_be_bytes(315 datagram[payload_start + 2..payload_start + 4]316 .try_into()317 .expect("fixed slice length"),318 ));319 payload_start = payload_start320 .checked_add(4)321 .and_then(|value| value.checked_add(words.checked_mul(4)?))322 .context("RTP extension length overflow")?;323 ensure!(324 payload_start <= datagram.len(),325 "truncated RTP extension data"326 );327 }328 let padding_len = if padding {329 usize::from(*datagram.last().context("RTP padding byte is missing")?)330 } else {331 0332 };333 ensure!(!padding || padding_len > 0, "RTP padding length is zero");334 ensure!(335 padding_len <= datagram.len().saturating_sub(payload_start),336 "invalid RTP padding"337 );338 let payload_end = datagram.len() - padding_len;339 let payload = &datagram[payload_start..payload_end];340 ensure!(!payload.is_empty(), "RTP Opus payload is empty");341 ensure!(342 payload.len() <= MAX_AUDIO_PAYLOAD_LEN,343 "RTP Opus payload exceeds protocol limit"344 );345 Ok(RtpOpusPacket {346 sequence: u16::from_be_bytes(datagram[2..4].try_into().expect("fixed slice length")),347 timestamp: u32::from_be_bytes(datagram[4..8].try_into().expect("fixed slice length")),348 payload,349 })350}351352353struct RtpClock {354 previous: Option<u32>,355 elapsed_samples: u64,356 anchor_micros: Option<u64>,357}358359impl RtpClock {360 fn pts_us(&mut self, timestamp: u32, observed_micros: u64) -> Result<i64> {361 if let Some(previous) = self.previous {362 let delta = timestamp.wrapping_sub(previous);363 // One packet may be absent, but a jump beyond ten seconds is a364 // reset or malicious timestamp rather than useful audio timing.365 ensure!(366 delta <= OPUS_SAMPLE_RATE * 10,367 "RTP timestamp advanced beyond the bounded clock window"368 );369 self.elapsed_samples = self.elapsed_samples.saturating_add(u64::from(delta));370 }371 self.previous = Some(timestamp);372 let anchor_micros = *self.anchor_micros.get_or_insert(observed_micros);373 Ok(clock::anchored_timestamp_micros(374 anchor_micros,375 self.elapsed_samples,376 OPUS_SAMPLE_RATE,377 ))378 }379}380381382mod tests {383 use std::collections::BTreeMap;384 use std::path::PathBuf;385386 use super::*;387 use crate::config::{388 GamescopeEncoderBackend, GamescopePipewireConfig, InputMode, PipewireAudioConfig,389 VaapiRateControl, VideoSource,390 };391 use crate::source::resolve_pipewire_node_exact;392393 fn config() -> HostConfig {394 HostConfig {395 identity_path: PathBuf::from("/tmp/id"),396 state_path: PathBuf::from("/tmp/state"),397 max_viewers: crate::config::DEFAULT_MAX_VIEWERS,398 focus_owner: None,399 source: VideoSource::GamescopePipewire,400 width: Some(1280),401 height: Some(800),402 framerate: 60,403 codec: "h264".into(),404 input_mode: InputMode::Disabled,405 uinput: None,406 ffmpeg_path: "ffmpeg".into(),407 gamescope_pipewire: Some(GamescopePipewireConfig {408 xwayland_display: None,409 node_name: "gamescope".into(),410 media_class: "Video/Source".into(),411 match_properties: BTreeMap::new(),412 pw_dump_path: "/usr/bin/pw-dump".into(),413 gst_launch_path: "/usr/bin/gst-launch-1.0".into(),414 gst_inspect_path: "/usr/bin/gst-inspect-1.0".into(),415 encoder_backend: GamescopeEncoderBackend::ExternalGstLaunch,416 vaapi_encoder: "vah264enc".into(),417 vaapi_render_node: "/dev/dri/renderD128".into(),418 rate_control: VaapiRateControl::Cbr,419 bitrate_kbps: Some(12_000),420 quantizer: None,421 }),422 audio: Some(PipewireAudioConfig {423 node_name: "sigil-game-audio".into(),424 media_class: "Audio/Sink".into(),425 match_properties: BTreeMap::new(),426 bitrate_bps: 96_000,427 }),428 }429 }430431 fn rtp(sequence: u16, timestamp: u32, ssrc: u32, payload: &[u8]) -> Vec<u8> {432 let mut datagram = vec![0x80, RTP_PAYLOAD_TYPE];433 datagram.extend_from_slice(&sequence.to_be_bytes());434 datagram.extend_from_slice(×tamp.to_be_bytes());435 datagram.extend_from_slice(&ssrc.to_be_bytes());436 datagram.extend_from_slice(payload);437 datagram438 }439440 441 fn pipeline_is_fixed_bounded_and_loopback_only() {442 let args: Vec<String> = audio_pipeline_args(&config(), "50", 34_567, 123)443 .unwrap()444 .into_iter()445 .map(|arg| arg.into_string().unwrap())446 .collect();447 for required in [448 "target-object=50",449 "max-size-buffers=3",450 "max-size-bytes=0",451 "max-size-time=0",452 "leaky=downstream",453 "audio/x-raw,format=S16LE,rate=48000,channels=2",454 "audio-type=restricted-lowdelay",455 "bitrate=96000",456 "bitrate-type=cbr",457 "frame-size=20",458 "dtx=false",459 "inband-fec=false",460 "max-payload-size=512",461 "host=127.0.0.1",462 "port=34567",463 "ssrc=123",464 "sync=false",465 "async=false",466 ] {467 assert!(args.contains(&required.to_owned()), "missing {required}");468 }469 assert!(audio_pipeline_args(&config(), "gamescope", 1, 1).is_err());470 }471472 473 fn resolves_exactly_one_audio_sink_without_accepting_a_microphone() {474 let dump = br#"[475 {"type":"PipeWire:Interface:Node","info":{"props":{476 "node.name":"sigil-game-audio","media.class":"Audio/Sink",477 "device.profile.name":"stereo","object.serial":50}}},478 {"type":"PipeWire:Interface:Node","info":{"props":{479 "node.name":"sigil-game-audio","media.class":"Audio/Source",480 "device.profile.name":"stereo","object.serial":51}}}481 ]"#;482 let properties = BTreeMap::from([("device.profile.name".into(), "stereo".into())]);483 assert_eq!(484 resolve_pipewire_node_exact(dump, "sigil-game-audio", "Audio/Sink", &properties)485 .unwrap(),486 "50"487 );488489 let duplicated = String::from_utf8(dump.to_vec())490 .unwrap()491 .replace("]", ",{\"type\":\"PipeWire:Interface:Node\",\"info\":{\"props\":{\"node.name\":\"sigil-game-audio\",\"media.class\":\"Audio/Sink\",\"device.profile.name\":\"stereo\",\"object.serial\":52}}}]");492 assert!(493 resolve_pipewire_node_exact(494 duplicated.as_bytes(),495 "sigil-game-audio",496 "Audio/Sink",497 &properties498 )499 .is_err()500 );501 }502503 504 fn parses_strict_bounded_rtp_and_rejects_wrong_identity() {505 let datagram = rtp(7, 960, 123, &[1, 2, 3]);506 assert_eq!(507 parse_rtp_opus(&datagram, 123).unwrap(),508 RtpOpusPacket {509 sequence: 7,510 timestamp: 960,511 payload: &[1, 2, 3],512 }513 );514 assert!(parse_rtp_opus(&datagram, 124).is_err());515 let mut wrong_payload_type = datagram.clone();516 wrong_payload_type[1] = 110;517 assert!(parse_rtp_opus(&wrong_payload_type, 123).is_err());518 let mut wrong_version = datagram;519 wrong_version[0] = 0;520 assert!(parse_rtp_opus(&wrong_version, 123).is_err());521 }522523 524 fn rtp_parser_handles_csrc_extension_and_padding_safely() {525 let mut datagram = vec![0xb1, RTP_PAYLOAD_TYPE]; // padding, extension, one CSRC526 datagram.extend_from_slice(&1_u16.to_be_bytes());527 datagram.extend_from_slice(&2_u32.to_be_bytes());528 datagram.extend_from_slice(&3_u32.to_be_bytes());529 datagram.extend_from_slice(&4_u32.to_be_bytes()); // CSRC530 datagram.extend_from_slice(&0xbede_u16.to_be_bytes());531 datagram.extend_from_slice(&1_u16.to_be_bytes());532 datagram.extend_from_slice(&[0, 0, 0, 0]);533 datagram.extend_from_slice(&[9, 8, 7]);534 datagram.extend_from_slice(&[0, 0, 3]);535 assert_eq!(parse_rtp_opus(&datagram, 3).unwrap().payload, &[9, 8, 7]);536537 for length in 0..12 {538 assert!(parse_rtp_opus(&datagram[..length], 3).is_err());539 }540 let mut bad_padding = datagram;541 *bad_padding.last_mut().unwrap() = u8::MAX;542 assert!(parse_rtp_opus(&bad_padding, 3).is_err());543 }544545 546 fn rtp_clock_handles_wrap_and_rejects_implausible_jumps() {547 let mut clock = RtpClock::default();548 assert_eq!(clock.pts_us(u32::MAX - 479, 1_000_000).unwrap(), 1_000_000);549 assert_eq!(clock.pts_us(480, 9_000_000).unwrap(), 1_020_000);550 assert_eq!(clock.pts_us(1_440, 9_000_000).unwrap(), 1_040_000);551 assert!(552 clock553 .pts_us(1_440 + OPUS_SAMPLE_RATE * 11, 9_000_000)554 .is_err()555 );556 }557558 559 fn rejects_oversized_and_empty_rtp_payloads() {560 assert!(parse_rtp_opus(&rtp(0, 0, 1, &[]), 1).is_err());561 assert!(parse_rtp_opus(&rtp(0, 0, 1, &vec![0; MAX_AUDIO_PAYLOAD_LEN + 1]), 1).is_err());562 }563}