moq_rtc/server/
whep.rs

1//! `server subscribe`: WHEP server.
2//!
3//! `POST /<broadcast-path>` accepts a WHEP SDP offer and returns an SDP
4//! answer sourced from the matching MoQ broadcast on the subscribe origin.
5
6use axum::{
7	Router,
8	body::Bytes,
9	extract::{OriginalUri, Path, State},
10	http::{HeaderMap, HeaderValue, StatusCode, header},
11	response::{IntoResponse, Response as HttpResponse},
12	routing::post,
13};
14use str0m::Candidate;
15
16use std::time::Duration;
17
18use crate::{Error, Result, egress::EgressSource, sdp, server::Server, session};
19
20pub use crate::server::Response;
21
22/// How long WHEP negotiation waits for the broadcast's first catalog snapshot
23/// before failing the request. A broadcast can be announced (or served by a
24/// dynamic origin fallback) yet never publish a catalog; without a bound the
25/// `accept` future parks forever inside the HTTP handler, leaking the request.
26const CATALOG_TIMEOUT: Duration = Duration::from_secs(5);
27
28/// Build the WHEP axum router.
29pub fn router(server: Server) -> Router {
30	Router::new()
31		.route("/{*path}", post(handle).delete(crate::server::delete))
32		.with_state(server)
33}
34
35async fn handle(
36	server: State<Server>,
37	path: Path<String>,
38	OriginalUri(uri): OriginalUri,
39	headers: HeaderMap,
40	body: Bytes,
41) -> HttpResponse {
42	let (server, path) = (server.0, path.0);
43	match accept_offer(&server, &path, &headers, body).await {
44		Ok(response) => {
45			let Response {
46				resource_id,
47				answer,
48				session,
49			} = response;
50			let mut response_headers = HeaderMap::new();
51			response_headers.insert(header::CONTENT_TYPE, HeaderValue::from_static("application/sdp"));
52			if let Some(loc) = crate::server::session_location(&uri, &resource_id) {
53				response_headers.insert(header::LOCATION, loc);
54			}
55			tokio::spawn(async move {
56				let _ = session.run().await;
57			});
58			(StatusCode::CREATED, response_headers, answer).into_response()
59		}
60		Err(err) => {
61			tracing::warn!(%err, "whep request failed");
62			(status_for(&err), err.to_string()).into_response()
63		}
64	}
65}
66
67/// Router glue: enforce the WHEP `Content-Type` then hand the raw offer to
68/// [`accept`], using the request path as the (unauthenticated) broadcast name.
69async fn accept_offer(server: &Server, path: &str, headers: &HeaderMap, body: Bytes) -> Result<Response> {
70	if !is_sdp(headers) {
71		return Err(Error::InvalidSdp("expected Content-Type: application/sdp".into()));
72	}
73	let offer = std::str::from_utf8(&body).map_err(|err| Error::InvalidSdp(err.to_string()))?;
74	accept(server, server.subscriber(), path, offer).await
75}
76
77/// Accept a WHEP SDP offer and egress the MoQ broadcast `broadcast` (a path
78/// relative to `subscriber`'s root) to the negotiated WebRTC peer.
79///
80/// This is the negotiation core behind [`router`], exposed so an embedder can own
81/// the HTTP route and authentication: verify the request, resolve the authorized
82/// broadcast name, scope `subscriber` to the caller's grants, then hand the raw
83/// SDP offer here. Taking the consumer explicitly (rather than using the server's)
84/// lets the embedder egress through a *scoped* origin, so the subscribe scope is
85/// enforced by moq-net exactly as for a native session; the bundled [`router`]
86/// passes the server's own (unauthenticated) consumer. It parses the offer,
87/// resolves the broadcast on `subscriber`, restricts the answer to the codecs the
88/// catalog actually has, registers a media session on the shared mux, and
89/// returns the SDP answer plus an opaque `resource_id` for the WHEP `Location`
90/// header. The caller must run the returned [`Response`] to drive the MoQ->RTP
91/// session. Mirrors [`whip::accept`](super::whip::accept).
92///
93/// `offer` is the raw SDP body; the caller is responsible for checking the
94/// `Content-Type: application/sdp` request header. Fails with [`Error::InvalidSdp`]
95/// on a malformed offer, and surfaces a not-announced broadcast (or one outside
96/// `subscriber`'s scope) as [`Error::Other`].
97pub async fn accept(
98	server: &Server,
99	subscriber: &moq_net::origin::Consumer,
100	broadcast: impl moq_net::AsPath,
101	offer: &str,
102) -> Result<Response> {
103	let offer = sdp::parse_offer(offer)?;
104
105	// Resolve the broadcast on the subscribe origin by path. The `Source` fetches it (and any
106	// sibling broadcast a rendition's catalog `broadcast` field references, e.g. `./source`)
107	// via `request_broadcast`, which resolves an announced broadcast immediately and falls back
108	// to a dynamic handler; with neither it errors and the WHEP client retries (typical).
109	let broadcast = broadcast.as_path().to_string();
110	let source = moq_mux::Source::new(subscriber.clone(), &broadcast);
111
112	// Bound the wait for resolution + the first catalog: an unannounced broadcast, or an
113	// announced-but-catalog-less one, would otherwise park this handler forever.
114	let source = tokio::time::timeout(CATALOG_TIMEOUT, EgressSource::new(source))
115		.await
116		.map_err(|_| {
117			Error::Other(anyhow::anyhow!(
118				"broadcast {broadcast} did not resolve with a catalog within {CATALOG_TIMEOUT:?}"
119			))
120		})??;
121	let codecs = source.catalog_codecs();
122	if codecs.is_empty() {
123		return Err(Error::Other(anyhow::anyhow!(
124			"catalog has no codecs we can egress (Opus / H.264 / H.265 / VP8 / VP9 / AV1)"
125		)));
126	}
127
128	// Register a session on the shared media mux (see whip::accept). Restrict our
129	// CodecConfig before accept_offer so the answer intersects the peer's offer
130	// with what the catalog actually has, instead of agreeing to a codec we can't
131	// fulfil; set the mux's known ICE credentials on the same config.
132	let mux = server.mux().await?;
133	let (creds, inbound, registration) = mux.register();
134	let mut rtc = session::rtc_config_with_codecs(&codecs)
135		.set_local_ice_credentials(creds)
136		.build(std::time::Instant::now());
137	for addr in mux.candidates() {
138		let cand = Candidate::host(*addr, "udp").map_err(str0m::RtcError::from)?;
139		rtc.add_local_candidate(cand);
140	}
141
142	let answer = rtc.sdp_api().accept_offer(offer).map_err(Error::Rtc)?;
143	let resource_id = sdp::new_resource_id();
144	let session = session::Session::egress(rtc, mux.socket(), mux.candidates().to_vec(), inbound, source);
145
146	// Register before returning so a DELETE that races startup still finds the
147	// session; Response::run unregisters itself when it ends.
148	let cancel = server.register_session(resource_id.clone());
149
150	Ok(Response {
151		resource_id: resource_id.clone(),
152		answer: sdp::render_answer(&answer),
153		session: crate::server::AcceptedSession {
154			server: server.clone(),
155			resource_id,
156			session: Some(session),
157			registration: Some(registration),
158			cancel: Some(cancel),
159			role: "whep server",
160			broadcast: None,
161		},
162	})
163}
164
165fn is_sdp(headers: &HeaderMap) -> bool {
166	headers
167		.get(header::CONTENT_TYPE)
168		.and_then(|v| v.to_str().ok())
169		.map(|v| v.eq_ignore_ascii_case("application/sdp"))
170		.unwrap_or(false)
171}
172
173fn status_for(err: &Error) -> StatusCode {
174	match err {
175		Error::InvalidSdp(_) => StatusCode::BAD_REQUEST,
176		Error::UnsupportedCodec(_) => StatusCode::UNSUPPORTED_MEDIA_TYPE,
177		Error::SessionNotFound => StatusCode::NOT_FOUND,
178		_ => StatusCode::INTERNAL_SERVER_ERROR,
179	}
180}
moq_rtc/server/whep.rs

moq_rtc/server/
whep.rs
