use std::sync::Arc;
use std::time::Duration;
use 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 openrtc::broadcast::native_publisher::NativeCarrier;
use openrtc::broadcast::whip_ingest::{H264RtpIngest, OpusRtpIngest};
use openrtc::broadcast::BroadcastSession;
use openrtc::media::{
EncodedMediaSample, MediaCodec, MediaKind, MediaPublicationConfig, PublicationId,
};
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,
}
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(),
}
}
async fn open_broadcast(harness_api: &str) -> Result<(BroadcastSession, NativeCarrier)> {
let code = std::env::var("OPENRTC_LIVE_PAIR_CODE").ok();
let origin =
std::env::var("OPENRTC_LIVE_PUBLIC_ORIGIN").unwrap_or_else(|_| DEFAULT_ORIGIN.into());
openrtc::broadcast::native_publisher::open_broadcast(
harness_api,
&origin,
code.as_deref(),
None,
)
.await
}
#[cfg(test)]
mod origin_tests {
use openrtc::broadcast::native_publisher::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, receiver, _| {
let sender = media.clone();
Box::pin(async move {
tokio::spawn(async move { while receiver.read_rtcp().await.is_ok() {} });
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
}