1use std::time::Duration;23use anyhow::{Context, Result};4use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};5use sigil_protocol::{6 GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,7 SignedMediaGenerationCertificate,8};91011pub(crate) enum MoqCatalogMode {12 GoqV2Authenticated,13 GoqV1,14 AbsentStaticTrackCompat,15}1617impl MoqCatalogMode {18 pub(crate) const fn label(self) -> &'static str {19 match self {20 Self::GoqV2Authenticated => "goq-v2-authenticated",21 Self::GoqV1 => "goq-v1",22 Self::AbsentStaticTrackCompat => "absent-static-track-compat",23 }24 }25}2627pub(crate) struct MoqCatalogSelection {28 pub(crate) track: TrackConsumer,29 pub(crate) mode: MoqCatalogMode,30 pub(crate) certificate: Option<SignedMediaGenerationCertificate>,31}3233fn is_track_not_found(error: &MoqError) -> bool {34 matches!(error, MoqError::NotFound)35 || matches!(error, MoqError::Remote(code) if *code == MoqError::NotFound.to_code())36}3738fn subscribe_static_video_track(broadcast: &BroadcastConsumer) -> Result<MoqCatalogSelection> {39 let track = broadcast40 .subscribe_track(&Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))41 .context("subscribing to static MoQ H.264 compatibility track")?;42 Ok(MoqCatalogSelection {43 track,44 mode: MoqCatalogMode::AbsentStaticTrackCompat,45 certificate: None,46 })47}4849pub(crate) async fn subscribe_goq_video_track(50 broadcast: &BroadcastConsumer,51 timeout: Duration,52) -> Result<MoqCatalogSelection> {53 let catalog_track = match broadcast.subscribe_track(&hang::Catalog::default_track()) {54 Ok(track) => track,55 Err(MoqError::NotFound) => {56 return subscribe_static_video_track(broadcast);57 }58 Err(error) => return Err(error).context("subscribing to catalog.json track"),59 };6061 let mut catalog_track = catalog_track;62 let deadline = tokio::time::Instant::now() + timeout;63 let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {64 Err(_) => anyhow::bail!("timed out waiting for Goq catalog snapshot"),65 Ok(Err(error)) if is_track_not_found(&error) => {66 return subscribe_static_video_track(broadcast);67 }68 Ok(Err(error)) => return Err(error).context("reading Goq catalog snapshot"),69 Ok(Ok(None)) => anyhow::bail!("Goq catalog track ended before its snapshot"),70 Ok(Ok(Some(group))) => group,71 };72 anyhow::ensure!(73 group.sequence == 0,74 "Goq catalog snapshot must be immutable group 0"75 );76 let frame_count = tokio::time::timeout_at(deadline, group.finished())77 .await78 .context("timed out waiting for Goq catalog group completion")?79 .context("finishing Goq catalog group")?;80 anyhow::ensure!(81 frame_count == 1,82 "Goq catalog snapshot group must contain exactly one frame"83 );84 let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))85 .await86 .context("timed out waiting for Goq catalog frame")?87 .context("opening Goq catalog frame")?88 .context("Goq catalog snapshot group has no frame 0")?;89 anyhow::ensure!(90 frame.size <= MAX_MOQ_CATALOG_BYTES as u64,91 "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"92 );93 let snapshot = tokio::time::timeout_at(deadline, frame.read_all())94 .await95 .context("timed out reading Goq catalog frame")?96 .context("reading Goq catalog frame")?;97 let document: GoqCatalogDocument =98 serde_json::from_slice(&snapshot).context("decoding Goq catalog snapshot")?;99 document100 .validate()101 .context("validating Goq catalog document")?;102 let track = broadcast103 .subscribe_track(&Track::new(document.goq.video.track.name))104 .context("subscribing to catalog-selected Goq video track")?;105 Ok(MoqCatalogSelection {106 track,107 mode: MoqCatalogMode::GoqV1,108 certificate: None,109 })110}111112pub(crate) async fn subscribe_authenticated_goq_video_track(113 broadcast: &BroadcastConsumer,114 timeout: Duration,115 expected_host: [u8; 32],116 expected_generation: u64,117 now_unix: u64,118) -> Result<MoqCatalogSelection> {119 let mut catalog_track = broadcast120 .subscribe_track(&hang::Catalog::default_track())121 .context("authenticated v2 media requires catalog.json")?;122 let deadline = tokio::time::Instant::now() + timeout;123 let mut group = tokio::time::timeout_at(deadline, catalog_track.next_group())124 .await125 .context("timed out waiting for authenticated Goq catalog")?126 .context("reading authenticated Goq catalog")?127 .context("authenticated Goq catalog ended before its snapshot")?;128 anyhow::ensure!(129 group.sequence == 0,130 "authenticated Goq catalog must be immutable group 0"131 );132 let frame_count = tokio::time::timeout_at(deadline, group.finished())133 .await134 .context("timed out finishing authenticated Goq catalog")?135 .context("finishing authenticated Goq catalog")?;136 anyhow::ensure!(137 frame_count == 1,138 "authenticated Goq catalog must contain one frame"139 );140 let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))141 .await142 .context("timed out opening authenticated Goq catalog")?143 .context("opening authenticated Goq catalog")?144 .context("authenticated Goq catalog has no frame 0")?;145 anyhow::ensure!(146 frame.size <= MAX_MOQ_CATALOG_BYTES as u64,147 "authenticated Goq catalog exceeds {MAX_MOQ_CATALOG_BYTES} bytes"148 );149 let snapshot = tokio::time::timeout_at(deadline, frame.read_all())150 .await151 .context("timed out reading authenticated Goq catalog")?152 .context("reading authenticated Goq catalog")?;153 let document: GoqCatalogDocumentV2 =154 serde_json::from_slice(&snapshot).context("decoding authenticated Goq catalog")?;155 let certificate = document156 .validate()157 .context("validating authenticated Goq catalog")?;158 certificate159 .verify_binding(expected_host, expected_generation, now_unix)160 .context("verifying media generation certificate binding")?;161 let track = broadcast162 .subscribe_track(&Track::new(document.goq.video.track.name))163 .context("subscribing to authenticated Goq video track")?;164 Ok(MoqCatalogSelection {165 track,166 mode: MoqCatalogMode::GoqV2Authenticated,167 certificate: Some(certificate),168 })169}170171172mod tests {173 use super::*;174 use moq_net::Broadcast;175176 fn publish_catalog(177 broadcast: &mut moq_net::BroadcastProducer,178 document: &GoqCatalogDocument,179 ) -> moq_net::TrackProducer {180 let mut track = broadcast181 .create_track(hang::Catalog::default_track())182 .unwrap();183 track184 .write_frame(serde_json::to_vec(document).unwrap())185 .unwrap();186 track187 }188189 190 async fn catalog_selects_the_exact_goq_video_track() {191 let mut broadcast = Broadcast::new().produce();192 let _video = broadcast193 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))194 .unwrap();195 let _catalog = publish_catalog(&mut broadcast, &GoqCatalogDocument::video_h264());196 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))197 .await198 .unwrap();199 assert_eq!(selection.mode, MoqCatalogMode::GoqV1);200 }201202 203 async fn only_an_absent_catalog_uses_the_legacy_static_track() {204 let mut broadcast = Broadcast::new().produce();205 let _video = broadcast206 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))207 .unwrap();208 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))209 .await210 .unwrap();211 assert_eq!(selection.mode, MoqCatalogMode::AbsentStaticTrackCompat);212 }213214 215 async fn asynchronous_remote_not_found_uses_static_track_compatibility() {216 let mut broadcast = Broadcast::new().produce();217 let _video = broadcast218 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))219 .unwrap();220 let mut dynamic = broadcast.dynamic();221 let consumer = broadcast.consume();222223 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));224 let reject = async {225 let mut requested = dynamic.requested_track().await.unwrap();226 assert_eq!(requested.name, hang::Catalog::default_track().name);227 requested228 .abort(MoqError::Remote(MoqError::NotFound.to_code()))229 .unwrap();230 };231 let (selection, ()) = tokio::join!(select, reject);232 assert_eq!(233 selection.unwrap().mode,234 MoqCatalogMode::AbsentStaticTrackCompat235 );236 }237238 239 async fn asynchronous_non_not_found_error_fails_closed() {240 let mut broadcast = Broadcast::new().produce();241 let _video = broadcast242 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))243 .unwrap();244 let mut dynamic = broadcast.dynamic();245 let consumer = broadcast.consume();246247 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));248 let reject = async {249 let mut requested = dynamic.requested_track().await.unwrap();250 requested251 .abort(MoqError::Remote(MoqError::WrongSize.to_code()))252 .unwrap();253 };254 let (selection, ()) = tokio::join!(select, reject);255 let error = selection.err().expect("non-NotFound must fail closed");256 assert!(257 format!("{error:#}").contains("remote error: code=14"),258 "{error:#}"259 );260 }261262 263 async fn present_invalid_or_stalled_catalogs_fail_closed() {264 let mut invalid_broadcast = Broadcast::new().produce();265 let _video = invalid_broadcast266 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))267 .unwrap();268 let mut invalid = GoqCatalogDocument::video_h264();269 invalid.goq.video.object_format = "hang/legacy".into();270 let _catalog = publish_catalog(&mut invalid_broadcast, &invalid);271 let result =272 subscribe_goq_video_track(&invalid_broadcast.consume(), Duration::from_millis(100))273 .await;274 assert!(result.err().is_some_and(|error| {275 format!("{error:#}").contains("validating Goq catalog document")276 }));277278 let mut stalled_broadcast = Broadcast::new().produce();279 let _catalog = stalled_broadcast280 .create_track(hang::Catalog::default_track())281 .unwrap();282 let result =283 subscribe_goq_video_track(&stalled_broadcast.consume(), Duration::from_millis(10))284 .await;285 assert!(286 result287 .err()288 .is_some_and(|error| error.to_string().contains("timed out"))289 );290291 let mut oversized_broadcast = Broadcast::new().produce();292 let _video = oversized_broadcast293 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))294 .unwrap();295 let mut catalog = oversized_broadcast296 .create_track(hang::Catalog::default_track())297 .unwrap();298 catalog299 .write_frame(vec![b' '; MAX_MOQ_CATALOG_BYTES + 1])300 .unwrap();301 let result =302 subscribe_goq_video_track(&oversized_broadcast.consume(), Duration::from_millis(100))303 .await;304 assert!(305 result306 .err()307 .is_some_and(|error| error.to_string().contains("exceeds 4096 bytes"))308 );309310 let mut multi_frame_broadcast = Broadcast::new().produce();311 let _video = multi_frame_broadcast312 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))313 .unwrap();314 let mut catalog = multi_frame_broadcast315 .create_track(hang::Catalog::default_track())316 .unwrap();317 let mut group = catalog.append_group().unwrap();318 group319 .write_frame(serde_json::to_vec(&GoqCatalogDocument::video_h264()).unwrap())320 .unwrap();321 group.write_frame(b"{}".as_slice()).unwrap();322 group.finish().unwrap();323 let result =324 subscribe_goq_video_track(&multi_frame_broadcast.consume(), Duration::from_millis(100))325 .await;326 assert!(327 result328 .err()329 .is_some_and(|error| error.to_string().contains("exactly one frame"))330 );331 }332}