use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use anyhow::{anyhow, ensure, Context, Result};
use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path, State};
use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::post;
use axum::Router;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine as _;
use clap::Parser;
use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
use openrtc::broadcast::draft14::{Draft14RelayAccess, NativeDraft14Publisher};
use openrtc::broadcast::sfu::{NativeSfuPublisher, SfuControlAccess};
use openrtc::broadcast::whip_ingest::{H264RtpIngest, OpusRtpIngest};
use openrtc::broadcast::{BroadcastPublisherSigner, BroadcastSession, Broadcasts};
use openrtc::media::{
EncodedMediaSample, MediaCodec, MediaKind, MediaPublicationConfig, PublicationId,
};
use serde::Deserialize;
use tokio::sync::{mpsc, watch, Mutex};
use webrtc::api::interceptor_registry::register_default_interceptors;
use webrtc::api::media_engine::{MediaEngine, MIME_TYPE_H264, MIME_TYPE_OPUS};
use webrtc::api::setting_engine::SettingEngine;
use webrtc::api::APIBuilder;
use webrtc::interceptor::registry::Registry;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use webrtc::peer_connection::RTCPeerConnection;
use webrtc::rtp_transceiver::rtp_codec::{
RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType,
};
const DEFAULT_ORIGIN: &str = "http://localhost:5173";
const MAX_SDP_BYTES: usize = 64 * 1024;
const CHANNEL_CAPACITY: usize = 128;
#[derive(Parser)]
struct Args {
#[arg(long, default_value = "http://localhost:8787")]
harness_api: String,
#[arg(long, default_value_t = 8790)]
whip_port: u16,
}
#[derive(Deserialize)]
struct GrantResponse {
grant: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct AccessResponse {
issuer_public_key: String,
relay: RelayResponse,
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
enum RelayResponse {
Moq {
#[serde(rename = "relayUrl")]
relay_url: String,
token: String,
namespace: String,
},
Sfu {
#[serde(rename = "controlUrl")]
control_url: String,
},
}
enum NativeCarrier {
Moq(NativeDraft14Publisher),
Sfu(NativeSfuPublisher),
}
impl NativeCarrier {
async fn flush(&self) -> Result<()> {
match self {
Self::Moq(carrier) => carrier.flush().await,
Self::Sfu(carrier) => carrier.flush().await,
}
}
async fn close(self) {
match self {
Self::Moq(carrier) => carrier.close(),
Self::Sfu(carrier) => carrier.close().await,
}
}
}
fn random_token() -> Result<String> {
let mut bytes = [0_u8; 24];
getrandom::getrandom(&mut bytes).context("WHIP token entropy failed")?;
Ok(URL_SAFE_NO_PAD.encode(bytes))
}
fn whip_token() -> Result<String> {
match std::env::var("OPENRTC_LIVE_WHIP_TOKEN") {
Ok(token) => {
ensure!(
token.len() >= 32
&& token.len() <= 128
&& token
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"_-".contains(&byte)),
"OPENRTC_LIVE_WHIP_TOKEN must be 32-128 base64url characters"
);
Ok(token)
}
Err(_) => random_token(),
}
}
fn key(value: &str) -> Result<[u8; 32]> {
URL_SAFE_NO_PAD
.decode(value)
.context("invalid issuer key")?
.try_into()
.map_err(|_| anyhow!("issuer key must be 32 bytes"))
}
fn checked_api_origin(value: &str, paired: bool) -> Result<url::Url> {
let api = url::Url::parse(value).context("invalid broadcast API URL")?;
let loopback_http = api.scheme() == "http"
&& matches!(api.host_str(), Some("localhost" | "127.0.0.1" | "[::1]"));
ensure!(
(loopback_http || (paired && api.scheme() == "https"))
&& api.username().is_empty()
&& api.password().is_none()
&& api.query().is_none()
&& api.fragment().is_none()
&& (value == api.origin().ascii_serialization()
|| value == format!("{}/", api.origin().ascii_serialization())),
"broadcast API must be loopback HTTP or a paired exact HTTPS origin"
);
Ok(api)
}
async fn open_broadcast(harness_api: &str) -> Result<(BroadcastSession, NativeCarrier)> {
let pairing_code = std::env::var("OPENRTC_LIVE_PAIR_CODE").ok();
let api = checked_api_origin(harness_api, pairing_code.is_some())?;
let origin =
std::env::var("OPENRTC_LIVE_PUBLIC_ORIGIN").unwrap_or_else(|_| DEFAULT_ORIGIN.to_string());
let origin_url = url::Url::parse(&origin).context("invalid Live public origin")?;
ensure!(
matches!(origin_url.scheme(), "http" | "https")
&& origin_url.origin().ascii_serialization() == origin
&& origin_url.username().is_empty()
&& origin_url.password().is_none(),
"OPENRTC_LIVE_PUBLIC_ORIGIN must be an exact HTTP(S) origin"
);
if api.scheme() == "https" {
ensure!(
origin == api.origin().ascii_serialization(),
"paired broadcast origin must match the HTTPS API"
);
}
let mut seed = [0_u8; 32];
getrandom::getrandom(&mut seed).context("native binding key entropy failed")?;
let binding = SigningKey::from_bytes(&seed);
seed.fill(0);
let binding_public_jwk = serde_json::json!({
"kty": "OKP", "crv": "Ed25519",
"x": URL_SAFE_NO_PAD.encode(binding.verifying_key().to_bytes())
});
let publisher_signer = BroadcastPublisherSigner::generate()?;
let publication_key = URL_SAFE_NO_PAD.encode(publisher_signer.verifying_key());
let http = reqwest::Client::new();
let grant_path = if pairing_code.is_some() {
"/api/broadcast/obs/redeem"
} else {
"/api/broadcast/publisher"
};
let publisher: GrantResponse = http
.post(api.join(grant_path)?)
.header("Origin", &origin)
.json(&serde_json::json!({
"code": pairing_code,
"bindingPublicJwk": binding_public_jwk,
"publicationVerifyingKey": publication_key
}))
.send()
.await?
.error_for_status()?
.json()
.await?;
let access: AccessResponse = http
.post(api.join("/api/broadcast/access")?)
.header("Origin", &origin)
.json(&serde_json::json!({
"grant": publisher.grant,
"bindingPublicJwk": binding_public_jwk,
"publicationVerifyingKey": publication_key
}))
.send()
.await?
.error_for_status()?
.json()
.await?;
let issuer = VerifyingKey::from_bytes(&key(&access.issuer_public_key)?)
.context("issuer key was rejected")?;
let now_ms = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64;
let broadcasts = Broadcasts::default();
let challenge = broadcasts.prepare_grant_verification(&publisher.grant, &issuer, now_ms)?;
let proof = binding.sign(&challenge.signing_bytes);
let grant = broadcasts.complete_grant_verification(
&publisher.grant,
&issuer,
&challenge.handle,
&proof.to_bytes(),
now_ms,
)?;
let session = broadcasts.open_publisher(grant, &publisher_signer, now_ms)?;
let carrier = match access.relay {
RelayResponse::Moq {
relay_url,
token,
namespace,
} => {
let relay = Draft14RelayAccess::new(&relay_url, token, namespace)?;
NativeCarrier::Moq(NativeDraft14Publisher::connect_live(session.clone(), relay).await?)
}
RelayResponse::Sfu { control_url } => {
let control_url = api.join(&control_url)?;
ensure!(
control_url.origin() == api.origin() && control_url.scheme() == api.scheme(),
"SFU control must stay on the broadcast API origin"
);
NativeCarrier::Sfu(
NativeSfuPublisher::connect(
session.clone(),
SfuControlAccess {
control_url,
origin,
grant: publisher.grant,
binding_public_jwk,
publication_verifying_key: publication_key,
},
)
.await?,
)
}
};
Ok((session, carrier))
}
#[cfg(test)]
mod origin_tests {
use super::checked_api_origin;
#[test]
fn public_api_requires_pairing_and_https() {
assert!(checked_api_origin("https://live.openrtc.app", true).is_ok());
assert!(checked_api_origin("https://live.openrtc.app", false).is_err());
assert!(checked_api_origin("http://live.openrtc.app", true).is_err());
assert!(checked_api_origin("https://live.openrtc.app/path", true).is_err());
assert!(checked_api_origin("https://user@live.openrtc.app", true).is_err());
assert!(checked_api_origin("http://127.0.0.1:8787", false).is_ok());
}
}
struct MediaEvent {
kind: MediaKind,
sample: EncodedMediaSample,
}
struct WhipState {
token: String,
resource_id: String,
peer: Mutex<Option<Arc<RTCPeerConnection>>>,
media: mpsc::Sender<MediaEvent>,
stopped: watch::Sender<bool>,
}
impl WhipState {
fn authorized(&self, headers: &HeaderMap) -> bool {
headers
.get(header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| value.strip_prefix("Bearer ") == Some(self.token.as_str()))
}
}
fn error(status: StatusCode, message: &str) -> Response {
(status, message.to_string()).into_response()
}
fn cors(mut response: Response) -> Response {
let headers = response.headers_mut();
headers.insert(
header::ACCESS_CONTROL_ALLOW_ORIGIN,
HeaderValue::from_static(DEFAULT_ORIGIN),
);
headers.insert(
header::ACCESS_CONTROL_ALLOW_METHODS,
HeaderValue::from_static("POST, DELETE, OPTIONS"),
);
headers.insert(
header::ACCESS_CONTROL_ALLOW_HEADERS,
HeaderValue::from_static("authorization, content-type"),
);
headers.insert(
header::ACCESS_CONTROL_EXPOSE_HEADERS,
HeaderValue::from_static("location"),
);
response
}
fn offer_has_codec(sdp: &str, codec: &str) -> bool {
sdp.lines().any(|line| {
line.trim_end()
.to_ascii_lowercase()
.starts_with("a=rtpmap:")
&& line.to_ascii_lowercase().contains(codec)
})
}
async fn on_track(
track: Arc<webrtc::track::track_remote::TrackRemote>,
media: mpsc::Sender<MediaEvent>,
) {
let codec = track.codec().capability.mime_type;
let kind = if codec.eq_ignore_ascii_case(MIME_TYPE_H264) {
MediaKind::Video
} else if codec.eq_ignore_ascii_case(MIME_TYPE_OPUS) {
MediaKind::Audio
} else {
eprintln!("WHIP rejected negotiated codec {codec}; select H.264 and Opus in OBS");
return;
};
let mut video = H264RtpIngest::default();
let mut audio = OpusRtpIngest::default();
let mut inspected = 0;
let mut emitted = 0;
while let Ok((packet, _)) = track.read_rtp().await {
if inspected < 8 && !packet.payload.is_empty() {
let head = &packet.payload[..packet.payload.len().min(12)];
eprintln!(
"WHIP {codec} RTP seq={} marker={} bytes={} head={head:02x?}",
packet.header.sequence_number,
packet.header.marker,
packet.payload.len()
);
inspected += 1;
}
let result = if kind == MediaKind::Video {
video.push(
packet.header.sequence_number,
packet.header.timestamp,
packet.header.marker,
&packet.payload,
)
} else {
audio.push(
packet.header.sequence_number,
packet.header.timestamp,
&packet.payload,
)
};
match result {
Ok(Some(sample)) => {
emitted += 1;
if emitted <= 5 || emitted % 100 == 0 {
eprintln!(
"WHIP {codec} produced sample {emitted}, bytes={}, keyframe={}",
sample.payload.len(),
sample.keyframe
);
}
if media.send(MediaEvent { kind, sample }).await.is_err() {
break;
}
}
Ok(None) => {}
Err(cause) => eprintln!("WHIP RTP packet rejected: {cause:#}"),
}
}
}
fn peer_api() -> Result<webrtc::api::API> {
let mut engine = MediaEngine::default();
engine.register_codec(
RTCRtpCodecParameters {
capability: RTCRtpCodecCapability {
mime_type: MIME_TYPE_OPUS.to_owned(),
clock_rate: 48_000,
channels: 2,
sdp_fmtp_line: "minptime=10;useinbandfec=1".to_owned(),
..Default::default()
},
payload_type: 111,
..Default::default()
},
RTPCodecType::Audio,
)?;
engine.register_codec(
RTCRtpCodecParameters {
capability: RTCRtpCodecCapability {
mime_type: MIME_TYPE_H264.to_owned(),
clock_rate: 90_000,
sdp_fmtp_line:
"level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f"
.to_owned(),
..Default::default()
},
payload_type: 102,
..Default::default()
},
RTPCodecType::Video,
)?;
let registry = register_default_interceptors(Registry::new(), &mut engine)?;
let mut settings = SettingEngine::default();
settings.set_include_loopback_candidate(true);
settings.set_ip_filter(Box::new(|ip| ip.is_loopback()));
Ok(APIBuilder::new()
.with_media_engine(engine)
.with_interceptor_registry(registry)
.with_setting_engine(settings)
.build())
}
async fn post_whip(
State(state): State<Arc<WhipState>>,
headers: HeaderMap,
body: Bytes,
) -> Response {
cors(post_whip_inner(state, headers, body).await)
}
async fn post_whip_inner(state: Arc<WhipState>, headers: HeaderMap, body: Bytes) -> Response {
if !state.authorized(&headers) {
return error(StatusCode::UNAUTHORIZED, "WHIP bearer token required");
}
if headers
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
!= Some("application/sdp")
{
return error(
StatusCode::UNSUPPORTED_MEDIA_TYPE,
"WHIP requires application/sdp",
);
}
if body.is_empty() || body.len() > MAX_SDP_BYTES {
return error(StatusCode::BAD_REQUEST, "WHIP SDP is empty or too large");
}
let Ok(sdp) = std::str::from_utf8(&body) else {
return error(StatusCode::BAD_REQUEST, "WHIP SDP must be UTF-8");
};
if !offer_has_codec(sdp, "h264/90000") || !offer_has_codec(sdp, "opus/48000") {
return error(
StatusCode::NOT_ACCEPTABLE,
"OBS must offer H.264 video and Opus audio; use WebRTC/WHIP output",
);
}
let mut slot = state.peer.lock().await;
if slot.is_some() {
return error(StatusCode::CONFLICT, "one WHIP publisher is already active");
}
let result: Result<(Arc<RTCPeerConnection>, String)> = async {
let peer = Arc::new(
peer_api()?
.new_peer_connection(RTCConfiguration::default())
.await?,
);
let media = state.media.clone();
peer.on_track(Box::new(move |track, _, _| {
let sender = media.clone();
Box::pin(async move {
tokio::spawn(on_track(track, sender));
})
}));
let stopped = state.stopped.clone();
peer.on_peer_connection_state_change(Box::new(move |connection| {
let stopped = stopped.clone();
Box::pin(async move {
if matches!(
connection,
RTCPeerConnectionState::Failed | RTCPeerConnectionState::Closed
) {
let _ = stopped.send(true);
}
})
}));
peer.set_remote_description(RTCSessionDescription::offer(sdp.to_owned())?)
.await?;
let answer = peer.create_answer(None).await?;
let mut gathering = peer.gathering_complete_promise().await;
peer.set_local_description(answer).await?;
tokio::time::timeout(Duration::from_secs(10), gathering.recv())
.await
.context("loopback ICE gathering timed out")?;
let local = peer
.local_description()
.await
.context("WHIP answer has no SDP")?;
Ok((peer, local.sdp))
}
.await;
let (peer, answer) = match result {
Ok(result) => result,
Err(cause) => {
eprintln!("WHIP offer rejected: {cause:#}");
return error(
StatusCode::BAD_REQUEST,
"WHIP offer could not be negotiated",
);
}
};
*slot = Some(peer);
println!("OBS connected through loopback WHIP");
let mut response = (StatusCode::CREATED, answer).into_response();
response.headers_mut().insert(
header::CONTENT_TYPE,
HeaderValue::from_static("application/sdp"),
);
response.headers_mut().insert(
header::LOCATION,
HeaderValue::from_str(&format!("/whip/{}", state.resource_id))
.expect("URL-safe resource ID"),
);
response
}
async fn delete_whip(
State(state): State<Arc<WhipState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> Response {
cors(delete_whip_inner(state, id, headers).await)
}
async fn delete_whip_inner(state: Arc<WhipState>, id: String, headers: HeaderMap) -> Response {
if !state.authorized(&headers) {
return error(StatusCode::UNAUTHORIZED, "WHIP bearer token required");
}
if id != state.resource_id {
return error(StatusCode::NOT_FOUND, "unknown WHIP resource");
}
let peer = state.peer.lock().await.take();
if let Some(peer) = peer {
let _ = peer.close().await;
let _ = state.stopped.send(true);
StatusCode::NO_CONTENT.into_response()
} else {
error(StatusCode::NOT_FOUND, "WHIP publisher is not active")
}
}
fn publication_id() -> Result<PublicationId> {
let mut bytes = [0_u8; 16];
getrandom::getrandom(&mut bytes).context("publication ID entropy failed")?;
Ok(PublicationId(bytes))
}
async fn publish_media(
session: &BroadcastSession,
carrier: &NativeCarrier,
receiver: &mut mpsc::Receiver<MediaEvent>,
mut stopped: watch::Receiver<bool>,
) -> Result<()> {
let mut video_id = None;
let mut audio_id = None;
loop {
tokio::select! {
event = receiver.recv() => {
let Some(event) = event else { break; };
let id_slot = if event.kind == MediaKind::Video { &mut video_id } else { &mut audio_id };
let id = if let Some(id) = *id_slot {
id
} else {
let config = MediaPublicationConfig {
publication_id: publication_id()?,
media_generation: 0,
kind: event.kind,
codec: if event.kind == MediaKind::Video { MediaCodec::H264 } else { MediaCodec::Opus },
clock_rate: if event.kind == MediaKind::Video { 90_000 } else { 48_000 },
coded_width: (event.kind == MediaKind::Video).then_some(1280),
coded_height: (event.kind == MediaKind::Video).then_some(720),
channels: (event.kind == MediaKind::Audio).then_some(2),
};
let publication = session.begin_publication("main", config)?;
*id_slot = Some(publication.publication_id);
publication.publication_id
};
session.publish(id, event.sample)?;
carrier.flush().await?;
}
change = stopped.changed() => {
if change.is_err() || *stopped.borrow() { break; }
}
}
}
for id in [video_id, audio_id].into_iter().flatten() {
session.retire_publication(id)?;
}
carrier.flush().await?;
Ok(())
}
#[tokio::main]
async fn main() -> Result<()> {
let _ = rustls::crypto::ring::default_provider().install_default();
let args = Args::parse();
let (session, carrier) = open_broadcast(&args.harness_api).await?;
let (media_tx, mut media_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (stopped_tx, stopped_rx) = watch::channel(false);
let state = Arc::new(WhipState {
token: whip_token()?,
resource_id: random_token()?,
peer: Mutex::new(None),
media: media_tx,
stopped: stopped_tx,
});
let app = Router::new()
.route(
"/whip",
post(post_whip).options(|| async { cors(StatusCode::NO_CONTENT.into_response()) }),
)
.route("/whip/{id}", axum::routing::delete(delete_whip))
.layer(DefaultBodyLimit::max(MAX_SDP_BYTES))
.with_state(state.clone());
let listener = tokio::net::TcpListener::bind(("127.0.0.1", args.whip_port)).await?;
println!("OBS WHIP server: http://127.0.0.1:{}/whip", args.whip_port);
println!("OBS bearer token (stream key): {}", state.token);
println!(
"Configure H.264 constrained baseline 720p30, Opus 48 kHz, 1 s keyframes, no B-frames"
);
println!("Waiting for OBS and a browser viewer; Ctrl-C stops this bridge");
let mut server = tokio::spawn(async move { axum::serve(listener, app).await });
let signal = state.stopped.clone();
let cancel_task = tokio::spawn(async move {
let _ = tokio::signal::ctrl_c().await;
let _ = signal.send(true);
});
let result = tokio::select! {
result = publish_media(&session, &carrier, &mut media_rx, stopped_rx) => result,
result = &mut server => result.context("WHIP server stopped")?.map_err(Into::into),
};
cancel_task.abort();
if let Some(peer) = state.peer.lock().await.take() {
let _ = peer.close().await;
}
server.abort();
carrier.close().await;
result
}