1use std::time::Duration;23use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};4use sigil_protocol::{5 GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,6 SignedMediaGenerationCertificate,7};8910pub(crate) enum MoqCatalogMode {11 GoqV2Authenticated,12 GoqV1,13 AbsentStaticTrackCompat,14}1516impl MoqCatalogMode {17 pub(crate) const fn label(self) -> &'static str {18 match self {19 Self::GoqV2Authenticated => "goq-v2-authenticated",20 Self::GoqV1 => "goq-v1",21 Self::AbsentStaticTrackCompat => "absent-static-track-compat",22 }23 }24}2526pub(crate) struct MoqCatalogSelection {27 pub(crate) track: TrackConsumer,28 pub(crate) audio_track: Option<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(39 broadcast: &BroadcastConsumer,40) -> Result<MoqCatalogSelection, String> {41 let track = broadcast42 .subscribe_track(&Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))43 .map_err(|error| {44 format!("Failed to subscribe to static MoQ H.264 compatibility track: {error}")45 })?;46 Ok(MoqCatalogSelection {47 track,48 audio_track: None,49 mode: MoqCatalogMode::AbsentStaticTrackCompat,50 certificate: None,51 })52}5354pub(crate) async fn subscribe_goq_video_track(55 broadcast: &BroadcastConsumer,56 timeout: Duration,57) -> Result<MoqCatalogSelection, String> {58 let catalog_track = match broadcast.subscribe_track(&hang::Catalog::default_track()) {59 Ok(track) => track,60 Err(MoqError::NotFound) => {61 return subscribe_static_video_track(broadcast);62 }63 Err(error) => {64 return Err(format!(65 "Failed to subscribe to catalog.json track: {error}"66 ));67 }68 };6970 let snapshot = match read_catalog_snapshot(catalog_track, timeout).await {71 Err(error) if error == "catalog-not-found" => {72 return subscribe_static_video_track(broadcast);73 }74 result => result?,75 };76 let document: GoqCatalogDocument = serde_json::from_slice(&snapshot)77 .map_err(|error| format!("Failed to decode Goq catalog snapshot: {error}"))?;78 document79 .validate()80 .map_err(|error| format!("Invalid Goq catalog: {error}"))?;81 let track = broadcast82 .subscribe_track(&Track::new(document.goq.video.track.name))83 .map_err(|error| {84 format!("Failed to subscribe to catalog-selected Goq video track: {error}")85 })?;86 Ok(MoqCatalogSelection {87 track,88 audio_track: None,89 mode: MoqCatalogMode::GoqV1,90 certificate: None,91 })92}9394pub(crate) async fn subscribe_authenticated_goq_video_track(95 broadcast: &BroadcastConsumer,96 timeout: Duration,97 expected_host: [u8; 32],98 expected_generation: u64,99 now_unix: u64,100) -> Result<MoqCatalogSelection, String> {101 let catalog_track = broadcast102 .subscribe_track(&hang::Catalog::default_track())103 .map_err(|error| format!("Authenticated v2 media requires catalog.json: {error}"))?;104 let snapshot = read_catalog_snapshot(catalog_track, timeout).await?;105 let document: GoqCatalogDocumentV2 = serde_json::from_slice(&snapshot)106 .map_err(|error| format!("Failed to decode authenticated Goq catalog: {error}"))?;107 let certificate = document108 .validate()109 .map_err(|error| format!("Invalid authenticated Goq catalog: {error}"))?;110 certificate111 .verify_binding(expected_host, expected_generation, now_unix)112 .map_err(|error| format!("Invalid media generation certificate: {error}"))?;113 let track = broadcast114 .subscribe_track(&Track::new(document.goq.video.track.name))115 .map_err(|error| {116 format!("Failed to subscribe to authenticated Goq video track: {error}")117 })?;118 let audio_track = document119 .goq120 .audio121 .as_ref()122 .map(|audio| {123 broadcast124 .subscribe_track(&Track::new(audio.track.name.clone()))125 .map_err(|error| {126 format!("Failed to subscribe to authenticated Goq audio track: {error}")127 })128 })129 .transpose()?;130 Ok(MoqCatalogSelection {131 track,132 audio_track,133 mode: MoqCatalogMode::GoqV2Authenticated,134 certificate: Some(certificate),135 })136}137138async fn read_catalog_snapshot(139 mut catalog_track: TrackConsumer,140 timeout: Duration,141) -> Result<bytes::Bytes, String> {142 let deadline = tokio::time::Instant::now() + timeout;143 let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {144 Err(_) => return Err("Timed out waiting for Goq catalog snapshot".to_string()),145 Ok(Err(error)) if is_track_not_found(&error) => return Err("catalog-not-found".to_string()),146 Ok(Err(error)) => return Err(format!("Failed to read Goq catalog snapshot: {error}")),147 Ok(Ok(None)) => return Err("Goq catalog track ended before its snapshot".to_string()),148 Ok(Ok(Some(group))) => group,149 };150 if group.sequence != 0 {151 return Err("Goq catalog snapshot must be immutable group 0".to_string());152 }153 let frame_count = tokio::time::timeout_at(deadline, group.finished())154 .await155 .map_err(|_| "Timed out waiting for Goq catalog group completion".to_string())?156 .map_err(|error| format!("Failed to finish Goq catalog group: {error}"))?;157 if frame_count != 1 {158 return Err("Goq catalog snapshot group must contain exactly one frame".to_string());159 }160 let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))161 .await162 .map_err(|_| "Timed out waiting for Goq catalog frame".to_string())?163 .map_err(|error| format!("Failed to open Goq catalog frame: {error}"))?164 .ok_or_else(|| "Goq catalog snapshot group has no frame 0".to_string())?;165 if frame.size > MAX_MOQ_CATALOG_BYTES as u64 {166 return Err(format!(167 "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"168 ));169 }170 tokio::time::timeout_at(deadline, frame.read_all())171 .await172 .map_err(|_| "Timed out reading Goq catalog frame".to_string())?173 .map_err(|error| format!("Failed to read Goq catalog frame: {error}"))174}175176177mod tests {178 use super::*;179 use ed25519_dalek::SigningKey;180 use moq_net::Broadcast;181182 fn v2_document() -> (GoqCatalogDocumentV2, [u8; 32]) {183 let host = SigningKey::from_bytes(&[7; 32]);184 let generation = sigil_protocol::MediaGenerationSigningKey::from_bytes(&[9; 32]);185 let certificate = generation186 .certify(187 host.verifying_key().to_bytes(),188 &[7; 32],189 42,190 1_700_000_000,191 1_700_000_600,192 )193 .unwrap();194 (195 GoqCatalogDocumentV2::video_h264_opus(42, &certificate).unwrap(),196 host.verifying_key().to_bytes(),197 )198 }199200 fn publish_catalog(201 broadcast: &mut moq_net::BroadcastProducer,202 document: &GoqCatalogDocument,203 ) -> moq_net::TrackProducer {204 let mut track = broadcast205 .create_track(hang::Catalog::default_track())206 .unwrap();207 track208 .write_frame(serde_json::to_vec(document).unwrap())209 .unwrap();210 track211 }212213 fn publish_catalog_v2(214 broadcast: &mut moq_net::BroadcastProducer,215 document: &GoqCatalogDocumentV2,216 ) -> moq_net::TrackProducer {217 let mut track = broadcast218 .create_track(hang::Catalog::default_track())219 .unwrap();220 track221 .write_frame(serde_json::to_vec(document).unwrap())222 .unwrap();223 track224 }225226 227 async fn catalog_selects_the_exact_goq_video_track() {228 let mut broadcast = Broadcast::new().produce();229 let _video = broadcast230 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))231 .unwrap();232 let _catalog = publish_catalog(&mut broadcast, &GoqCatalogDocument::video_h264());233 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))234 .await235 .unwrap();236 assert_eq!(selection.mode, MoqCatalogMode::GoqV1);237 }238239 240 async fn authenticated_catalog_verifies_host_generation_and_mode() {241 let mut broadcast = Broadcast::new().produce();242 let _video = broadcast243 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))244 .unwrap();245 let (document, host) = v2_document();246 let _audio = broadcast247 .create_track(Track::new(sigil_protocol::MOQ_AUDIO_OPUS_TRACK))248 .unwrap();249 let _catalog = publish_catalog_v2(&mut broadcast, &document);250 let selection = subscribe_authenticated_goq_video_track(251 &broadcast.consume(),252 Duration::from_millis(100),253 host,254 42,255 1_700_000_300,256 )257 .await258 .unwrap();259 assert_eq!(selection.mode, MoqCatalogMode::GoqV2Authenticated);260 assert!(selection.audio_track.is_some());261 assert_eq!(selection.certificate.unwrap().claims.generation_id, 42);262263 let mut wrong_host_broadcast = Broadcast::new().produce();264 let _video = wrong_host_broadcast265 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))266 .unwrap();267 let _catalog = publish_catalog_v2(&mut wrong_host_broadcast, &document);268 assert!(269 subscribe_authenticated_goq_video_track(270 &wrong_host_broadcast.consume(),271 Duration::from_millis(100),272 [1; 32],273 42,274 1_700_000_300,275 )276 .await277 .is_err()278 );279 }280281 282 async fn only_an_absent_catalog_uses_the_legacy_static_track() {283 let mut broadcast = Broadcast::new().produce();284 let _video = broadcast285 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))286 .unwrap();287 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))288 .await289 .unwrap();290 assert_eq!(selection.mode, MoqCatalogMode::AbsentStaticTrackCompat);291 }292293 294 async fn asynchronous_remote_not_found_uses_static_track_compatibility() {295 let mut broadcast = Broadcast::new().produce();296 let _video = broadcast297 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))298 .unwrap();299 let mut dynamic = broadcast.dynamic();300 let consumer = broadcast.consume();301302 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));303 let reject = async {304 let mut requested = dynamic.requested_track().await.unwrap();305 assert_eq!(requested.name, hang::Catalog::default_track().name);306 requested307 .abort(MoqError::Remote(MoqError::NotFound.to_code()))308 .unwrap();309 };310 let (selection, ()) = tokio::join!(select, reject);311 assert_eq!(312 selection.unwrap().mode,313 MoqCatalogMode::AbsentStaticTrackCompat314 );315 }316317 318 async fn asynchronous_non_not_found_error_fails_closed() {319 let mut broadcast = Broadcast::new().produce();320 let _video = broadcast321 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))322 .unwrap();323 let mut dynamic = broadcast.dynamic();324 let consumer = broadcast.consume();325326 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));327 let reject = async {328 let mut requested = dynamic.requested_track().await.unwrap();329 requested330 .abort(MoqError::Remote(MoqError::WrongSize.to_code()))331 .unwrap();332 };333 let (selection, ()) = tokio::join!(select, reject);334 assert!(335 selection336 .err()337 .is_some_and(|error| error.contains("remote error: code=14"))338 );339 }340341 342 async fn present_invalid_or_stalled_catalogs_fail_closed() {343 let mut invalid_broadcast = Broadcast::new().produce();344 let _video = invalid_broadcast345 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))346 .unwrap();347 let mut invalid = GoqCatalogDocument::video_h264();348 invalid.goq.video.object_format = "hang/legacy".into();349 let _catalog = publish_catalog(&mut invalid_broadcast, &invalid);350 let result =351 subscribe_goq_video_track(&invalid_broadcast.consume(), Duration::from_millis(100))352 .await;353 assert!(354 result355 .err()356 .is_some_and(|error| error.contains("Invalid Goq catalog"))357 );358359 let mut stalled_broadcast = Broadcast::new().produce();360 let _catalog = stalled_broadcast361 .create_track(hang::Catalog::default_track())362 .unwrap();363 let result =364 subscribe_goq_video_track(&stalled_broadcast.consume(), Duration::from_millis(10))365 .await;366 assert!(367 result368 .err()369 .is_some_and(|error| error.contains("Timed out"))370 );371372 let mut oversized_broadcast = Broadcast::new().produce();373 let _video = oversized_broadcast374 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))375 .unwrap();376 let mut catalog = oversized_broadcast377 .create_track(hang::Catalog::default_track())378 .unwrap();379 catalog380 .write_frame(vec![b' '; MAX_MOQ_CATALOG_BYTES + 1])381 .unwrap();382 let result =383 subscribe_goq_video_track(&oversized_broadcast.consume(), Duration::from_millis(100))384 .await;385 assert!(386 result387 .err()388 .is_some_and(|error| error.contains("exceeds 4096 bytes"))389 );390391 let mut multi_frame_broadcast = Broadcast::new().produce();392 let _video = multi_frame_broadcast393 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))394 .unwrap();395 let mut catalog = multi_frame_broadcast396 .create_track(hang::Catalog::default_track())397 .unwrap();398 let mut group = catalog.append_group().unwrap();399 group400 .write_frame(serde_json::to_vec(&GoqCatalogDocument::video_h264()).unwrap())401 .unwrap();402 group.write_frame(b"{}".as_slice()).unwrap();403 group.finish().unwrap();404 let result =405 subscribe_goq_video_track(&multi_frame_broadcast.consume(), Duration::from_millis(100))406 .await;407 assert!(408 result409 .err()410 .is_some_and(|error| error.contains("exactly one frame"))411 );412 }413}