diff --git a/synctv-api-common/src/impls/mod.rs b/synctv-api-common/src/impls/mod.rs index 5291bd6d..35ee3a51 100644 --- a/synctv-api-common/src/impls/mod.rs +++ b/synctv-api-common/src/impls/mod.rs @@ -821,6 +821,9 @@ pub fn map_livestream_stream_error(stream_error: &StreamError) -> ApiError { StreamError::PermissionDenied(_) => { ApiError::Authorization(LIVESTREAM_PERMISSION_DENIED_MESSAGE.to_string()) } + StreamError::Authentication(_) => { + ApiError::Authentication("Publish key is invalid or expired".to_string()) + } StreamError::ResourceExhausted(_) => { ApiError::RateLimited(LIVESTREAM_RATE_LIMITED_MESSAGE.to_string()) } diff --git a/synctv-api-http/src/http/livestream_webrtc.rs b/synctv-api-http/src/http/livestream_webrtc.rs index 1852030d..ed02ab91 100644 --- a/synctv-api-http/src/http/livestream_webrtc.rs +++ b/synctv-api-http/src/http/livestream_webrtc.rs @@ -539,6 +539,14 @@ mod tests { assert_eq!(error.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE); } + #[test] + fn invalid_publish_key_maps_to_unauthorized() { + let error = map_stream_error(&StreamError::Authentication( + "invalid publish key".to_string(), + )); + assert_eq!(error.status(), StatusCode::UNAUTHORIZED); + } + #[test] fn session_locations_use_canonical_routes() { let path = LiveWebRtcPath { diff --git a/synctv-common/src/ssrf.rs b/synctv-common/src/ssrf.rs index cd5439e3..716dbfb0 100644 --- a/synctv-common/src/ssrf.rs +++ b/synctv-common/src/ssrf.rs @@ -178,6 +178,15 @@ pub struct SsrfGuard { inner: Option>, } +impl std::fmt::Debug for SsrfGuard { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("SsrfGuard") + .field("enabled", &self.inner.is_some()) + .finish_non_exhaustive() + } +} + struct SsrfGuardInner { acl: HttpAcl, resolver: Arc, @@ -378,6 +387,18 @@ impl SsrfGuard { .is_some_and(|inner| inner.policy.is_host_blocked(host)) } + /// Check whether a hostname may be used without policy-aware DNS resolution. + /// + /// Protocols that cannot inject [`Self::dns_resolver`] must reject arbitrary + /// hostnames to prevent DNS rebinding. Disabled policies and explicitly + /// allowlisted hosts remain usable. + #[must_use] + pub fn allows_unresolved_host(&self, host: &str) -> bool { + self.inner + .as_ref() + .is_none_or(|inner| inner.policy.allowed_hosts.contains(&normalize_host(host))) + } + /// Check if a port is blocked for a concrete IP target. /// /// When private-network targets or explicit IP ranges are allowed, their @@ -799,6 +820,20 @@ mod tests { assert!(guard.dns_resolver().is_none()); assert!(!guard.is_host_blocked("localhost")); assert!(!guard.is_ip_blocked(&IpAddr::V4(Ipv4Addr::LOCALHOST))); + assert!(guard.allows_unresolved_host("peer.local")); + } + + #[test] + fn test_unresolved_hosts_require_an_explicit_allowlist() { + let strict = SsrfGuard::strict_policy(); + assert!(!strict.allows_unresolved_host("peer.local")); + assert!(!strict.allows_unresolved_host("example.com")); + + let allowlisted = SsrfGuard::builder() + .extra_allowed_host("ICE.EXAMPLE.COM.".to_string()) + .build(); + assert!(allowlisted.allows_unresolved_host("ice.example.com")); + assert!(!allowlisted.allows_unresolved_host("other.example.com")); } #[test] diff --git a/synctv-livestream/src/error.rs b/synctv-livestream/src/error.rs index cecc0be2..2d77a904 100644 --- a/synctv-livestream/src/error.rs +++ b/synctv-livestream/src/error.rs @@ -35,6 +35,9 @@ pub enum StreamError { #[error("Permission denied: {0}")] PermissionDenied(String), + #[error("Authentication failed: {0}")] + Authentication(String), + #[error("Invalid state: {0}")] InvalidState(String), diff --git a/synctv-livestream/src/grpc/stream_relay_service.rs b/synctv-livestream/src/grpc/stream_relay_service.rs index 41aff5f5..fdca1e74 100644 --- a/synctv-livestream/src/grpc/stream_relay_service.rs +++ b/synctv-livestream/src/grpc/stream_relay_service.rs @@ -96,6 +96,7 @@ fn map_webrtc_session_error(error: StreamError) -> Status { match error { StreamError::InvalidInput(message) => Status::invalid_argument(message), StreamError::PermissionDenied(message) => Status::permission_denied(message), + StreamError::Authentication(message) => Status::unauthenticated(message), StreamError::InvalidState(message) => Status::failed_precondition(message), StreamError::ResourceExhausted(message) => Status::resource_exhausted(message), StreamError::RegistryError(message) => Status::unavailable(message), diff --git a/synctv-livestream/src/grpc/webrtc_session_router.rs b/synctv-livestream/src/grpc/webrtc_session_router.rs index 44f365a8..5cd41ead 100644 --- a/synctv-livestream/src/grpc/webrtc_session_router.rs +++ b/synctv-livestream/src/grpc/webrtc_session_router.rs @@ -78,6 +78,7 @@ impl WebRtcSessionRouter { StreamError::InvalidInput(message) } tonic::Code::PermissionDenied => StreamError::PermissionDenied(message), + tonic::Code::Unauthenticated => StreamError::Authentication(message), tonic::Code::FailedPrecondition | tonic::Code::AlreadyExists | tonic::Code::Aborted => { StreamError::InvalidState(message) } @@ -88,7 +89,6 @@ impl WebRtcSessionRouter { } tonic::Code::Cancelled | tonic::Code::DeadlineExceeded - | tonic::Code::Unauthenticated | tonic::Code::Unimplemented | tonic::Code::Unavailable => StreamError::ConnectionFailed(message), tonic::Code::Ok => StreamError::Internal( @@ -243,7 +243,7 @@ mod tests { } let request = request.into_inner(); if request.publish_token == "denied" { - return Err(Status::permission_denied( + return Err(Status::unauthenticated( "WHIP session credentials do not match", )); } @@ -330,7 +330,7 @@ mod tests { Some("denied"), ) .await; - assert!(matches!(denied, Err(StreamError::PermissionDenied(_)))); + assert!(matches!(denied, Err(StreamError::Authentication(_)))); cancel.cancel(); tokio::time::timeout(Duration::from_secs(1), server).await???; diff --git a/synctv-livestream/src/livestream/webrtc_session_manager.rs b/synctv-livestream/src/livestream/webrtc_session_manager.rs index d3013080..80602628 100644 --- a/synctv-livestream/src/livestream/webrtc_session_manager.rs +++ b/synctv-livestream/src/livestream/webrtc_session_manager.rs @@ -10,7 +10,7 @@ use dashmap::DashMap; use subtle::ConstantTimeEq; use synctv_xiu::{ rtmp::{ - auth::{AuthCallback, RtmpStreamMode}, + auth::{AuthCallback, PublishAuthError, RtmpStreamMode}, session::common::RtmpStreamHandler, }, streamhub::{ @@ -193,9 +193,8 @@ impl WebRtcSessionManager { WebRtcError::EmptySdp | WebRtcError::SdpTooLarge { .. } | WebRtcError::InvalidSdp(_) - | WebRtcError::IncompatibleWhipVideoCodec => { - StreamError::InvalidInput(error.to_string()) - } + | WebRtcError::IncompatibleWhipVideoCodec + | WebRtcError::NoCompatibleWhipMedia => StreamError::InvalidInput(error.to_string()), WebRtcError::Negotiation(_) | WebRtcError::IceGatheringTimeout(_) | WebRtcError::MissingLocalDescription => { @@ -297,7 +296,13 @@ impl WebRtcSessionManager { let rewrite = auth .on_publish(generation_id, room_id, media_id, Some(&auth_query)) .await - .map_err(|error| StreamError::PermissionDenied(error.to_string()))?; + .map_err(|error| { + if let Some(auth_error) = error.downcast_ref::() { + StreamError::Authentication(auth_error.to_string()) + } else { + StreamError::PermissionDenied(error.to_string()) + } + })?; let (room_id, media_id, media_mode) = rewrite.map_or_else( || { ( @@ -772,7 +777,7 @@ impl WebRtcSessionManager { .append_pair("token", token) .finish(); if !bool::from(auth_query.as_bytes().ct_eq(supplied_auth_query.as_bytes())) { - return Err(StreamError::PermissionDenied( + return Err(StreamError::Authentication( "WHIP session credentials do not match".to_string(), )); } @@ -836,6 +841,8 @@ mod tests { struct VideoOnlyAuth; + struct InvalidPublishKeyAuth; + #[async_trait::async_trait] impl AuthCallback for VideoOnlyAuth { async fn on_publish( @@ -865,6 +872,31 @@ mod tests { } } + #[async_trait::async_trait] + impl AuthCallback for InvalidPublishKeyAuth { + async fn on_publish( + &self, + _generation_id: Uuid, + _app_name: &str, + _stream_name: &str, + _query: Option<&str>, + ) -> Result< + Option, + Box, + > { + Err(Box::new(PublishAuthError::new("invalid publish key"))) + } + + async fn on_play( + &self, + _app_name: &str, + _stream_name: &str, + _query: Option<&str>, + ) -> Result<(), Box> { + Ok(()) + } + } + #[test] fn capacity_is_bounded() { let (event_sender, _) = tokio::sync::mpsc::channel(1); @@ -904,4 +936,20 @@ mod tests { assert_eq!(auth_query, "token=secret"); assert_eq!(media_mode, RtmpStreamMode::VideoOnly); } + + #[tokio::test] + async fn publish_key_errors_remain_authentication_failures() { + let (event_sender, _) = tokio::sync::mpsc::channel(1); + let manager = WebRtcSessionManager::new( + event_sender, + Some(Arc::new(InvalidPublishKeyAuth)), + WebRtcSessionConfig::default(), + ); + + let error = manager + .authenticate_publish(Uuid::new(), "public-room", "public-media", "invalid") + .await + .expect_err("invalid publish key should fail authentication"); + assert!(matches!(error, StreamError::Authentication(_))); + } } diff --git a/synctv-xiu/src/rtmp/auth.rs b/synctv-xiu/src/rtmp/auth.rs index 5796178b..08fd56be 100644 --- a/synctv-xiu/src/rtmp/auth.rs +++ b/synctv-xiu/src/rtmp/auth.rs @@ -1,4 +1,5 @@ use async_trait::async_trait; +use thiserror::Error; use crate::streamhub::utils::Uuid; @@ -10,6 +11,22 @@ pub enum RtmpStreamMode { AudioOnly, } +/// Authentication failure returned by a publish-key validator. +#[derive(Debug, Error)] +#[error("{message}")] +pub struct PublishAuthError { + message: String, +} + +impl PublishAuthError { + #[must_use] + pub fn new(message: impl Into) -> Self { + Self { + message: message.into(), + } + } +} + /// Optional rewrite of RTMP identifiers returned by [`AuthCallback::on_publish`]. /// /// When the auth callback resolves a JWT token in `stream_name` to a logical diff --git a/synctv-xiu/src/webrtc/peer.rs b/synctv-xiu/src/webrtc/peer.rs index eb71fd1a..ffdba7f2 100644 --- a/synctv-xiu/src/webrtc/peer.rs +++ b/synctv-xiu/src/webrtc/peer.rs @@ -1,10 +1,11 @@ -use std::{io::Cursor, sync::Arc, time::Duration}; +use std::{io::Cursor, net::IpAddr, sync::Arc, time::Duration}; use bytes::Bytes; use sdp::description::session::{ SessionDescription, ATTR_KEY_INACTIVE, ATTR_KEY_RECV_ONLY, ATTR_KEY_SEND_ONLY, ATTR_KEY_SEND_RECV, }; +use synctv_common::ssrf::SsrfGuard; use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; use webrtc::{ @@ -77,6 +78,7 @@ pub struct WebRtcConfig { pub ice_servers: Vec, pub ice_gathering_timeout: Duration, pub max_sdp_bytes: usize, + pub ssrf_guard: SsrfGuard, } impl Default for WebRtcConfig { @@ -85,6 +87,7 @@ impl Default for WebRtcConfig { ice_servers: Vec::new(), ice_gathering_timeout: Duration::from_secs(10), max_sdp_bytes: 256 * 1024, + ssrf_guard: SsrfGuard::strict_policy(), } } } @@ -99,6 +102,10 @@ pub enum WebRtcError { InvalidSdp(String), #[error("WHIP offer has no H.264 format compatible with the livestream bridge")] IncompatibleWhipVideoCodec, + #[error( + "WHIP offer has no sendable Opus or H.264 media compatible with the livestream bridge" + )] + NoCompatibleWhipMedia, #[error("WebRTC negotiation failed: {0}")] Negotiation(String), #[error("ICE gathering timed out after {0:?}")] @@ -118,12 +125,13 @@ pub struct WhepClientSession { peer_connection: Arc, cancel_token: CancellationToken, max_sdp_bytes: usize, + ssrf_guard: SsrfGuard, } impl WhepClientSession { pub async fn apply_answer(&self, answer_sdp: &str) -> Result<(), WebRtcError> { - validate_sdp(answer_sdp, self.max_sdp_bytes)?; - let answer = RTCSessionDescription::answer(answer_sdp.to_string()) + let answer_sdp = sanitize_remote_sdp(answer_sdp, self.max_sdp_bytes, &self.ssrf_guard)?; + let answer = RTCSessionDescription::answer(answer_sdp) .map_err(|error| WebRtcError::InvalidSdp(error.to_string()))?; self.peer_connection .set_remote_description(answer) @@ -204,8 +212,68 @@ fn validate_sdp(sdp: &str, max_sdp_bytes: usize) -> Result<(), WebRtcError> { Ok(()) } -fn validate_offer(offer_sdp: &str, config: &WebRtcConfig) -> Result<(), WebRtcError> { - validate_sdp(offer_sdp, config.max_sdp_bytes) +fn candidate_target_allowed(candidate: &str, ssrf_guard: &SsrfGuard) -> Result { + let fields = candidate.split_whitespace().collect::>(); + if fields.len() < 6 { + return Err(WebRtcError::InvalidSdp( + "ICE candidate has too few fields".to_string(), + )); + } + let address = fields[4]; + fields[5].parse::().map_err(|error| { + WebRtcError::InvalidSdp(format!("ICE candidate has an invalid port: {error}")) + })?; + if let Ok(ip) = address.parse::() { + return Ok(!ssrf_guard.is_ip_blocked(&ip)); + } + Ok(!ssrf_guard.is_host_blocked(&address) && ssrf_guard.allows_unresolved_host(&address)) +} + +fn filter_remote_candidates( + attributes: &mut Vec, + ssrf_guard: &SsrfGuard, +) -> Result<(), WebRtcError> { + let mut filtered = Vec::with_capacity(attributes.len()); + for attribute in std::mem::take(attributes) { + if !attribute.key.eq_ignore_ascii_case("candidate") { + filtered.push(attribute); + continue; + } + let candidate = attribute.value.as_deref().ok_or_else(|| { + WebRtcError::InvalidSdp("ICE candidate attribute has no value".to_string()) + })?; + if candidate_target_allowed(candidate, ssrf_guard)? { + filtered.push(attribute); + } + } + *attributes = filtered; + Ok(()) +} + +fn sanitize_remote_sdp( + sdp: &str, + max_sdp_bytes: usize, + ssrf_guard: &SsrfGuard, +) -> Result { + validate_sdp(sdp, max_sdp_bytes)?; + let mut reader = Cursor::new(sdp.as_bytes()); + let mut session = SessionDescription::unmarshal(&mut reader) + .map_err(|error| WebRtcError::InvalidSdp(error.to_string()))?; + filter_remote_candidates(&mut session.attributes, ssrf_guard)?; + for media in &mut session.media_descriptions { + filter_remote_candidates(&mut media.attributes, ssrf_guard)?; + } + Ok(session.to_string()) +} + +fn validate_offer( + offer_sdp: &str, + config: &WebRtcConfig, +) -> Result { + let sanitized = sanitize_remote_sdp(offer_sdp, config.max_sdp_bytes, &config.ssrf_guard)?; + let mut reader = Cursor::new(sanitized.as_bytes()); + SessionDescription::unmarshal(&mut reader) + .map_err(|error| WebRtcError::InvalidSdp(error.to_string())) } fn remote_sends_media( @@ -261,25 +329,31 @@ fn validate_whip_offer( offer_sdp: &str, config: &WebRtcConfig, media_mode: RtmpStreamMode, -) -> Result<(), WebRtcError> { - validate_offer(offer_sdp, config)?; - if !whip_accepts_track(media_mode, RTPCodecType::Video) { - return Ok(()); - } - let mut reader = Cursor::new(offer_sdp.as_bytes()); - let session = SessionDescription::unmarshal(&mut reader) - .map_err(|error| WebRtcError::InvalidSdp(error.to_string()))?; - for media in session.media_descriptions.iter().filter(|media| { - media.media_name.media.eq_ignore_ascii_case("video") - && media.media_name.port.value != 0 - && remote_sends_media(&session, media) - }) { +) -> Result { + let session = validate_offer(offer_sdp, config)?; + let mut has_compatible_media = false; + for media in session + .media_descriptions + .iter() + .filter(|media| media.media_name.port.value != 0 && remote_sends_media(&session, media)) + { + let kind = if media.media_name.media.eq_ignore_ascii_case("audio") { + RTPCodecType::Audio + } else if media.media_name.media.eq_ignore_ascii_case("video") { + RTPCodecType::Video + } else { + continue; + }; + if !whip_accepts_track(media_mode, kind) { + continue; + } let media_session = SessionDescription { media_descriptions: vec![media.clone()], ..Default::default() }; let mut offered_h264 = false; let mut compatible_h264 = false; + let mut compatible_opus = false; for payload in &media.media_name.formats { let payload_type = payload .parse::() @@ -289,14 +363,26 @@ fn validate_whip_offer( .map_err(|error| WebRtcError::InvalidSdp(error.to_string()))?; if codec.name.eq_ignore_ascii_case("H264") { offered_h264 = true; - compatible_h264 |= h264_fmtp_is_bridge_compatible(&codec.fmtp); + compatible_h264 |= + codec.clock_rate == 90_000 && h264_fmtp_is_bridge_compatible(&codec.fmtp); + } else if codec.name.eq_ignore_ascii_case("opus") { + let channels = codec.encoding_parameters.parse::().unwrap_or(0); + compatible_opus |= codec.clock_rate == 48_000 && matches!(channels, 1 | 2); } } - if offered_h264 && !compatible_h264 { + if kind == RTPCodecType::Video && offered_h264 && !compatible_h264 { return Err(WebRtcError::IncompatibleWhipVideoCodec); } + has_compatible_media |= match kind { + RTPCodecType::Audio => compatible_opus, + RTPCodecType::Video => compatible_h264, + RTPCodecType::Unspecified => false, + }; } - Ok(()) + if !has_compatible_media { + return Err(WebRtcError::NoCompatibleWhipMedia); + } + Ok(session) } fn peer_configuration(config: &WebRtcConfig) -> RTCConfiguration { @@ -529,6 +615,7 @@ pub async fn create_whep_client_session( peer_connection: peer, cancel_token, max_sdp_bytes: config.max_sdp_bytes, + ssrf_guard: config.ssrf_guard.clone(), }) } @@ -663,7 +750,7 @@ pub async fn create_whip_session( media_mode: RtmpStreamMode, config: &WebRtcConfig, ) -> Result { - validate_whip_offer(offer_sdp, config, media_mode)?; + let offer_sdp = validate_whip_offer(offer_sdp, config, media_mode)?.to_string(); let peer = create_peer_connection(config).await?; let cancel_token = CancellationToken::new(); bind_connection_lifecycle(&peer, cancel_token.clone()); @@ -708,7 +795,7 @@ pub async fn create_whip_session( }) })); - let answer_sdp = match negotiate_answer(&peer, offer_sdp, config).await { + let answer_sdp = match negotiate_answer(&peer, &offer_sdp, config).await { Ok(answer) => answer, Err(error) => { cancel_token.cancel(); @@ -798,7 +885,7 @@ pub async fn create_whep_session( packet_receiver: PacketDataReceiver, config: &WebRtcConfig, ) -> Result { - validate_offer(offer_sdp, config)?; + let offer_sdp = validate_offer(offer_sdp, config)?.to_string(); let peer = create_peer_connection(config).await?; let cancel_token = CancellationToken::new(); bind_connection_lifecycle(&peer, cancel_token.clone()); @@ -826,7 +913,7 @@ pub async fn create_whep_session( }); } - let answer_sdp = match negotiate_answer(&peer, offer_sdp, config).await { + let answer_sdp = match negotiate_answer(&peer, &offer_sdp, config).await { Ok(answer) => answer, Err(error) => { cancel_token.cancel(); @@ -914,6 +1001,68 @@ mod tests { )); } + #[test] + fn remote_sdp_filters_candidates_blocked_by_ssrf_policy() { + const OFFER: &str = "v=0\r\n\ +o=- 1 1 IN IP4 127.0.0.1\r\n\ +s=-\r\n\ +t=0 0\r\n\ +a=candidate:1 1 UDP 2122260223 127.0.0.1 50000 typ host\r\n\ +m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n\ +a=sendonly\r\n\ +a=rtpmap:111 opus/48000/2\r\n\ +a=candidate:2 1 UDP 2122260222 192.168.1.20 50001 typ host\r\n\ +a=candidate:3 1 UDP 2122260221 peer.local 50002 typ host\r\n\ +a=candidate:4 1 UDP 2122260220 8.8.8.8 50003 typ srflx\r\n"; + + let sanitized = sanitize_remote_sdp( + OFFER, + WebRtcConfig::default().max_sdp_bytes, + &SsrfGuard::strict_policy(), + ) + .expect("valid SDP should be sanitized"); + + assert!(!sanitized.contains("127.0.0.1 50000")); + assert!(!sanitized.contains("192.168.1.20 50001")); + assert!(!sanitized.contains("peer.local 50002")); + assert!(sanitized.contains("8.8.8.8 50003")); + } + + #[test] + fn remote_sdp_preserves_private_candidates_when_policy_allows_them() { + const OFFER: &str = "v=0\r\n\ +o=- 1 1 IN IP4 127.0.0.1\r\n\ +s=-\r\n\ +t=0 0\r\n\ +m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n\ +a=sendonly\r\n\ +a=rtpmap:111 opus/48000/2\r\n\ +a=candidate:1 1 UDP 2122260223 192.168.1.20 50001 typ host\r\n"; + let guard = SsrfGuard::builder() + .allow_private_network_targets(true) + .build(); + + let sanitized = sanitize_remote_sdp(OFFER, 256 * 1024, &guard) + .expect("explicitly allowed private candidate should be retained"); + assert!(sanitized.contains("192.168.1.20 50001")); + } + + #[test] + fn remote_sdp_rejects_malformed_candidates() { + const OFFER: &str = "v=0\r\n\ +o=- 1 1 IN IP4 127.0.0.1\r\n\ +s=-\r\n\ +t=0 0\r\n\ +m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n\ +a=rtpmap:111 opus/48000/2\r\n\ +a=candidate:too-short\r\n"; + + assert!(matches!( + sanitize_remote_sdp(OFFER, 256 * 1024, &SsrfGuard::strict_policy()), + Err(WebRtcError::InvalidSdp(_)) + )); + } + #[test] fn whip_offer_rejects_h264_that_cannot_be_relayed_as_the_bridge_profile() { const HIGH_PROFILE_OFFER: &str = "v=0\r\n\ @@ -933,12 +1082,14 @@ a=fmtp:96 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640c1f ), Err(WebRtcError::IncompatibleWhipVideoCodec) )); - assert!(validate_whip_offer( - HIGH_PROFILE_OFFER, - &WebRtcConfig::default(), - RtmpStreamMode::AudioOnly, - ) - .is_ok()); + assert!(matches!( + validate_whip_offer( + HIGH_PROFILE_OFFER, + &WebRtcConfig::default(), + RtmpStreamMode::AudioOnly, + ), + Err(WebRtcError::NoCompatibleWhipMedia) + )); } #[test] @@ -976,6 +1127,50 @@ a=fmtp:98 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e029 .is_ok()); } + #[test] + fn whip_offer_requires_a_sendable_bridge_codec() { + const VP8_ONLY_OFFER: &str = "v=0\r\n\ +o=- 1 1 IN IP4 127.0.0.1\r\n\ +s=-\r\n\ +t=0 0\r\n\ +m=video 9 UDP/TLS/RTP/SAVPF 96\r\n\ +a=sendonly\r\n\ +a=rtpmap:96 VP8/90000\r\n"; + const INACTIVE_H264_OFFER: &str = "v=0\r\n\ +o=- 1 1 IN IP4 127.0.0.1\r\n\ +s=-\r\n\ +t=0 0\r\n\ +m=video 9 UDP/TLS/RTP/SAVPF 96\r\n\ +a=inactive\r\n\ +a=rtpmap:96 H264/90000\r\n\ +a=fmtp:96 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f\r\n"; + + for offer in [VP8_ONLY_OFFER, INACTIVE_H264_OFFER] { + assert!(matches!( + validate_whip_offer(offer, &WebRtcConfig::default(), RtmpStreamMode::Default,), + Err(WebRtcError::NoCompatibleWhipMedia) + )); + } + } + + #[test] + fn whip_offer_accepts_sendable_opus_for_audio_mode() { + const OPUS_OFFER: &str = "v=0\r\n\ +o=- 1 1 IN IP4 127.0.0.1\r\n\ +s=-\r\n\ +t=0 0\r\n\ +m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n\ +a=sendonly\r\n\ +a=rtpmap:111 opus/48000/2\r\n"; + + assert!(validate_whip_offer( + OPUS_OFFER, + &WebRtcConfig::default(), + RtmpStreamMode::AudioOnly, + ) + .is_ok()); + } + #[test] fn builds_configured_ice_server() { let config = WebRtcConfig { @@ -1103,6 +1298,7 @@ a=fmtp:98 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e029 async fn whip_to_whep_forwards_rtp_and_emits_avc_frames() -> Result<()> { let config = WebRtcConfig { ice_gathering_timeout: Duration::from_secs(5), + ssrf_guard: SsrfGuard::disabled(), ..WebRtcConfig::default() }; let (frame_sender, mut frame_receiver) = mpsc::channel(256); diff --git a/synctv/src/bootstrap/livestream.rs b/synctv/src/bootstrap/livestream.rs index 306c1cb2..bb9fa371 100644 --- a/synctv/src/bootstrap/livestream.rs +++ b/synctv/src/bootstrap/livestream.rs @@ -185,6 +185,7 @@ pub async fn init_livestream( config.livestream.webrtc.ice_gathering_timeout_seconds, ), max_sdp_bytes: config.livestream.webrtc.max_sdp_bytes, + ssrf_guard: config.security.ssrf_guard(), }, max_sessions: config.livestream.webrtc.max_sessions, max_session_duration: std::time::Duration::from_secs( diff --git a/synctv/src/rtmp_auth/mod.rs b/synctv/src/rtmp_auth/mod.rs index 231fee1f..bbfc3f74 100644 --- a/synctv/src/rtmp_auth/mod.rs +++ b/synctv/src/rtmp_auth/mod.rs @@ -25,7 +25,7 @@ use synctv_adapter::PublicIdCodec; use synctv_core::{ models::{MediaId, Room, RoomId, RoomStatus, UserId, UserStatus}, service::{RoomService, StreamingPublishKeyService, UserService}, - RedisConnectionRuntime, SharedStateMode, SharedStateProfile, + Error as CoreError, RedisConnectionRuntime, SharedStateMode, SharedStateProfile, }; use synctv_livestream::{ PublisherControlHandle, PublisherStopOutcome, PublisherStopRequest, StreamRegistryTrait, @@ -33,12 +33,21 @@ use synctv_livestream::{ }; // TTL for the per-user rtmp:user_stream:{user_id} Redis key, matching the publisher TTL. use synctv_xiu::rtmp::auth::{ - AuthCallback, AuthPublishRewrite, RtmpStreamMode as XiuRtmpStreamMode, + AuthCallback, AuthPublishRewrite, PublishAuthError, RtmpStreamMode as XiuRtmpStreamMode, }; use synctv_xiu::streamhub::utils::Uuid; const STREAMHUB_RESTARTING_MESSAGE: &str = "StreamHub is restarting, please retry in a few seconds"; +fn map_publish_key_validation_error(error: CoreError) -> Box { + match error { + error @ CoreError::Authentication(_) => Box::new(PublishAuthError::new(format!( + "Invalid stream key: {error}" + ))), + error => Box::new(error), + } +} + #[async_trait] pub trait UserStreamIndex: Send + Sync { async fn put( @@ -858,7 +867,9 @@ impl SyncTvRtmpAuth { // is reserved for the media binding and must not be overloaded as a token. let token_owned: Option = query.and_then(extract_token_from_query); let token = token_owned.as_deref().ok_or_else(|| { - "Missing token query parameter; RTMP publish must use ?token=".to_string() + PublishAuthError::new( + "Missing token query parameter; RTMP publish must use ?token=", + ) })?; // Validate JWT stream_key @@ -866,7 +877,7 @@ impl SyncTvRtmpAuth { .publish_key_service .validate_publish_key_for_stream_claims(token, &expected_room_id, &expected_media_id) .await - .map_err(|e| format!("Invalid stream key: {e}"))?; + .map_err(map_publish_key_validation_error)?; // Re-verify user status at connection time let user_id = claims @@ -1046,6 +1057,23 @@ mod tests { StreamRegistryTrait, }; + #[test] + fn publish_key_error_mapping_only_marks_authentication_failures() { + let authentication = map_publish_key_validation_error(CoreError::Authentication( + "publish key expired".to_string(), + )); + assert!(authentication.downcast_ref::().is_some()); + + for error in [ + CoreError::Authorization("publish key has insufficient scope".to_string()), + CoreError::Internal("publish-key store unavailable".to_string()), + ] { + let mapped = map_publish_key_validation_error(error); + assert!(mapped.downcast_ref::().is_none()); + assert!(mapped.downcast_ref::().is_some()); + } + } + struct FlakyUnregisterRegistry { inner: Arc, fail_unregister_if_lease_matches_times: AtomicUsize,