From e7248534d0bbc6196302dc3d101f91609de260b6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 22 Jul 2026 09:30:57 -0700 Subject: [PATCH] feat(moq-hls)!: expose a credential-aware serve surface for embedders Lets a consumer embed the fetch-on-demand HLS origin as a library and put its own authorization in front, without a policy hook inside this crate. moq.pro's relay serves live HLS off its in-process origin and authorizes each request against its existing token verifier and subscribe scoping. Two things were missing. **A serve surface.** The broadcaster pool and the per-rendition accessors were crate-private, so the only way to serve HLS was `Server::router()` plus wrapping middleware -- which runs before routing and sees only the raw percent-encoded URI, an awkward seam for authorizing against a decoded broadcast path. Now public: Server::broadcaster() the shared, fanned-out pool Rendition::media_playlist() render this rendition's playlist Rendition::segment() fetch one segment by sequence Rendition::playable() await readiness `renditions` and `segments` were already public and already expose this data, but as cursors: the right shape for a recorder walking a stream once, the wrong shape for a server answering arbitrary sequence numbers for many viewers sitting at different positions. Random access by sequence is the one capability those cursors cannot express. Kept private everything an embedder can derive: `Snapshot`, `Segment` and `render_media` stay crate-internal (`media_playlist` returns the rendered string, and `None` when nothing is playable yet, which also subsumes `is_playable`); `Broadcaster::is_empty` (a `ready()` timeout means the catalog is empty); and `Rendition::init` (`segments::Consumer::init` is already public and delegates to it). No `Authorizer`/callback trait: the dependency arrow stays one-way and this crate keeps knowing nothing about tokens. **Credential propagation.** The renderers emit relative child URLs (`//media.m3u8`, `init.mp4`, `seg/.m4s`). A stock player -- Safari especially -- does not replay request headers onto those follow-ups, so a token supplied only via `Authorization` or `?jwt=` on the master request is gone by the time the player fetches a segment. An optional `query` (e.g. `jwt=`, no leading `?`) is appended to every child URL. It is a plain value, not a policy: the caller decides the parameter name. It is an argument rather than a `Config` field on purpose. `Config` is pool-wide (`Server::new` stores one and every broadcaster clones it) while a single broadcaster fans out to viewers holding *different* tokens, so a credential in `Config` would embed one viewer's JWT in another viewer's playlist. Deployment -scoped policy (a future signing key) would belong in `Config`; this cannot. `Server::router()` is unchanged in behaviour and still works as the no-auth convenience path; it now propagates whatever query reached each playlist, so it also works behind token-in-URL auth middleware. BREAKING CHANGE: `Broadcaster::master_playlist` takes a `query` argument. Co-Authored-By: Claude Opus 4.8 --- rs/moq-hls/src/export/master.rs | 31 ++++++++++++++++++++---------- rs/moq-hls/src/export/mod.rs | 20 +++++++++++++++---- rs/moq-hls/src/export/playlist.rs | 22 ++++++++++++++++----- rs/moq-hls/src/export/rendition.rs | 29 +++++++++++++++++++--------- rs/moq-hls/src/server/mod.rs | 2 +- rs/moq-hls/src/server/routes.rs | 17 +++++++++------- 6 files changed, 85 insertions(+), 36 deletions(-) diff --git a/rs/moq-hls/src/export/master.rs b/rs/moq-hls/src/export/master.rs index 40e9b9c91a..9bf549b748 100644 --- a/rs/moq-hls/src/export/master.rs +++ b/rs/moq-hls/src/export/master.rs @@ -69,7 +69,7 @@ fn group_audio(audio: &[AudioVariant]) -> Vec> { .collect() } -fn render_video(out: &mut String, variant: &VideoVariant, audio: Option<&AudioGroup<'_>>) { +fn render_video(out: &mut String, variant: &VideoVariant, audio: Option<&AudioGroup<'_>>, suffix: &str) { let bandwidth = variant .bandwidth .saturating_add(audio.map_or(0, |group| group.bandwidth)); @@ -86,11 +86,17 @@ fn render_video(out: &mut String, variant: &VideoVariant, audio: Option<&AudioGr let _ = write!(line, ",AUDIO=\"{}\"", group.id); } let _ = writeln!(out, "{line}"); - let _ = writeln!(out, "{}/{}/media.m3u8", Kind::Video.as_str(), variant.name); + let _ = writeln!(out, "{}/{}/media.m3u8{suffix}", Kind::Video.as_str(), variant.name); } /// Render the multivariant playlist. The first rendition in each audio codec group is default. -pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String { +/// +/// `query` is an optional query string (without the leading `?`, e.g. `jwt=`) +/// appended to every child media-playlist URL, so a credential the master was fetched +/// with propagates to the rendition playlists a stock player loads next. +pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant], query: Option<&str>) -> String { + let suffix = query.map(|q| format!("?{q}")).unwrap_or_default(); + let mut out = String::new(); let _ = writeln!(out, "#EXTM3U"); let _ = writeln!(out, "#EXT-X-VERSION:{VERSION}"); @@ -101,7 +107,7 @@ pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String { let default = if index == 0 { "YES" } else { "NO" }; let _ = writeln!( out, - "#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"{}\",NAME=\"{}\",DEFAULT={default},AUTOSELECT=YES,URI=\"{}/{}/media.m3u8\"", + "#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"{}\",NAME=\"{}\",DEFAULT={default},AUTOSELECT=YES,URI=\"{}/{}/media.m3u8{suffix}\"", group.id, variant.name, Kind::Audio.as_str(), @@ -112,10 +118,10 @@ pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String { for variant in video { if audio_groups.is_empty() { - render_video(&mut out, variant, None); + render_video(&mut out, variant, None, &suffix); } else { for group in &audio_groups { - render_video(&mut out, variant, Some(group)); + render_video(&mut out, variant, Some(group), &suffix); } } } @@ -128,7 +134,7 @@ pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String { "#EXT-X-STREAM-INF:BANDWIDTH={},CODECS=\"{}\"", variant.bandwidth, variant.codec ); - let _ = writeln!(out, "{}/{}/media.m3u8", Kind::Audio.as_str(), variant.name); + let _ = writeln!(out, "{}/{}/media.m3u8{suffix}", Kind::Audio.as_str(), variant.name); } } @@ -154,7 +160,7 @@ mod tests { codec: "mp4a.40.2".into(), }]; - let out = render_master(&video, &audio); + let out = render_master(&video, &audio, None); assert!(out.starts_with("#EXTM3U\n#EXT-X-VERSION:9\n")); assert!(out.contains( "#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"aud\",NAME=\"audio\",DEFAULT=YES,AUTOSELECT=YES,URI=\"audio/audio/media.m3u8\"\n" @@ -163,6 +169,11 @@ mod tests { "#EXT-X-STREAM-INF:BANDWIDTH=2628000,RESOLUTION=1280x720,CODECS=\"avc1.42c01f,mp4a.40.2\",AUDIO=\"aud\"\n" )); assert!(out.contains("\nvideo/video/media.m3u8\n")); + + // A credential rides every child media-playlist URL, audio and video alike. + let signed = render_master(&video, &audio, Some("jwt=abc.def")); + assert!(signed.contains("URI=\"audio/audio/media.m3u8?jwt=abc.def\"\n")); + assert!(signed.contains("\nvideo/video/media.m3u8?jwt=abc.def\n")); } #[test] @@ -192,7 +203,7 @@ mod tests { }, ]; - let out = render_master(&video, &audio); + let out = render_master(&video, &audio, None); assert!(out.contains("GROUP-ID=\"aud-0\",NAME=\"aac-low\",DEFAULT=YES")); assert!(out.contains("GROUP-ID=\"aud-0\",NAME=\"aac-high\",DEFAULT=NO")); assert!(out.contains("GROUP-ID=\"aud-1\",NAME=\"opus\",DEFAULT=YES")); @@ -208,7 +219,7 @@ mod tests { bandwidth: 128_000, codec: "opus".into(), }]; - let out = render_master(&[], &audio); + let out = render_master(&[], &audio, None); assert!(out.contains("#EXT-X-STREAM-INF:BANDWIDTH=128000,CODECS=\"opus\"\n")); assert!(out.contains("\naudio/audio/media.m3u8\n")); } diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 71252de4e3..d9e8f58d75 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -118,7 +118,14 @@ impl Broadcaster { } /// Render the multivariant (master) playlist from the current renditions. - pub fn master_playlist(&self) -> String { + /// + /// `query` is an optional query string (without the leading `?`, e.g. `jwt=`) + /// appended to every child media-playlist URL, so a credential the master was fetched + /// with propagates to the rendition playlists a stock player loads next. It is an + /// argument rather than a [`Config`] field because one broadcaster fans out to viewers + /// holding different tokens; a credential in `Config` would embed one viewer's in + /// another's playlist. + pub fn master_playlist(&self, query: Option<&str>) -> String { let mut video = Vec::new(); let mut audio = Vec::new(); for rendition in self.renditions.snapshot() { @@ -137,7 +144,7 @@ impl Broadcaster { }), } } - master::render_master(&video, &audio) + master::render_master(&video, &audio, query) } /// Whether the current catalog contains no servable renditions (serve path). @@ -254,7 +261,7 @@ mod tests { .expect("rendition discovered from the catalog"); let _ = tokio::time::timeout(Duration::from_secs(5), rendition.playable()).await; - let master = broadcaster.master_playlist(); + let master = broadcaster.master_playlist(None); assert!(master.contains("video/video0/media.m3u8"), "master lists the rendition"); let playlist = rendition.playlist(); @@ -265,11 +272,16 @@ mod tests { assert_eq!(playlist.target_duration, 2, "ceil of the longest timeline gap"); assert!(!playlist.finished); - let rendered = render_media(&playlist); + let rendered = rendition.media_playlist(None).expect("playable"); assert!(rendered.contains("#EXT-X-MAP:URI=\"init.mp4\"\n")); assert!(rendered.contains("seg/0.m4s\n")); assert!(rendered.contains("seg/1.m4s\n")); + // The same render, but carrying a credential into every child URL. + let signed = rendition.media_playlist(Some("jwt=abc.def")).expect("playable"); + assert!(signed.contains("#EXT-X-MAP:URI=\"init.mp4?jwt=abc.def\"\n")); + assert!(signed.contains("seg/0.m4s?jwt=abc.def\n")); + let init = rendition.init().await.unwrap().expect("init segment"); assert_eq!(&init[4..8], b"ftyp"); diff --git a/rs/moq-hls/src/export/playlist.rs b/rs/moq-hls/src/export/playlist.rs index 035f3a01bf..aeb4db825b 100644 --- a/rs/moq-hls/src/export/playlist.rs +++ b/rs/moq-hls/src/export/playlist.rs @@ -36,13 +36,19 @@ pub(crate) struct Segment { } /// Render a media playlist for one rendition from a [`Snapshot`]. -pub(crate) fn render_media(snapshot: &Snapshot) -> String { +/// +/// `query` is an optional query string (without the leading `?`, e.g. `jwt=`) +/// appended to every child URL (the init map and each segment), so a stock player that +/// does not replay request headers still carries a credential on its follow-up requests. +pub(crate) fn render_media(snapshot: &Snapshot, query: Option<&str>) -> String { + let suffix = query.map(|q| format!("?{q}")).unwrap_or_default(); + let mut out = String::new(); let _ = writeln!(out, "#EXTM3U"); let _ = writeln!(out, "#EXT-X-VERSION:{VERSION}"); let _ = writeln!(out, "#EXT-X-TARGETDURATION:{}", snapshot.target_duration); let _ = writeln!(out, "#EXT-X-MEDIA-SEQUENCE:{}", snapshot.media_sequence); - let _ = writeln!(out, "#EXT-X-MAP:URI=\"init.mp4\""); + let _ = writeln!(out, "#EXT-X-MAP:URI=\"init.mp4{suffix}\""); for (index, segment) in snapshot.segments.iter().enumerate() { if index == 0 @@ -55,7 +61,7 @@ pub(crate) fn render_media(snapshot: &Snapshot) -> String { ); } let _ = writeln!(out, "#EXTINF:{:.5},", segment.duration); - let _ = writeln!(out, "seg/{}.m4s", segment.group); + let _ = writeln!(out, "seg/{}.m4s{suffix}", segment.group); } if snapshot.finished { @@ -90,7 +96,7 @@ mod tests { program_date_time: Some(SystemTime::UNIX_EPOCH + Duration::from_millis(1_751_846_400_123)), }; - let out = render_media(&snapshot); + let out = render_media(&snapshot, None); assert!(out.starts_with("#EXTM3U\n#EXT-X-VERSION:6\n")); assert!(out.contains("#EXT-X-TARGETDURATION:2\n")); assert!(out.contains("#EXT-X-MEDIA-SEQUENCE:10\n")); @@ -99,6 +105,12 @@ mod tests { assert!(out.contains("#EXTINF:2.00000,\nseg/10.m4s\n")); assert!(out.contains("#EXTINF:1.96000,\nseg/11.m4s\n")); assert!(!out.contains("#EXT-X-ENDLIST")); + + // A credential rides every child URL so a header-less player keeps sending it. + let signed = render_media(&snapshot, Some("jwt=abc.def")); + assert!(signed.contains("#EXT-X-MAP:URI=\"init.mp4?jwt=abc.def\"\n")); + assert!(signed.contains("\nseg/10.m4s?jwt=abc.def\n")); + assert!(signed.contains("\nseg/11.m4s?jwt=abc.def\n")); } #[test] @@ -114,7 +126,7 @@ mod tests { program_date_time: None, }; - let out = render_media(&snapshot); + let out = render_media(&snapshot, None); assert!(out.contains("#EXT-X-ENDLIST\n")); assert!(!out.contains("PROGRAM-DATE-TIME")); } diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index eb45f4423b..f06d3982e4 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -83,8 +83,7 @@ pub struct Rendition { live: Arc, /// The largest `EXT-X-TARGETDURATION` advertised so far, in seconds. Latched monotonically: /// HLS forbids a live playlist's target duration from changing, so it never shrinks even - /// after a long segment evicts from the window. Read only by the serve path's `playlist`. - #[cfg_attr(not(feature = "server"), allow(dead_code))] + /// after a long segment evicts from the window. target_duration: std::sync::atomic::AtomicU64, /// The init segment, built on first request. init: tokio::sync::Mutex>, @@ -185,15 +184,28 @@ impl Rendition { self.watcher.abort(); } - /// Resolve once the playlist is renderable ([`is_playable`](Self::is_playable)). Bounding - /// the wait is the caller's policy (the serve path wraps this in its own timeout). - #[cfg_attr(not(feature = "server"), allow(dead_code))] - pub(crate) async fn playable(&self) { + /// Resolve once the playlist is renderable, i.e. once [`media_playlist`](Self::media_playlist) + /// would return `Some`. Bounding the wait is the caller's policy (the serve path wraps this + /// in its own timeout). + pub async fn playable(&self) { kio::wait(|waiter| self.live.poll_playable(waiter)).await; } + /// Render this rendition's media playlist from the current timeline window, or `None` when + /// there is nothing to serve yet (no complete segment, and the broadcast has not ended) -- + /// a playlist with no segments confuses players, so a server should treat `None` as "not + /// ready" rather than serve it. + /// + /// `query` is an optional query string (without the leading `?`, e.g. `jwt=`) + /// appended to every child URL (the init map and each segment), so a stock player that does + /// not replay request headers still carries a credential on its follow-up requests. It is an + /// argument rather than a [`Config`](super::Config) field because one broadcaster fans out + /// to viewers holding different tokens. + pub fn media_playlist(&self, query: Option<&str>) -> Option { + self.is_playable().then(|| super::render_media(&self.playlist(), query)) + } + /// Render the media playlist from the current timeline window. - #[cfg_attr(not(feature = "server"), allow(dead_code))] pub(crate) fn playlist(&self) -> Snapshot { let window = self.live.window(); @@ -228,7 +240,6 @@ impl Rendition { /// Whether the playlist has anything to serve yet (at least one complete segment, or the /// broadcast already ended). - #[cfg_attr(not(feature = "server"), allow(dead_code))] pub(crate) fn is_playable(&self) -> bool { self.live.is_playable() } @@ -301,7 +312,7 @@ impl Rendition { /// Fetches every group the segment covers (audio timelines skip groups, so a segment may span /// several) and encodes them as a single CMAF fragment. `None` when the segment isn't in the /// playlist window or its groups already left the relay cache. - pub(crate) async fn segment(&self, sequence: u64) -> Result> { + pub async fn segment(&self, sequence: u64) -> Result> { let Some((start, end)) = self.live.segment_groups(sequence) else { return Ok(None); }; diff --git a/rs/moq-hls/src/server/mod.rs b/rs/moq-hls/src/server/mod.rs index 4514a1078b..5d6e959d69 100644 --- a/rs/moq-hls/src/server/mod.rs +++ b/rs/moq-hls/src/server/mod.rs @@ -84,7 +84,7 @@ impl Server { /// Get or create the [`Broadcaster`] for `name`, resolving the broadcast from /// the relay (waiting briefly for its announcement). Returns `None` if the /// broadcast never shows up. - pub(crate) async fn broadcaster(&self, name: &str) -> Option> { + pub async fn broadcaster(&self, name: &str) -> Option> { { let mut broadcasters = self.inner.broadcasters.lock().unwrap(); if let Some(existing) = broadcasters.get(name) { diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index c44d0bb3fa..55d909b5d9 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::time::Duration; use axum::Router; -use axum::extract::{Path, State}; +use axum::extract::{Path, RawQuery, State}; use axum::http::{StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::get; @@ -29,7 +29,7 @@ pub fn router(server: Server) -> Router { .with_state(server) } -async fn master(State(server): State, Path(broadcast): Path) -> Response { +async fn master(State(server): State, Path(broadcast): Path, RawQuery(query): RawQuery) -> Response { let Some(broadcaster) = server.broadcaster(&broadcast).await else { return not_found(); }; @@ -37,12 +37,15 @@ async fn master(State(server): State, Path(broadcast): Path) -> if broadcaster.is_empty() { return not_found(); } - m3u8(broadcaster.master_playlist()) + // Propagate whatever query reached the master (e.g. a credential a wrapping + // middleware required) down to the child media-playlist URLs. + m3u8(broadcaster.master_playlist(query.as_deref())) } async fn media( State(server): State, Path((broadcast, kind, rendition)): Path<(String, String, String)>, + RawQuery(query): RawQuery, ) -> Response { let Some(rendition) = rendition_for(&server, &broadcast, &kind, &rendition).await else { return not_found(); @@ -51,9 +54,6 @@ async fn media( // A playlist with no segments confuses players; give the timeline a moment to index the // first complete segment before answering. let _ = tokio::time::timeout(READY_TIMEOUT, rendition.playable()).await; - if !rendition.is_playable() { - return not_found(); - } // The playlist references init.mp4 via EXT-X-MAP. Make sure it's actually buildable before // advertising it (an inline-codec init needs a keyframe group fetched first), so a player @@ -64,7 +64,10 @@ async fn media( Err(err) => return server_error(err), } - m3u8(crate::export::render_media(&rendition.playlist())) + match rendition.media_playlist(query.as_deref()) { + Some(playlist) => m3u8(playlist), + None => not_found(), + } } async fn init(