diff --git a/Cargo.lock b/Cargo.lock index 0597e5d..5a7f956 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1151,7 +1151,6 @@ checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" dependencies = [ "futures-channel", "futures-core", - "futures-executor", "futures-io", "futures-sink", "futures-task", @@ -1202,17 +1201,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" -[[package]] -name = "futures-macro" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "futures-sink" version = "0.3.32" @@ -1231,10 +1219,8 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ - "futures-channel", "futures-core", "futures-io", - "futures-macro", "futures-sink", "futures-task", "memchr", @@ -2916,16 +2902,12 @@ dependencies = [ "chrono", "dashmap", "entity", - "futures", "migration", "opus", - "rand 0.10.2", - "regex", "rml_rtmp", "rubato", "sea-orm", "serde", - "serde_json", "str0m", "symphonia", "sysinfo", diff --git a/crates/entity/src/stream_key.rs b/crates/entity/src/stream_key.rs index beb1d33..c21aec9 100644 --- a/crates/entity/src/stream_key.rs +++ b/crates/entity/src/stream_key.rs @@ -91,27 +91,3 @@ impl Entity { .await } } - -impl ActiveModel { - pub async fn change_label_value( - mut self, - db: &DatabaseConnection, - value: String, - ) -> Result { - self.label = Set(value); - self.update(db).await - } - - pub async fn password( - mut self, - db: &DatabaseConnection, - value: String, - ) -> Result { - self.password = if value.is_empty() { - Set(None) - } else { - Set(Some(value)) - }; - self.update(db).await - } -} diff --git a/crates/entity/src/stream_session.rs b/crates/entity/src/stream_session.rs index 637429d..32fcf6a 100644 --- a/crates/entity/src/stream_session.rs +++ b/crates/entity/src/stream_session.rs @@ -50,17 +50,6 @@ impl Model { .insert(db) .await } - pub async fn get_stream_session( - db: &DatabaseConnection, - stream_session_id: i32, - ) -> Result { - Entity::find_by_id(stream_session_id) - .one(db) - .await? - .ok_or(DbErr::RecordNotFound(format!( - "stream_session {stream_session_id}" - ))) - } pub async fn get_all_active_sessions(db: &DatabaseConnection) -> Result, DbErr> { Ok(Entity::find() .filter(Column::EndedAt.is_null()) diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 4bde874..98c410d 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -15,20 +15,17 @@ path = "src/main.rs" async-broadcast = "0.7.2" bytes = "1" dashmap = "6.2.1" -rand = "0.10.1" rml_rtmp = "0.8.0" str0m = "0.20.0" tokio = { version = "1", features = ["full"] } axum = { version = "0.8", features = ["macros"] } serde = { version = "1.0.228", features = ["serde_derive"] } -serde_json = "1.0.150" sea-orm = { version = "1", features = [ "sqlx-sqlite", "runtime-tokio-rustls", "macros" ] } entity = {path = "../entity"} migration = {path = "../migration"} argon2 = "0.5.3" uuid = { version = "1.23.3", features = ["v4"] } tower-http = { version = "0.6", features = ["cors"] } -futures = "0.3.32" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } symphonia = { version = "0.5", features = ["aac"] } @@ -36,7 +33,6 @@ opus = "0.3.1" rubato = "3.0.0" chrono = "0.4.45" time = "0.3" -regex = "1" thiserror = "2.0.18" sysinfo = "0.36" axum-extra = { version = "0.12.6", features = ["cookie"] } diff --git a/crates/server/src/audio.rs b/crates/server/src/audio.rs index d3294df..ba01814 100644 --- a/crates/server/src/audio.rs +++ b/crates/server/src/audio.rs @@ -1,4 +1,4 @@ -use std::{error::Error, fmt::Display}; +use std::error::Error; use bytes::Bytes; use rubato::{audioadapter_buffers::direct::InterleavedSlice, Fft, Resampler}; @@ -95,23 +95,6 @@ pub struct AACParser { decoder: Option>, } -#[derive(Debug)] -enum AudioParseError { - InvalidCodec, - NoConfigPacket, -} - -impl Display for AudioParseError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::InvalidCodec => write!(f, "Not the right codec provided"), - Self::NoConfigPacket => write!(f, "No config packet cached"), - } - } -} - -impl Error for AudioParseError {} - // Byte 0: upper nibble = sound format (10 = AAC) // Byte 1: 0 = AudioSpecificConfig, 1 = raw AAC frame impl AACParser { @@ -129,7 +112,7 @@ impl AACParser { } if (bytes[0] >> 4) != 10 { - return Err(Box::new(AudioParseError::InvalidCodec)); + return Err("not the right codec provided".into()); } if bytes[1] == 0 { @@ -142,10 +125,7 @@ impl AACParser { return Ok(None); } - let decoder = self - .decoder - .as_mut() - .ok_or(AudioParseError::NoConfigPacket)?; + let decoder = self.decoder.as_mut().ok_or("no config packet cached")?; let packet = Packet::new_from_boxed_slice( 0, diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 17749fd..8d44e2f 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -1,28 +1,20 @@ use std::{ net::SocketAddr, sync::{ - Arc, LazyLock, + Arc, atomic::{AtomicI32, Ordering}, }, - time::Duration, + time::{Duration, Instant}, }; use axum_extra::extract::{CookieJar, cookie::Cookie}; -use regex::Regex; -// The Rust `regex` crate is guaranteed linear-time and therefore does NOT -// support lookaround, so the JS source pattern -// /^(?=.*[A-Za-z])[A-Za-z0-9_-]{1,67}$/ -// cannot be ported verbatim. The `(?=.*[A-Za-z])` lookahead only means -// "must contain at least one letter" — we drop it from the pattern and -// enforce that condition with a separate `.chars().any(..)` check below. -static KEY_RE: LazyLock = LazyLock::new(|| Regex::new(r"^[A-Za-z0-9'-]{1,67}$").unwrap()); - -// Allowed charset for stream-key passwords and custom IDs: letters, numbers, -// dashes and apostrophes only — no spaces. Mirrors the frontend -// `/^[A-Za-z0-9'-]{0,67}$/` used by the keys-page popups; empty clears the value. -static PASSWORD_RE: LazyLock = - LazyLock::new(|| Regex::new(r"^[A-Za-z0-9'-]{0,67}$").unwrap()); +/// Allowed charset for labels, passwords and custom IDs: letters, numbers, +/// dashes and apostrophes — no spaces. Mirrors the frontend +/// `/^[A-Za-z0-9'-]{0,67}$/` used by the keys-page popups. +fn valid_charset(s: &str) -> bool { + s.len() <= 67 && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '\'') +} use axum::{ Json, Router, @@ -58,23 +50,15 @@ use crate::{ const MAX_USERNAME_LEN: usize = 32; const MAX_LABEL_LEN: usize = 64; -pub struct HttpServerConfig { - pub signup_code: String, -} - -pub struct ServerInfo { - pub version: &'static str, - pub start_time: std::time::Instant, -} - pub struct HttpServer { pub offer_tx: Sender<(i32, i32, String)>, pub accept_rx: async_broadcast::InactiveReceiver<(i32, Result)>, pub appstate: Arc>, pub request_count: AtomicI32, pub db: DatabaseConnection, - pub config: Arc, - pub info: Arc, + pub signup_code: String, + pub version: &'static str, + pub start_time: Instant, } impl HttpServer { @@ -112,8 +96,6 @@ impl HttpServer { .get(get_all_stream_keys) .patch(edit_stream_key), ) - // .route("/api/admin/server_stats", get(todo!())) - // .route("/api/whip", post(handle_whip_injest)) .route("/api/login", post(login_handler)) .route("/api/stream/{slug}", post(stream_handler)) .route("/api/whip", post(handle_whip_injest)) @@ -121,14 +103,8 @@ impl HttpServer { "/api/whip/{slug}", delete(handle_whip_injest_delete).patch(handle_whip_injest_patch), ) - // .route( - // "/api/whip/{slug}", - // delete(handle_whip_injest_delete).patch(handle_whip_injest_patch), - // ) .route("/api/meow", get(meow_handler)) .route("/api/health", get(health_handler)) - .route("/api/uptime", get(uptime_handler)) - .route("/api/version", get(version_handler)) .route("/api/stats", get(stats_handler)) .layer(cors) .with_state(state); @@ -172,25 +148,23 @@ async fn edit_stream_key( ) -> Result { // Trim surrounding whitespace so labels aren't stored with leading/trailing spaces. let new_label = payload.new.as_deref().map(str::trim); - // Length (1..=67) and allowed charset ([A-Za-z0-9'-], no spaces). + // Length (1..=67), allowed charset, and at least one letter. if let Some(label) = new_label { - if !KEY_RE.is_match(label) { + if label.is_empty() || !valid_charset(label) { return Err(HttpError::BadRequest("invalid label".into())); } - // Replaces the JS lookahead: label must contain at least one letter. if !label.chars().any(|c| c.is_ascii_alphabetic()) { return Err(HttpError::BadRequest("label must contain a letter".into())); } } - // Custom IDs and passwords share a charset: letters, numbers, dashes, - // apostrophes — no spaces. Empty values are allowed (they clear the field). + // Empty values are allowed (they clear the field). if let Some(custom_id) = payload.custom_id.as_deref() { - if !custom_id.is_empty() && !PASSWORD_RE.is_match(custom_id) { + if !custom_id.is_empty() && !valid_charset(custom_id) { return Err(HttpError::BadRequest("invalid custom id".into())); } } if let Some(pwd) = payload.password.as_deref() { - if !pwd.is_empty() && !PASSWORD_RE.is_match(pwd) { + if !pwd.is_empty() && !valid_charset(pwd) { return Err(HttpError::BadRequest("invalid password".into())); } } @@ -288,22 +262,6 @@ struct CreateStreamKeyBody { label: String, } -struct SessionCookie(Option); - -impl FromRequestParts for SessionCookie -where - S: Send + Sync, -{ - type Rejection = HttpError; - - async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { - let jar = CookieJar::from_request_parts(parts, _state).await.unwrap(); - Ok(SessionCookie( - jar.get("session").map(|c| c.value().to_string()), - )) - } -} - struct AuthUser(entity::users::Model); impl FromRequestParts> for AuthUser { @@ -409,13 +367,10 @@ fn cookie_for_token(token: &str, dev: bool) -> Cookie<'static> { #[axum::debug_handler] async fn login_handler( - session: SessionCookie, State(state): State>, DevFlag(dev): DevFlag, Json(payload): Json, ) -> Result { - tracing::debug!(session = ?session.0, "login: existing session cookie"); - let user = users::Entity::find_by_username(&state.db, payload.username.clone()) .await? .ok_or_else(|| { @@ -447,13 +402,11 @@ struct CreateUserForm { } async fn create_user_handler( - session: SessionCookie, State(state): State>, DevFlag(dev): DevFlag, Json(payload): Json, ) -> Result { - tracing::debug!(session = ?session.0, "create_user: existing session cookie"); - if state.config.signup_code.is_empty() || payload.ref_token != state.config.signup_code { + if state.signup_code.is_empty() || payload.ref_token != state.signup_code { warn!(username = %payload.username, "signup rejected: invalid signup code"); return Err(HttpError::Unauthorized); } @@ -596,43 +549,17 @@ struct HealthResponse { async fn health_handler( State(state): State>, ) -> Result { - let uptime = state.info.start_time.elapsed().as_secs(); + let uptime = state.start_time.elapsed().as_secs(); Ok(( StatusCode::OK, Json(HealthResponse { status: "ok", uptime_seconds: uptime, - version: state.info.version, + version: state.version, }), )) } -#[derive(Serialize)] -struct UptimeResponse { - uptime_seconds: u64, -} - -async fn uptime_handler( - State(state): State>, -) -> Result { - let uptime = state.info.start_time.elapsed().as_secs(); - Ok(( - StatusCode::OK, - Json(UptimeResponse { - uptime_seconds: uptime, - }), - )) -} - -async fn version_handler( - State(state): State>, -) -> Result { - Ok(( - StatusCode::OK, - Json(serde_json::json!({ "version": state.info.version })), - )) -} - #[derive(Serialize)] struct StatsResponse { version: &'static str, @@ -645,7 +572,7 @@ struct StatsResponse { async fn stats_handler( State(state): State>, ) -> Result { - let uptime = state.info.start_time.elapsed().as_secs(); + let uptime = state.start_time.elapsed().as_secs(); let mut system = System::new(); system.refresh_cpu_all(); let cpu_usage = system.global_cpu_usage(); @@ -657,7 +584,7 @@ async fn stats_handler( Ok(( StatusCode::OK, Json(StatsResponse { - version: state.info.version, + version: state.version, uptime_seconds: uptime, cpu_usage_percent: cpu_usage, active_streams, diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 7bb547a..aeec6e2 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -18,8 +18,8 @@ use uuid::Uuid; use crate::{ audio::OpusAudioFrame, codec::VideoFrame, - http::{HttpServer, HttpServerConfig, ServerInfo}, - webrtc_proxy::{WebRtcProxyConfig, WebrtcProxy}, + http::HttpServer, + webrtc_proxy::WebrtcProxy, }; const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -34,7 +34,6 @@ mod webrtc; mod webrtc_ingest; mod webrtc_proxy; -// #[derive(Debug)] pub struct AppState { pub stream_sessions: Arc>, pub webrtc_proxy: WebrtcProxy, @@ -47,6 +46,19 @@ pub enum StreamCodec { AV1, } +impl StreamCodec { + /// Map str0m's codec enum to ours; None for non-video codecs. + pub fn from_str0m(c: str0m::format::Codec) -> Option { + use str0m::format::Codec; + match c { + Codec::H264 => Some(Self::H264), + Codec::H265 => Some(Self::H265), + Codec::Av1 => Some(Self::AV1), + _ => None, + } + } +} + pub struct StreamSession { pub stream_key_id: i32, pub stream_key_label: String, @@ -92,12 +104,10 @@ async fn main() -> Result<(), Box> { stream_session::Model::clean_unended_streams(&db) .await .unwrap(); - let proxyconfig = WebRtcProxyConfig { - proxy_port: env::var("RTC_PORT") - .unwrap_or("6969".into()) - .parse() - .expect("RTC_PORT needs to be a number (i32)"), - }; + let proxyconfig = env::var("RTC_PORT") + .unwrap_or("6969".into()) + .parse() + .expect("RTC_PORT needs to be a number (i32)"); let proxy = webrtc_proxy::WebrtcProxy::new(proxyconfig).await.unwrap(); let appstate = Arc::new(Mutex::new(AppState { @@ -120,17 +130,12 @@ async fn main() -> Result<(), Box> { appstate: appstate.clone(), request_count: std::sync::atomic::AtomicI32::new(0), db: db.clone(), - config: HttpServerConfig { - signup_code: env::var("SIGNUP_CODE").unwrap_or_else(|_| { - warn!("SIGNUP_CODE not set; signup will be disabled"); - String::new() - }), - } - .into(), - info: Arc::new(ServerInfo { - version: SERVER_VERSION, - start_time: std::time::Instant::now(), + signup_code: env::var("SIGNUP_CODE").unwrap_or_else(|_| { + warn!("SIGNUP_CODE not set; signup will be disabled"); + String::new() }), + version: SERVER_VERSION, + start_time: std::time::Instant::now(), }; let app = appstate.lock().await; @@ -164,7 +169,6 @@ async fn main() -> Result<(), Box> { res = workers.join_next() => { if let Some(Err(e)) = res { tracing::error!("worker panicked: {:?}", e); - // fs::File:: } else { tracing::error!("a worker exited unexpectedly"); } diff --git a/crates/server/src/rtmp.rs b/crates/server/src/rtmp.rs index 10c30e8..97b7a5f 100644 --- a/crates/server/src/rtmp.rs +++ b/crates/server/src/rtmp.rs @@ -14,7 +14,7 @@ use tokio::{ net::{TcpListener, TcpStream}, time::timeout, }; -use tracing::{debug, info, trace, warn}; +use tracing::{debug, info, warn}; use uuid::Uuid; use crate::{ @@ -134,7 +134,6 @@ impl Rtmp { // ponytail: fixed 512; per-stream dynamic sizing if memory ever matters. let (mut video_tx, video_rx) = broadcast::>(512); let (mut audio_tx, audio_rx) = broadcast::>(512); - // video_rx.cycle video_tx.set_overflow(true); audio_tx.set_overflow(true); @@ -369,9 +368,6 @@ impl Rtmp { } current_stream_key_id = None; } - // TODO: We can totally replace the broadcast with a - // circular_buff - // Arc>> ServerSessionEvent::VideoDataReceived { data, timestamp, .. } => match Self::parse_video_codec(&data) { @@ -385,52 +381,13 @@ impl Rtmp { } codec_stamped = true; } - match codec { - StreamCodec::H264 => { - let p = parser.get_or_insert_with(|| { - Box::new(H264CodecParser::new()) - }); - if let Some(frame) = p.parse(&data, timestamp.value) - { - video_tx.broadcast(Arc::new(frame)).await.ok(); - } - } - StreamCodec::H265 => { - let p = parser.get_or_insert_with(|| { - Box::new(H265CodecParser::new()) - }); - if let Some(frame) = p.parse(&data, timestamp.value) - { - video_tx.broadcast(Arc::new(frame)).await.ok(); - } - } - StreamCodec::AV1 => { - let p = parser.get_or_insert_with(|| { - Box::new(Av1CodecParser::new()) - }); - let pkt_type = data[0] & 0x0F; - match p.parse(&data, timestamp.value) { - Some(frame) => { - trace!( - pkt_type, - is_keyframe = frame.is_keyframe, - ts = frame.timestamp_ms, - bytes = frame.data.len(), - "AV1 frame → broadcast" - ); - video_tx - .broadcast(Arc::new(frame)) - .await - .ok(); - } - None => { - trace!( - pkt_type, - "AV1 packet produced no frame (seq header or unknown type)" - ); - } - } - } + let p = parser.get_or_insert_with(|| match codec { + StreamCodec::H264 => Box::new(H264CodecParser::new()), + StreamCodec::H265 => Box::new(H265CodecParser::new()), + StreamCodec::AV1 => Box::new(Av1CodecParser::new()), + }); + if let Some(frame) = p.parse(&data, timestamp.value) { + video_tx.broadcast(Arc::new(frame)).await.ok(); } } Err(_err) => { diff --git a/crates/server/src/webrtc.rs b/crates/server/src/webrtc.rs index 4ec944e..8191c66 100644 --- a/crates/server/src/webrtc.rs +++ b/crates/server/src/webrtc.rs @@ -58,9 +58,6 @@ impl Webrtc { continue; } - // let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - // let local_addr = socket.local_addr().unwrap(); - let local_addr = self.proxy.public_addr(); let session = self.sessions_ref.get(&stream_id); diff --git a/crates/server/src/webrtc_ingest.rs b/crates/server/src/webrtc_ingest.rs index 5c70b00..0bbe726 100644 --- a/crates/server/src/webrtc_ingest.rs +++ b/crates/server/src/webrtc_ingest.rs @@ -33,18 +33,21 @@ use crate::{ /// Kill a WHIP publish if it delivers no media (video or audio) this long. const NO_MEDIA_TIMEOUT: Duration = Duration::from_secs(30); +fn bearer_token(headers: &HeaderMap) -> Option { + headers + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(|s| s.trim().to_string()) +} + pub async fn handle_whip_injest_delete( State(state): State>, ConnectInfo(remote): ConnectInfo, Path(_slug): Path, headers: HeaderMap, ) -> Result { - let token = headers - .get(header::AUTHORIZATION) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.strip_prefix("Bearer ")) - .map(|s| s.trim().to_string()); - let Some(token) = token else { + let Some(token) = bearer_token(&headers) else { warn!("Whip: missing bearer token from {}", remote); return Err(HttpError::Unauthorized); }; @@ -120,12 +123,7 @@ pub async fn handle_whip_injest( }; let public_addr = state.appstate.lock().await.webrtc_proxy.public_addr(); - let token = headers - .get(header::AUTHORIZATION) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.strip_prefix("Bearer ")) - .map(|s| s.trim().to_string()); - let Some(token) = token else { + let Some(token) = bearer_token(&headers) else { warn!("Whip: missing bearer token from {}", remote); return err(StatusCode::UNAUTHORIZED, "missing bearer token"); }; @@ -187,17 +185,13 @@ pub async fn handle_whip_injest( cc.clear(); cc.enable_opus(true); match &stream_codec { - Some(crate::StreamCodec::H264) => { - info!("Whip: enabling H.264 codec"); - cc.enable_h264(true); - } - Some(crate::StreamCodec::H265) => { - info!("Whip: enabling H.265 codec"); - cc.enable_h265(true); - } - Some(crate::StreamCodec::AV1) => { - info!("Whip: enabling AV1 codec"); - cc.enable_av1(true); + Some(codec) => { + info!("Whip: enabling {codec:?} codec"); + match codec { + crate::StreamCodec::H264 => cc.enable_h264(true), + crate::StreamCodec::H265 => cc.enable_h265(true), + crate::StreamCodec::AV1 => cc.enable_av1(true), + } } None => { warn!("Whip: no video codec detected in offer, enabling H.264 as fallback"); @@ -366,12 +360,6 @@ pub async fn handle_whip_injest( .status(StatusCode::CREATED) .header(header::CONTENT_TYPE, "application/sdp") .header(header::LOCATION, &location) - // // Provide a STUN server via Link header so OBS can gather - // // ICE candidates even without explicit STUN configuration. - // .header( - // header::LINK, - // "; rel=\"ice-server\"", - // ) .body(axum::body::Body::from(answer_sdp)) .unwrap() } @@ -422,7 +410,6 @@ async fn detach_inject_rtc( let mut audio_mid: Option = None; let mut video_tx: Option>> = None; let mut audio_tx: Option>> = None; - let mut _connected = false; let mut disconnect_timer: Option = None; loop { @@ -525,7 +512,6 @@ async fn detach_inject_rtc( video_tx = Some(session.frame_channel.clone()); audio_tx = Some(session.audio_channel.clone()); } - _connected = true; } _ => {} }, @@ -610,22 +596,14 @@ async fn detach_inject_rtc( pub fn extract_negotiated_codec_info( answer: &str0m::change::SdpAnswer, ) -> (Option, Option, Option) { - use str0m::format::Codec; - for line in answer.media_lines.iter() { - // Iterate rtp_params on every m-line; video check via - // codec.is_video() avoids needing str0m's private MediaType. for p in line.rtp_params() { if p.spec().codec.is_video() { - let pt = Some(*p.pt()); - let profile = p.spec().format.profile_level_id; - let codec = match p.spec().codec { - Codec::H264 => Some(crate::StreamCodec::H264), - Codec::H265 => Some(crate::StreamCodec::H265), - Codec::Av1 => Some(crate::StreamCodec::AV1), - _ => None, - }; - return (codec, pt, profile); + return ( + crate::StreamCodec::from_str0m(p.spec().codec), + Some(*p.pt()), + p.spec().format.profile_level_id, + ); } } } @@ -634,24 +612,14 @@ pub fn extract_negotiated_codec_info( /// Extract the video codec from an SDP offer's media lines. /// -/// Walks every m= line's `rtp_params()`, maps str0m's `Codec` enum -/// to our `StreamCodec`. Returns the first video codec found. +/// Walks every m= line's `rtp_params()`; returns the first video codec found. pub fn video_codec_from_sdp_offer( sdp_offer: &str0m::change::SdpOffer, ) -> Option { - use str0m::format::Codec; - - // SdpOffer derefs to Sdp, which has pub media_lines. - // MediaLine is pub(crate) in str0m, but we can call its pub methods. for line in sdp_offer.media_lines.iter() { for p in line.rtp_params() { if p.spec().codec.is_video() { - return match p.spec().codec { - Codec::H264 => Some(crate::StreamCodec::H264), - Codec::H265 => Some(crate::StreamCodec::H265), - Codec::Av1 => Some(crate::StreamCodec::AV1), - _ => None, - }; + return crate::StreamCodec::from_str0m(p.spec().codec); } } } diff --git a/crates/server/src/webrtc_proxy.rs b/crates/server/src/webrtc_proxy.rs index ed84e61..9f5db81 100644 --- a/crates/server/src/webrtc_proxy.rs +++ b/crates/server/src/webrtc_proxy.rs @@ -14,10 +14,6 @@ use tokio::{ use tracing::{debug, error, info, trace, warn}; use uuid::Uuid; -pub struct WebRtcProxyConfig { - pub proxy_port: i32, -} - #[derive(Clone)] pub struct WebrtcProxy { clients_ufrag: Arc>>, @@ -34,8 +30,8 @@ pub struct WebrtcProxy { const STUN_MAGIC: u32 = 0x2112A442; impl WebrtcProxy { - pub async fn new(config: WebRtcProxyConfig) -> Result> { - let sock = UdpSocket::bind(format!("0.0.0.0:{}", config.proxy_port)).await?; + pub async fn new(proxy_port: i32) -> Result> { + let sock = UdpSocket::bind(format!("0.0.0.0:{}", proxy_port)).await?; let port = sock.local_addr()?.port(); let public_ip = match env::var("PUBLIC_DOMAIN") @@ -165,10 +161,6 @@ impl WebrtcProxy { self.clients_ufrag.insert(ufrag, tx); (self.socket.clone(), rx) } - pub fn local_addr(&self) -> SocketAddr { - self.socket.local_addr().unwrap() - } - pub fn public_addr(&self) -> SocketAddr { self.public_addr } @@ -180,26 +172,30 @@ impl WebrtcProxy { if magic != STUN_MAGIC { return None; } - // attribies start at 20 - let mut pos = 20usize; - while (pos + 4) <= b.len() { - let attr_type: u16 = u16::from_be_bytes(b[pos..pos + 2].try_into().ok()?); - let attr_len: u16 = u16::from_be_bytes(b[pos + 2..pos + 4].try_into().ok()?); - pos += 4; - if attr_type == 0x0006 { - let value = - std::str::from_utf8(b[pos..pos + (attr_len as usize)].try_into().ok()?).ok()?; - let mut parts = value.split(':'); - let first = parts.next()?.to_string(); - let second = parts.next().map(|s| s.to_string()); - return Some((first, second)); - } - pos += (attr_len as usize + 3) & !3; - } - None + let value = stun_attributes(b).find(|(t, _)| *t == 0x0006)?.1; + let value = std::str::from_utf8(value).ok()?; + let mut parts = value.split(':'); + Some(( + parts.next()?.to_string(), + parts.next().map(|s| s.to_string()), + )) } } +/// Iterate `(attr_type, attr_value)` pairs over a STUN message's attributes +/// (RFC 5389 §15: header is 20 bytes, each attribute padded to a 4-byte boundary). +fn stun_attributes(data: &[u8]) -> impl Iterator { + let mut pos = 20usize; + std::iter::from_fn(move || { + let attr_type = u16::from_be_bytes(data.get(pos..pos + 2)?.try_into().ok()?); + let attr_len = u16::from_be_bytes(data.get(pos + 2..pos + 4)?.try_into().ok()?) as usize; + pos += 4; + let value = data.get(pos..pos + attr_len)?; + pos += (attr_len + 3) & !3; + Some((attr_type, value)) + }) +} + /// IP of the interface holding the default route, via a UDP connect() trick: /// connect() only does a route lookup (no packets sent), so the kernel binds /// the source address the OS would use for outbound traffic. @@ -251,27 +247,15 @@ fn parse_xor_mapped_address(data: &[u8]) -> Option { return None; } - let mut pos = 20usize; - while pos + 4 <= data.len() { - let attr_type = u16::from_be_bytes(data[pos..pos + 2].try_into().ok()?); - let attr_len = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?) as usize; - pos += 4; - if pos + attr_len > data.len() { - break; - } - if attr_type == 0x0020 && attr_len >= 8 { - // byte 0: reserved, byte 1: family (0x01=IPv4, 0x02=IPv6) - let family = data[pos + 1]; - let x_port = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?); - let _ = x_port ^ (STUN_MAGIC >> 16) as u16; // port (unused here) - - if family == 0x01 { - let x_addr = u32::from_be_bytes(data[pos + 4..pos + 8].try_into().ok()?); - let addr = x_addr ^ STUN_MAGIC; - return Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(addr))); - } - } - pos += (attr_len + 3) & !3; + let value = stun_attributes(data).find(|(t, _)| *t == 0x0020)?.1; + if value.len() < 8 { + return None; + } + // byte 0: reserved, byte 1: family (0x01=IPv4, 0x02=IPv6) + if value[1] == 0x01 { + let x_addr = u32::from_be_bytes(value[4..8].try_into().ok()?); + Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(x_addr ^ magic))) + } else { + None } - None }