use crate::error::{MultimuxError, Result};
use crate::output::OutputKind;
use broadcast_auth::Credentials;
use serde::Deserialize;
use std::net::{IpAddr, SocketAddr};
use std::path::Path;
fn default_outputs() -> Vec<OutputKind> {
vec![OutputKind::LlHls]
}
#[non_exhaustive]
#[derive(Clone, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum InputSpec {
Rtsp {
url: String,
#[serde(default)]
auth: Option<AuthSpec>,
},
Rtp {
addr: String,
sdp: String,
#[serde(default)]
multicast_group: Option<String>,
},
TsUdp {
addr: String,
#[serde(default)]
multicast_group: Option<String>,
},
TsHttp {
url: String,
#[serde(default)]
auth: Option<AuthSpec>,
},
HlsPull {
url: String,
#[serde(default)]
auth: Option<AuthSpec>,
},
DashPull {
url: String,
#[serde(default)]
auth: Option<AuthSpec>,
},
SmoothPull {
url: String,
#[serde(default)]
auth: Option<AuthSpec>,
},
Rtmp {
listen: String,
#[serde(default)]
app: Option<String>,
#[serde(default)]
stream_key: Option<String>,
},
Srt {
#[serde(default)]
listen: Option<String>,
#[serde(default)]
remote: Option<String>,
#[serde(default)]
stream_id: Option<String>,
#[serde(default)]
latency_ms: Option<u16>,
},
Custom {
type_tag: String,
#[serde(default)]
params: serde_json::Value,
},
}
#[non_exhaustive]
#[derive(Clone, Deserialize)]
#[serde(untagged)]
pub enum AuthSpec {
Password {
username: String,
password: String,
},
Bearer {
bearer_token: String,
},
}
impl std::fmt::Debug for AuthSpec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AuthSpec::Password { username, .. } => f
.debug_struct("Password")
.field("username", username)
.field("password", &"***")
.finish(),
AuthSpec::Bearer { .. } => f
.debug_struct("Bearer")
.field("bearer_token", &"***")
.finish(),
}
}
}
impl AuthSpec {
#[allow(dead_code)]
pub(crate) fn to_credentials(&self) -> Credentials {
match self {
AuthSpec::Password { username, password } => {
Credentials::new(username.clone(), password.clone())
}
AuthSpec::Bearer { bearer_token } => Credentials::bearer(bearer_token.clone()),
}
}
}
#[non_exhaustive]
#[derive(Clone, Deserialize)]
#[serde(tag = "scheme", rename_all = "snake_case")]
pub enum OutputAuthSpec {
Basic {
username: String,
password: String,
},
Digest {
username: String,
password: String,
},
Bearer {
token: String,
},
Forwarded {
#[serde(default = "default_forwarded_user_header")]
user_header: String,
#[serde(default = "default_forwarded_for_header")]
forwarded_for_header: Option<String>,
},
Custom {
type_tag: String,
#[serde(default)]
params: serde_json::Value,
},
}
fn default_forwarded_user_header() -> String {
"X-Forwarded-User".to_string()
}
fn default_forwarded_for_header() -> Option<String> {
Some("X-Forwarded-For".to_string())
}
impl std::fmt::Debug for OutputAuthSpec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
OutputAuthSpec::Basic { username, .. } => f
.debug_struct("Basic")
.field("username", username)
.field("password", &"***")
.finish(),
OutputAuthSpec::Digest { username, .. } => f
.debug_struct("Digest")
.field("username", username)
.field("password", &"***")
.finish(),
OutputAuthSpec::Bearer { .. } => {
f.debug_struct("Bearer").field("token", &"***").finish()
}
OutputAuthSpec::Forwarded {
user_header,
forwarded_for_header,
} => f
.debug_struct("Forwarded")
.field("user_header", user_header)
.field("forwarded_for_header", forwarded_for_header)
.finish(),
OutputAuthSpec::Custom { type_tag, .. } => f
.debug_struct("Custom")
.field("type_tag", type_tag)
.field("params", &"<params>")
.finish(),
}
}
}
impl OutputAuthSpec {
pub(crate) fn build_verifier(&self, realm: &str) -> broadcast_auth::Verifier {
match self {
OutputAuthSpec::Basic { username, password } => broadcast_auth::Verifier::new(
Credentials::Basic {
username: username.clone(),
password: password.clone(),
},
realm,
),
OutputAuthSpec::Digest { username, password } => broadcast_auth::Verifier::new(
Credentials::Digest {
username: username.clone(),
password: password.clone(),
},
realm,
),
OutputAuthSpec::Bearer { token } => {
broadcast_auth::Verifier::new(Credentials::bearer(token.clone()), realm)
}
OutputAuthSpec::Forwarded {
user_header,
forwarded_for_header,
} => broadcast_auth::Verifier::forwarded(
user_header.clone(),
forwarded_for_header.clone(),
),
OutputAuthSpec::Custom { .. } => unreachable!(
"OutputAuthSpec::Custom cannot build a Verifier without a SchemeRegistry — \
crate::origin::serve_with_registry resolves it via `registry.auth(type_tag)` \
before this method is ever called on a Custom variant"
),
}
}
fn validate(&self) -> Result<()> {
match self {
OutputAuthSpec::Basic { username, .. } | OutputAuthSpec::Digest { username, .. }
if username.is_empty() =>
{
Err(MultimuxError::ConfigInvalid {
field: "output_auth.username",
reason: "must not be empty".into(),
})
}
OutputAuthSpec::Bearer { token } if token.is_empty() => {
Err(MultimuxError::ConfigInvalid {
field: "output_auth.token",
reason: "must not be empty".into(),
})
}
OutputAuthSpec::Forwarded { user_header, .. } if user_header.is_empty() => {
Err(MultimuxError::ConfigInvalid {
field: "output_auth.user_header",
reason: "must not be empty".into(),
})
}
OutputAuthSpec::Forwarded {
forwarded_for_header: Some(header),
..
} if header.is_empty() => Err(MultimuxError::ConfigInvalid {
field: "output_auth.forwarded_for_header",
reason: "must not be empty (use null to disable)".into(),
}),
_ => Ok(()),
}
}
}
impl std::fmt::Debug for InputSpec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
InputSpec::Rtsp { url, auth } => f
.debug_struct("Rtsp")
.field("url", &crate::redact::redact_url(url))
.field("auth", auth)
.finish(),
InputSpec::Rtp {
addr,
sdp,
multicast_group,
} => f
.debug_struct("Rtp")
.field("addr", addr)
.field("sdp_len", &sdp.len())
.field("multicast_group", multicast_group)
.finish(),
InputSpec::TsUdp {
addr,
multicast_group,
} => f
.debug_struct("TsUdp")
.field("addr", addr)
.field("multicast_group", multicast_group)
.finish(),
InputSpec::TsHttp { url, auth } => f
.debug_struct("TsHttp")
.field("url", &crate::redact::redact_url(url))
.field("auth", auth)
.finish(),
InputSpec::HlsPull { url, auth } => f
.debug_struct("HlsPull")
.field("url", &crate::redact::redact_url(url))
.field("auth", auth)
.finish(),
InputSpec::DashPull { url, auth } => f
.debug_struct("DashPull")
.field("url", &crate::redact::redact_url(url))
.field("auth", auth)
.finish(),
InputSpec::SmoothPull { url, auth } => f
.debug_struct("SmoothPull")
.field("url", &crate::redact::redact_url(url))
.field("auth", auth)
.finish(),
InputSpec::Rtmp {
listen,
app,
stream_key,
} => f
.debug_struct("Rtmp")
.field("listen", listen)
.field("app", app)
.field("stream_key", &stream_key.as_ref().map(|_| "***"))
.finish(),
InputSpec::Srt {
listen,
remote,
stream_id,
latency_ms,
} => f
.debug_struct("Srt")
.field("listen", listen)
.field("remote", remote)
.field("stream_id", stream_id)
.field("latency_ms", latency_ms)
.finish(),
InputSpec::Custom { type_tag, .. } => f
.debug_struct("Custom")
.field("type_tag", type_tag)
.field("params", &"<params>")
.finish(),
}
}
}
impl InputSpec {
fn validate(&self) -> Result<()> {
match self {
InputSpec::Rtsp { url, auth } => {
validate_rtsp_url(url)?;
validate_auth(auth)
}
InputSpec::Rtp {
addr,
sdp,
multicast_group,
} => {
validate_udp_addr(addr)?;
validate_sdp(sdp)?;
if let Some(group) = multicast_group {
validate_multicast_group(group)?;
}
Ok(())
}
InputSpec::TsUdp {
addr,
multicast_group,
} => {
validate_udp_addr(addr)?;
if let Some(group) = multicast_group {
validate_multicast_group(group)?;
}
Ok(())
}
InputSpec::TsHttp { url, auth } => {
validate_http_url(url)?;
validate_auth(auth)
}
InputSpec::HlsPull { url, auth } => {
validate_http_url(url)?;
validate_auth(auth)
}
InputSpec::DashPull { url, auth } => {
validate_http_url(url)?;
validate_auth(auth)
}
InputSpec::SmoothPull { url, auth } => {
validate_http_url(url)?;
validate_auth(auth)
}
InputSpec::Rtmp { listen, .. } => validate_listen_addr(listen),
InputSpec::Srt { listen, remote, .. } => match (listen, remote) {
(Some(_), Some(_)) => Err(MultimuxError::ConfigInvalid {
field: "routes.input.listen",
reason: "exactly one of listen/remote must be set, got both".into(),
}),
(None, None) => Err(MultimuxError::ConfigInvalid {
field: "routes.input.listen",
reason: "exactly one of listen/remote must be set, got neither".into(),
}),
(Some(listen), None) => validate_listen_addr(listen),
(None, Some(remote)) => validate_host_port(remote),
},
InputSpec::Custom { .. } => Ok(()),
}
}
}
fn validate_rtsp_url(url: &str) -> Result<()> {
let parsed = url::Url::parse(url).map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.url",
reason: format!("bad rtsp(s) URL {url:?}: {e}"),
})?;
match parsed.scheme() {
"rtsp" | "rtsps" => Ok(()),
other => Err(MultimuxError::ConfigInvalid {
field: "routes.input.url",
reason: format!("scheme must be rtsp or rtsps, got {other:?}"),
}),
}
}
fn validate_http_url(url: &str) -> Result<()> {
let parsed = url::Url::parse(url).map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.url",
reason: format!("bad http(s) URL {url:?}: {e}"),
})?;
match parsed.scheme() {
"http" | "https" => Ok(()),
other => Err(MultimuxError::ConfigInvalid {
field: "routes.input.url",
reason: format!("scheme must be http or https, got {other:?}"),
}),
}
}
fn validate_auth(auth: &Option<AuthSpec>) -> Result<()> {
match auth {
None => Ok(()),
Some(AuthSpec::Password { username, .. }) if username.is_empty() => {
Err(MultimuxError::ConfigInvalid {
field: "routes.input.auth.username",
reason: "must not be empty".into(),
})
}
Some(AuthSpec::Bearer { bearer_token }) if bearer_token.is_empty() => {
Err(MultimuxError::ConfigInvalid {
field: "routes.input.auth.bearer_token",
reason: "must not be empty".into(),
})
}
Some(_) => Ok(()),
}
}
fn validate_playlist_name(name: &str) -> Result<()> {
if name.is_empty() {
return Err(MultimuxError::ConfigInvalid {
field: "playlist_name",
reason: "must not be empty".into(),
});
}
if !name.ends_with(".m3u8") {
return Err(MultimuxError::ConfigInvalid {
field: "playlist_name",
reason: format!("must end in .m3u8, got {name:?}"),
});
}
if name.contains('/') {
return Err(MultimuxError::ConfigInvalid {
field: "playlist_name",
reason: format!("must not contain a slash, got {name:?}"),
});
}
if name == "master.m3u8" {
return Err(MultimuxError::ConfigInvalid {
field: "playlist_name",
reason: "must not be \"master.m3u8\" (that name is the master playlist route)".into(),
});
}
Ok(())
}
fn validate_udp_addr(addr: &str) -> Result<()> {
addr.parse::<SocketAddr>()
.map(|_| ())
.map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.addr",
reason: format!("bad UDP address {addr:?}: {e}"),
})
}
fn validate_listen_addr(addr: &str) -> Result<()> {
addr.parse::<SocketAddr>()
.map(|_| ())
.map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.listen",
reason: format!("bad listen address {addr:?}: {e}"),
})
}
fn validate_host_port(addr: &str) -> Result<()> {
let (host, port) = addr
.rsplit_once(':')
.ok_or_else(|| MultimuxError::ConfigInvalid {
field: "routes.input.remote",
reason: format!("bad host:port {addr:?}: missing \":port\""),
})?;
if host.is_empty() {
return Err(MultimuxError::ConfigInvalid {
field: "routes.input.remote",
reason: format!("bad host:port {addr:?}: empty host"),
});
}
port.parse::<u16>()
.map(|_| ())
.map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.remote",
reason: format!("bad host:port {addr:?}: invalid port: {e}"),
})
}
fn validate_multicast_group(group: &str) -> Result<()> {
let ip: IpAddr = group.parse().map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.multicast_group",
reason: format!("bad multicast group {group:?}: {e}"),
})?;
if !ip.is_multicast() {
return Err(MultimuxError::ConfigInvalid {
field: "routes.input.multicast_group",
reason: format!("{group} is not a multicast address"),
});
}
Ok(())
}
fn validate_sdp(sdp: &str) -> Result<()> {
if sdp.is_empty() {
return Err(MultimuxError::ConfigInvalid {
field: "routes.input.sdp",
reason: "must not be empty".into(),
});
}
let Some(path) = sdp.strip_prefix('@') else {
return sdp_types::Session::parse(sdp.as_bytes())
.map(|_| ())
.map_err(|e| MultimuxError::ConfigInvalid {
field: "routes.input.sdp",
reason: format!("unparsable inline SDP: {e}"),
});
};
if path.is_empty() {
return Err(MultimuxError::ConfigInvalid {
field: "routes.input.sdp",
reason: "@ file reference must name a path".into(),
});
}
Ok(())
}
#[derive(Clone, Deserialize)]
pub struct Route {
pub name: String,
pub input: InputSpec,
#[serde(default = "default_outputs")]
pub outputs: Vec<OutputKind>,
}
impl std::fmt::Debug for Route {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Route")
.field("name", &self.name)
.field("input", &self.input)
.field("outputs", &self.outputs)
.finish()
}
}
#[derive(Debug, Clone, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct Config {
pub bind: String,
pub target_duration_secs: f64,
pub part_target_ms: u32,
pub window_segments: usize,
pub routes: Vec<Route>,
pub request_timeout_secs: f64,
pub max_concurrent_requests: usize,
pub max_request_body_bytes: usize,
pub ingest_connect_timeout_secs: f64,
pub ingest_read_timeout_secs: f64,
#[serde(default = "default_playlist_name")]
pub playlist_name: String,
#[serde(default)]
pub output_auth: Option<OutputAuthSpec>,
}
fn default_playlist_name() -> String {
crate::output::llhls::DEFAULT_PLAYLIST_NAME.to_string()
}
impl Default for Config {
fn default() -> Self {
Config {
bind: "0.0.0.0:8080".to_string(),
target_duration_secs: 4.0,
part_target_ms: 500,
window_segments: 8,
routes: Vec::new(),
request_timeout_secs: crate::origin::DEFAULT_REQUEST_TIMEOUT.as_secs_f64(),
max_concurrent_requests: crate::origin::DEFAULT_MAX_CONCURRENT_REQUESTS,
max_request_body_bytes: crate::origin::DEFAULT_MAX_REQUEST_BODY_BYTES,
ingest_connect_timeout_secs: crate::source::DEFAULT_CONNECT_TIMEOUT.as_secs_f64(),
ingest_read_timeout_secs: crate::source::DEFAULT_READ_TIMEOUT.as_secs_f64(),
playlist_name: default_playlist_name(),
output_auth: None,
}
}
}
const MIN_REQUEST_TIMEOUT_SECS: f64 = 5.0;
impl Config {
pub fn from_json_file(path: &Path) -> Result<Config> {
let bytes = std::fs::read(path).map_err(|source| MultimuxError::ConfigRead {
path: path.to_path_buf(),
source,
})?;
let cfg: Config =
serde_json::from_slice(&bytes).map_err(|e| MultimuxError::ConfigParse {
path: path.to_path_buf(),
reason: e.to_string(),
})?;
cfg.validate()?;
Ok(cfg)
}
pub fn validate(&self) -> Result<()> {
if self.routes.is_empty() {
return Err(MultimuxError::ConfigInvalid {
field: "routes",
reason: "no routes configured".into(),
});
}
if self.target_duration_secs <= 0.0 {
return Err(MultimuxError::ConfigInvalid {
field: "target_duration_secs",
reason: "must be positive".into(),
});
}
if self.part_target_ms == 0 {
return Err(MultimuxError::ConfigInvalid {
field: "part_target_ms",
reason: "must be positive".into(),
});
}
if self.window_segments == 0 {
return Err(MultimuxError::ConfigInvalid {
field: "window_segments",
reason: "must be positive".into(),
});
}
if self.request_timeout_secs <= MIN_REQUEST_TIMEOUT_SECS {
return Err(MultimuxError::ConfigInvalid {
field: "request_timeout_secs",
reason: format!(
"must exceed {MIN_REQUEST_TIMEOUT_SECS} (the LL-HLS blocking-reload cap), \
got {}",
self.request_timeout_secs
),
});
}
if self.max_concurrent_requests == 0 {
return Err(MultimuxError::ConfigInvalid {
field: "max_concurrent_requests",
reason: "must be positive".into(),
});
}
if self.max_request_body_bytes == 0 {
return Err(MultimuxError::ConfigInvalid {
field: "max_request_body_bytes",
reason: "must be positive".into(),
});
}
if self.ingest_connect_timeout_secs <= 0.0 {
return Err(MultimuxError::ConfigInvalid {
field: "ingest_connect_timeout_secs",
reason: "must be positive".into(),
});
}
if self.ingest_read_timeout_secs <= 0.0 {
return Err(MultimuxError::ConfigInvalid {
field: "ingest_read_timeout_secs",
reason: "must be positive".into(),
});
}
validate_playlist_name(&self.playlist_name)?;
if let Some(output_auth) = &self.output_auth {
output_auth.validate()?;
}
let mut seen = std::collections::HashSet::new();
for r in &self.routes {
if !seen.insert(r.name.as_str()) {
return Err(MultimuxError::ConfigInvalid {
field: "routes",
reason: format!("duplicate stream name {:?}", r.name),
});
}
if r.outputs.is_empty() {
return Err(MultimuxError::ConfigInvalid {
field: "routes.outputs",
reason: format!("route {:?} has no outputs configured", r.name),
});
}
r.input.validate()?;
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_json_config_with_rtsp_routes() {
let json = r#"{
"bind": "127.0.0.1:9000",
"target_duration_secs": 2.0,
"part_target_ms": 250,
"window_segments": 6,
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } },
{ "name": "cam2", "input": { "type": "rtsp", "url": "rtsp://host/stream2" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.bind, "127.0.0.1:9000");
assert_eq!(cfg.part_target_ms, 250);
assert_eq!(cfg.routes.len(), 2);
assert_eq!(cfg.routes[1].name, "cam2");
match &cfg.routes[1].input {
InputSpec::Rtsp { url, .. } => assert_eq!(url, "rtsp://host/stream2"),
other => panic!("expected InputSpec::Rtsp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn http_limits_default_when_omitted() {
let json = r#"{
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(
cfg.request_timeout_secs,
crate::origin::DEFAULT_REQUEST_TIMEOUT.as_secs_f64()
);
assert_eq!(
cfg.max_concurrent_requests,
crate::origin::DEFAULT_MAX_CONCURRENT_REQUESTS
);
assert_eq!(
cfg.max_request_body_bytes,
crate::origin::DEFAULT_MAX_REQUEST_BODY_BYTES
);
cfg.validate().unwrap();
}
#[test]
fn validate_rejects_request_timeout_at_or_below_blocking_cap() {
for bad in [1.0, 5.0] {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtsp {
url: "rtsp://a".into(),
auth: None,
},
outputs: default_outputs(),
}],
request_timeout_secs: bad,
..Config::default()
};
assert!(cfg.validate().is_err(), "{bad} must be rejected");
}
}
#[test]
fn validate_rejects_zero_max_concurrent_requests() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtsp {
url: "rtsp://a".into(),
auth: None,
},
outputs: default_outputs(),
}],
max_concurrent_requests: 0,
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_zero_max_request_body_bytes() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtsp {
url: "rtsp://a".into(),
auth: None,
},
outputs: default_outputs(),
}],
max_request_body_bytes: 0,
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn parses_json_config_with_http_limits() {
let json = r#"{
"request_timeout_secs": 15.0,
"max_concurrent_requests": 100,
"max_request_body_bytes": 2048,
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.request_timeout_secs, 15.0);
assert_eq!(cfg.max_concurrent_requests, 100);
assert_eq!(cfg.max_request_body_bytes, 2048);
cfg.validate().unwrap();
}
fn output_kind_names(kinds: &[OutputKind]) -> Vec<&str> {
kinds.iter().map(OutputKind::name).collect()
}
#[test]
fn route_outputs_defaults_to_llhls_only_when_omitted() {
let json = r#"{
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(output_kind_names(&cfg.routes[0].outputs), vec!["llhls"]);
cfg.validate().unwrap();
}
#[test]
fn route_outputs_parses_llhls_and_dash() {
let json = r#"{
"routes": [
{
"name": "cam1",
"input": { "type": "rtsp", "url": "rtsp://host/stream1" },
"outputs": ["llhls", "dash"]
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(
output_kind_names(&cfg.routes[0].outputs),
vec!["llhls", "dash"]
);
cfg.validate().unwrap();
}
#[test]
fn route_outputs_dash_only_is_valid() {
let json = r#"{
"routes": [
{
"name": "cam1",
"input": { "type": "rtsp", "url": "rtsp://host/stream1" },
"outputs": ["dash"]
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(output_kind_names(&cfg.routes[0].outputs), vec!["dash"]);
cfg.validate().unwrap();
}
#[test]
fn route_outputs_parses_ll_dash() {
let json = r#"{
"routes": [
{
"name": "cam1",
"input": { "type": "rtsp", "url": "rtsp://host/stream1" },
"outputs": ["llhls", "dash", "ll_dash"]
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(
output_kind_names(&cfg.routes[0].outputs),
vec!["llhls", "dash", "ll_dash"]
);
cfg.validate().unwrap();
}
#[test]
fn validate_rejects_empty_outputs_list() {
let json = r#"{
"routes": [
{
"name": "cam1",
"input": { "type": "rtsp", "url": "rtsp://host/stream1" },
"outputs": []
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert!(cfg.validate().is_err());
}
#[test]
fn rejects_unknown_output_kind() {
let json = r#"{
"routes": [
{
"name": "cam1",
"input": { "type": "rtsp", "url": "rtsp://host/stream1" },
"outputs": ["lldash"]
}
]
}"#;
let result: std::result::Result<Config, _> = serde_json::from_str(json);
assert!(result.is_err(), "unknown output kind must be rejected");
}
#[test]
fn parses_json_config_with_rtp_input() {
let sdp = "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\n\
m=video 0 RTP/AVP 96\r\na=rtpmap:96 H264/90000\r\n\
a=fmtp:96 packetization-mode=1;sprop-parameter-sets=Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==\r\n";
let json = serde_json::json!({
"bind": "127.0.0.1:9000",
"target_duration_secs": 2.0,
"part_target_ms": 250,
"window_segments": 6,
"routes": [
{
"name": "cam-rtp",
"input": {
"type": "rtp",
"addr": "0.0.0.0:5004",
"sdp": sdp,
"multicast_group": "239.1.1.1"
}
}
]
});
let cfg: Config = serde_json::from_value(json).unwrap();
assert_eq!(cfg.routes.len(), 1);
match &cfg.routes[0].input {
InputSpec::Rtp {
addr,
sdp: parsed_sdp,
multicast_group,
} => {
assert_eq!(addr, "0.0.0.0:5004");
assert_eq!(parsed_sdp, sdp);
assert_eq!(multicast_group.as_deref(), Some("239.1.1.1"));
}
other => panic!("expected InputSpec::Rtp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_ts_udp_input() {
let json = r#"{
"routes": [
{
"name": "cam-ts",
"input": { "type": "ts_udp", "addr": "0.0.0.0:5005" }
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.routes.len(), 1);
match &cfg.routes[0].input {
InputSpec::TsUdp {
addr,
multicast_group,
} => {
assert_eq!(addr, "0.0.0.0:5005");
assert_eq!(*multicast_group, None);
}
other => panic!("expected InputSpec::TsUdp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_ts_udp_multicast_group() {
let json = r#"{
"routes": [
{
"name": "cam-ts-mc",
"input": {
"type": "ts_udp",
"addr": "0.0.0.0:5006",
"multicast_group": "239.2.2.2"
}
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::TsUdp {
multicast_group, ..
} => assert_eq!(multicast_group.as_deref(), Some("239.2.2.2")),
other => panic!("expected InputSpec::TsUdp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_ts_http_input() {
let json = r#"{
"routes": [
{
"name": "cam-ts-http",
"input": { "type": "ts_http", "url": "http://host/stream.ts" }
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.routes.len(), 1);
match &cfg.routes[0].input {
InputSpec::TsHttp { url, .. } => assert_eq!(url, "http://host/stream.ts"),
other => panic!("expected InputSpec::TsHttp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_hls_pull_input() {
let json = r#"{
"routes": [
{
"name": "cam-hls-pull",
"input": { "type": "hls_pull", "url": "https://origin/live/media.m3u8" }
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.routes.len(), 1);
match &cfg.routes[0].input {
InputSpec::HlsPull { url, .. } => assert_eq!(url, "https://origin/live/media.m3u8"),
other => panic!("expected InputSpec::HlsPull, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_smooth_pull_input() {
let json = r#"{
"routes": [
{
"name": "cam-smooth-pull",
"input": { "type": "smooth_pull", "url": "https://origin/live.ism/Manifest" }
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.routes.len(), 1);
match &cfg.routes[0].input {
InputSpec::SmoothPull { url, .. } => {
assert_eq!(url, "https://origin/live.ism/Manifest")
}
other => panic!("expected InputSpec::SmoothPull, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_password_auth() {
let json = r#"{
"routes": [
{
"name": "cam-ts-http",
"input": {
"type": "ts_http",
"url": "http://host/stream.ts",
"auth": { "username": "admin", "password": "hunter2" }
}
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::TsHttp { auth, .. } => match auth {
Some(AuthSpec::Password { username, password }) => {
assert_eq!(username, "admin");
assert_eq!(password, "hunter2");
}
other => panic!("expected Some(AuthSpec::Password), got {other:?}"),
},
other => panic!("expected InputSpec::TsHttp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_bearer_auth() {
let json = r#"{
"routes": [
{
"name": "cam-hls-pull",
"input": {
"type": "hls_pull",
"url": "https://origin/live/media.m3u8",
"auth": { "bearer_token": "tok123" }
}
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::HlsPull { auth, .. } => match auth {
Some(AuthSpec::Bearer { bearer_token }) => assert_eq!(bearer_token, "tok123"),
other => panic!("expected Some(AuthSpec::Bearer), got {other:?}"),
},
other => panic!("expected InputSpec::HlsPull, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn parses_json_config_with_rtsp_password_auth() {
let json = r#"{
"routes": [
{
"name": "cam1",
"input": {
"type": "rtsp",
"url": "rtsp://host/stream",
"auth": { "username": "admin", "password": "hunter2" }
}
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::Rtsp { auth, .. } => {
assert!(matches!(auth, Some(AuthSpec::Password { .. })));
}
other => panic!("expected InputSpec::Rtsp, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn auth_defaults_to_none_when_omitted() {
let json = r#"{
"routes": [
{
"name": "cam-ts-http",
"input": { "type": "ts_http", "url": "http://host/stream.ts" }
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::TsHttp { auth, .. } => assert!(auth.is_none()),
other => panic!("expected InputSpec::TsHttp, got {other:?}"),
}
}
#[test]
fn validate_rejects_empty_auth_username() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::TsHttp {
url: "http://host/stream.ts".into(),
auth: Some(AuthSpec::Password {
username: String::new(),
password: "p".into(),
}),
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_empty_bearer_token() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::HlsPull {
url: "https://host/media.m3u8".into(),
auth: Some(AuthSpec::Bearer {
bearer_token: String::new(),
}),
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_accepts_empty_password() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::TsHttp {
url: "http://host/stream.ts".into(),
auth: Some(AuthSpec::Password {
username: "admin".into(),
password: String::new(),
}),
},
outputs: default_outputs(),
}],
..Config::default()
};
cfg.validate().unwrap();
}
#[test]
fn input_spec_debug_redacts_config_supplied_auth() {
let password_auth = InputSpec::TsHttp {
url: "http://host/stream.ts".into(),
auth: Some(AuthSpec::Password {
username: "admin".into(),
password: "hunter2secret".into(),
}),
};
let debug = format!("{password_auth:?}");
assert!(debug.contains("admin"), "username may render: {debug}");
assert!(
!debug.contains("hunter2secret"),
"debug leaked password: {debug}"
);
let bearer_auth = InputSpec::HlsPull {
url: "https://host/media.m3u8".into(),
auth: Some(AuthSpec::Bearer {
bearer_token: "supersecrettoken".into(),
}),
};
let debug = format!("{bearer_auth:?}");
assert!(
!debug.contains("supersecrettoken"),
"debug leaked bearer token: {debug}"
);
}
#[test]
fn auth_spec_to_credentials_converts_both_variants() {
let password = AuthSpec::Password {
username: "admin".into(),
password: "hunter2".into(),
};
assert_eq!(
password.to_credentials(),
Credentials::new("admin", "hunter2")
);
let bearer = AuthSpec::Bearer {
bearer_token: "tok".into(),
};
assert_eq!(bearer.to_credentials(), Credentials::bearer("tok"));
}
#[test]
fn validate_rejects_bad_ts_http_scheme() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::TsHttp {
url: "rtsp://host/stream.ts".into(),
auth: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_bad_hls_pull_scheme() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::HlsPull {
url: "ftp://host/media.m3u8".into(),
auth: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_unparsable_ts_http_url() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::TsHttp {
url: "not a url".into(),
auth: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn route_debug_redacts_ts_http_and_hls_pull_credentials() {
let ts_http = Route {
name: "cam-ts-http".into(),
input: InputSpec::TsHttp {
url: "http://user:secretpass@host/stream.ts".into(),
auth: None,
},
outputs: default_outputs(),
};
let debug = format!("{ts_http:?}");
assert!(!debug.contains("user"), "debug leaked username: {debug}");
assert!(
!debug.contains("secretpass"),
"debug leaked password: {debug}"
);
assert!(debug.contains("***@host"), "debug: {debug}");
let hls_pull = Route {
name: "cam-hls-pull".into(),
input: InputSpec::HlsPull {
url: "https://user:secretpass@origin/media.m3u8".into(),
auth: None,
},
outputs: default_outputs(),
};
let debug = format!("{hls_pull:?}");
assert!(!debug.contains("user"), "debug leaked username: {debug}");
assert!(
!debug.contains("secretpass"),
"debug leaked password: {debug}"
);
assert!(debug.contains("***@origin"), "debug: {debug}");
}
#[test]
fn validate_rejects_bad_rtsp_scheme() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtsp {
url: "http://host/stream".into(),
auth: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_unparsable_udp_addr() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::TsUdp {
addr: "not-an-addr".into(),
multicast_group: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_non_multicast_group() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::TsUdp {
addr: "0.0.0.0:5005".into(),
multicast_group: Some("10.0.0.1".into()),
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_empty_rtp_sdp() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtp {
addr: "0.0.0.0:5004".into(),
sdp: String::new(),
multicast_group: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_unparsable_inline_rtp_sdp() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtp {
addr: "0.0.0.0:5004".into(),
sdp: "not an sdp body".into(),
multicast_group: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_accepts_at_path_rtp_sdp_reference_without_reading_it() {
let cfg = Config {
routes: vec![Route {
name: "x".into(),
input: InputSpec::Rtp {
addr: "0.0.0.0:5004".into(),
sdp: "@/no/such/file/does-not-exist.sdp".into(),
multicast_group: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
cfg.validate().unwrap();
}
#[test]
fn validate_rejects_duplicate_stream_names() {
let cfg = Config {
routes: vec![
Route {
name: "x".into(),
input: InputSpec::Rtsp {
url: "rtsp://a".into(),
auth: None,
},
outputs: default_outputs(),
},
Route {
name: "x".into(),
input: InputSpec::Rtsp {
url: "rtsp://b".into(),
auth: None,
},
outputs: default_outputs(),
},
],
..Config::default()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_no_routes() {
assert!(Config::default().validate().is_err());
}
#[test]
fn rejects_unknown_config_key() {
let json = r#"{
"bind": "127.0.0.1:9000",
"window_segment": 6,
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let result: std::result::Result<Config, _> = serde_json::from_str(json);
assert!(
result.is_err(),
"unknown key must be rejected, not silently ignored"
);
}
#[test]
fn rejects_unknown_input_type() {
let json = r#"{
"routes": [
{ "name": "cam1", "input": { "type": "rtmp", "url": "rtmp://host/stream1" } }
]
}"#;
let result: std::result::Result<Config, _> = serde_json::from_str(json);
assert!(result.is_err(), "unknown input type must be rejected");
}
#[test]
fn route_debug_redacts_rtsp_credentials() {
let route = Route {
name: "cam1".into(),
input: InputSpec::Rtsp {
url: "rtsp://user:secretpass@host/s".into(),
auth: None,
},
outputs: default_outputs(),
};
let debug = format!("{route:?}");
assert!(!debug.contains("user"), "debug leaked username: {debug}");
assert!(
!debug.contains("secretpass"),
"debug leaked password: {debug}"
);
assert!(debug.contains("***@host"), "debug: {debug}");
}
#[test]
fn config_debug_redacts_route_credentials() {
let cfg = Config {
routes: vec![Route {
name: "cam1".into(),
input: InputSpec::Rtsp {
url: "rtsp://user:secretpass@host/s".into(),
auth: None,
},
outputs: default_outputs(),
}],
..Config::default()
};
let debug = format!("{cfg:?}");
assert!(!debug.contains("user"), "config debug leaked username");
assert!(
!debug.contains("secretpass"),
"config debug leaked password"
);
assert!(debug.contains("***@host"));
}
#[test]
fn route_debug_shows_sdp_length_not_full_body() {
let long_sdp = "v=0\r\n".repeat(50);
let route = Route {
name: "cam-rtp".into(),
input: InputSpec::Rtp {
addr: "0.0.0.0:5004".into(),
sdp: long_sdp.clone(),
multicast_group: None,
},
outputs: default_outputs(),
};
let debug = format!("{route:?}");
assert!(!debug.contains(&long_sdp), "debug: {debug}");
assert!(
debug.contains(&long_sdp.len().to_string()),
"debug: {debug}"
);
}
fn cfg_with_one_route() -> Config {
Config {
routes: vec![Route {
name: "cam1".into(),
input: InputSpec::Rtsp {
url: "rtsp://host/stream".into(),
auth: None,
},
outputs: default_outputs(),
}],
..Config::default()
}
}
#[test]
fn playlist_name_defaults_to_media_m3u8_when_omitted() {
let json = r#"{
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.playlist_name, "media.m3u8");
cfg.validate().unwrap();
}
#[test]
fn playlist_name_parses_from_json() {
let json = r#"{
"playlist_name": "index.m3u8",
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert_eq!(cfg.playlist_name, "index.m3u8");
cfg.validate().unwrap();
}
#[test]
fn validate_rejects_empty_playlist_name() {
let cfg = Config {
playlist_name: String::new(),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_playlist_name_without_m3u8_suffix() {
let cfg = Config {
playlist_name: "media.mpd".into(),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_playlist_name_with_slash() {
let cfg = Config {
playlist_name: "sub/media.m3u8".into(),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_playlist_name_master_m3u8_collision() {
let cfg = Config {
playlist_name: "master.m3u8".into(),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_accepts_a_valid_non_default_playlist_name() {
let cfg = Config {
playlist_name: "index.m3u8".into(),
..cfg_with_one_route()
};
cfg.validate().unwrap();
}
#[test]
fn output_auth_defaults_to_none_when_omitted() {
let json = r#"{
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert!(cfg.output_auth.is_none());
cfg.validate().unwrap();
}
#[test]
fn output_auth_parses_basic() {
let json = r#"{
"output_auth": { "scheme": "basic", "username": "admin", "password": "hunter2" },
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.output_auth {
Some(OutputAuthSpec::Basic { username, password }) => {
assert_eq!(username, "admin");
assert_eq!(password, "hunter2");
}
other => panic!("expected Some(OutputAuthSpec::Basic), got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn output_auth_parses_digest() {
let json = r#"{
"output_auth": { "scheme": "digest", "username": "admin", "password": "hunter2" },
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
assert!(matches!(
&cfg.output_auth,
Some(OutputAuthSpec::Digest { .. })
));
cfg.validate().unwrap();
}
#[test]
fn output_auth_parses_bearer() {
let json = r#"{
"output_auth": { "scheme": "bearer", "token": "tok123" },
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.output_auth {
Some(OutputAuthSpec::Bearer { token }) => assert_eq!(token, "tok123"),
other => panic!("expected Some(OutputAuthSpec::Bearer), got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn output_auth_parses_forwarded() {
let json = r#"{
"output_auth": {
"scheme": "forwarded",
"user_header": "X-Auth-User",
"forwarded_for_header": "X-Real-IP"
},
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.output_auth {
Some(OutputAuthSpec::Forwarded {
user_header,
forwarded_for_header,
}) => {
assert_eq!(user_header, "X-Auth-User");
assert_eq!(forwarded_for_header.as_deref(), Some("X-Real-IP"));
}
other => panic!("expected Some(OutputAuthSpec::Forwarded), got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn output_auth_forwarded_defaults_headers_when_omitted() {
let json = r#"{
"output_auth": { "scheme": "forwarded" },
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.output_auth {
Some(OutputAuthSpec::Forwarded {
user_header,
forwarded_for_header,
}) => {
assert_eq!(user_header, "X-Forwarded-User");
assert_eq!(forwarded_for_header.as_deref(), Some("X-Forwarded-For"));
}
other => panic!("expected Some(OutputAuthSpec::Forwarded), got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn output_auth_forwarded_for_header_can_be_disabled() {
let json = r#"{
"output_auth": {
"scheme": "forwarded",
"forwarded_for_header": null
},
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.output_auth {
Some(OutputAuthSpec::Forwarded {
forwarded_for_header,
..
}) => assert_eq!(*forwarded_for_header, None),
other => panic!("expected Some(OutputAuthSpec::Forwarded), got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn validate_rejects_output_auth_forwarded_empty_user_header() {
let cfg = Config {
output_auth: Some(OutputAuthSpec::Forwarded {
user_header: String::new(),
forwarded_for_header: None,
}),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_output_auth_forwarded_empty_forwarded_for_header() {
let cfg = Config {
output_auth: Some(OutputAuthSpec::Forwarded {
user_header: "X-Forwarded-User".into(),
forwarded_for_header: Some(String::new()),
}),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn output_auth_rejects_unknown_scheme() {
let json = r#"{
"output_auth": { "scheme": "hmac", "username": "admin", "password": "p" },
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let result: std::result::Result<Config, _> = serde_json::from_str(json);
assert!(
result.is_err(),
"unknown output_auth scheme must be rejected"
);
}
#[test]
fn validate_rejects_output_auth_empty_username() {
let cfg = Config {
output_auth: Some(OutputAuthSpec::Basic {
username: String::new(),
password: "p".into(),
}),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_rejects_output_auth_empty_bearer_token() {
let cfg = Config {
output_auth: Some(OutputAuthSpec::Bearer {
token: String::new(),
}),
..cfg_with_one_route()
};
assert!(cfg.validate().is_err());
}
#[test]
fn validate_accepts_output_auth_empty_password() {
let cfg = Config {
output_auth: Some(OutputAuthSpec::Basic {
username: "admin".into(),
password: String::new(),
}),
..cfg_with_one_route()
};
cfg.validate().unwrap();
}
#[test]
fn output_auth_spec_build_verifier_preserves_scheme_exactly() {
let basic = OutputAuthSpec::Basic {
username: "admin".into(),
password: "p".into(),
};
assert!(
basic
.build_verifier("realm")
.challenge()
.starts_with("Basic ")
);
let digest = OutputAuthSpec::Digest {
username: "admin".into(),
password: "p".into(),
};
assert!(
digest
.build_verifier("realm")
.challenge()
.starts_with("Digest ")
);
let bearer = OutputAuthSpec::Bearer {
token: "tok".into(),
};
assert_eq!(bearer.build_verifier("realm").challenge(), "Bearer");
let forwarded = OutputAuthSpec::Forwarded {
user_header: "X-Forwarded-User".into(),
forwarded_for_header: Some("X-Forwarded-For".into()),
};
assert_eq!(forwarded.build_verifier("realm").challenge(), "Forwarded");
}
#[test]
fn output_auth_spec_debug_redacts_secret() {
let basic = OutputAuthSpec::Basic {
username: "admin".into(),
password: "supersecretpass".into(),
};
let debug = format!("{basic:?}");
assert!(debug.contains("admin"), "username may render: {debug}");
assert!(!debug.contains("supersecretpass"), "debug: {debug}");
let bearer = OutputAuthSpec::Bearer {
token: "supersecrettoken".into(),
};
let debug = format!("{bearer:?}");
assert!(!debug.contains("supersecrettoken"), "debug: {debug}");
}
#[test]
fn output_auth_spec_forwarded_debug_shows_header_names() {
let forwarded = OutputAuthSpec::Forwarded {
user_header: "X-Forwarded-User".into(),
forwarded_for_header: Some("X-Forwarded-For".into()),
};
let debug = format!("{forwarded:?}");
assert!(debug.contains("X-Forwarded-User"), "debug: {debug}");
assert!(debug.contains("X-Forwarded-For"), "debug: {debug}");
}
#[test]
fn input_spec_custom_deserializes_with_type_tag_and_params() {
let json = r#"{
"routes": [
{
"name": "cam-custom",
"input": {
"type": "custom",
"type_tag": "webrtc",
"params": { "offer_url": "https://example/offer" }
}
}
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::Custom { type_tag, params } => {
assert_eq!(type_tag, "webrtc");
assert_eq!(
params.get("offer_url").and_then(|v| v.as_str()),
Some("https://example/offer")
);
}
other => panic!("expected InputSpec::Custom, got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn input_spec_custom_params_defaults_to_null_when_omitted() {
let json = r#"{
"routes": [
{ "name": "cam-custom", "input": { "type": "custom", "type_tag": "webrtc" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.routes[0].input {
InputSpec::Custom { params, .. } => assert!(params.is_null()),
other => panic!("expected InputSpec::Custom, got {other:?}"),
}
}
#[test]
fn input_spec_custom_debug_redacts_params() {
let spec = InputSpec::Custom {
type_tag: "webrtc".into(),
params: serde_json::json!({ "password": "s3cret" }),
};
let debug = format!("{spec:?}");
assert!(debug.contains("webrtc"), "type_tag may render: {debug}");
assert!(!debug.contains("s3cret"), "debug leaked params: {debug}");
}
#[test]
fn output_auth_spec_custom_deserializes_with_type_tag_and_params() {
let json = r#"{
"output_auth": {
"scheme": "custom",
"type_tag": "hmac",
"params": { "key_id": "abc" }
},
"routes": [
{ "name": "cam1", "input": { "type": "rtsp", "url": "rtsp://host/stream1" } }
]
}"#;
let cfg: Config = serde_json::from_str(json).unwrap();
match &cfg.output_auth {
Some(OutputAuthSpec::Custom { type_tag, params }) => {
assert_eq!(type_tag, "hmac");
assert_eq!(params.get("key_id").and_then(|v| v.as_str()), Some("abc"));
}
other => panic!("expected Some(OutputAuthSpec::Custom), got {other:?}"),
}
cfg.validate().unwrap();
}
#[test]
fn output_auth_spec_custom_debug_redacts_params() {
let spec = OutputAuthSpec::Custom {
type_tag: "hmac".into(),
params: serde_json::json!({ "shared_secret": "topsecret" }),
};
let debug = format!("{spec:?}");
assert!(debug.contains("hmac"), "type_tag may render: {debug}");
assert!(!debug.contains("topsecret"), "debug leaked params: {debug}");
}
#[test]
#[should_panic(expected = "SchemeRegistry")]
fn output_auth_spec_custom_build_verifier_is_unreachable() {
let spec = OutputAuthSpec::Custom {
type_tag: "hmac".into(),
params: serde_json::Value::Null,
};
let _ = spec.build_verifier("realm");
}
fn srt_input(listen: Option<&str>, remote: Option<&str>) -> InputSpec {
InputSpec::Srt {
listen: listen.map(str::to_string),
remote: remote.map(str::to_string),
stream_id: None,
latency_ms: None,
}
}
#[test]
fn srt_caller_remote_accepts_a_hostname() {
let input = srt_input(None, Some("example.com:9000"));
input.validate().expect("hostname remote must validate");
}
#[test]
fn srt_caller_remote_accepts_a_literal_socket_addr() {
let input = srt_input(None, Some("127.0.0.1:9000"));
input
.validate()
.expect("literal socket addr remote must validate");
}
#[test]
fn srt_caller_remote_without_port_fails_with_remote_field() {
let input = srt_input(None, Some("nonsense"));
let err = input.validate().expect_err("remote with no port must fail");
match err {
MultimuxError::ConfigInvalid { field, .. } => {
assert_eq!(
field, "routes.input.remote",
"field must name remote, got {err:?}"
);
}
other => panic!("expected ConfigInvalid, got {other:?}"),
}
}
#[test]
fn srt_caller_remote_with_empty_host_fails_with_remote_field() {
let input = srt_input(None, Some(":9000"));
let err = input
.validate()
.expect_err("remote with empty host must fail");
match err {
MultimuxError::ConfigInvalid { field, .. } => {
assert_eq!(field, "routes.input.remote");
}
other => panic!("expected ConfigInvalid, got {other:?}"),
}
}
#[test]
fn srt_caller_remote_with_non_numeric_port_fails_with_remote_field() {
let input = srt_input(None, Some("example.com:notaport"));
let err = input
.validate()
.expect_err("remote with a non-numeric port must fail");
match err {
MultimuxError::ConfigInvalid { field, .. } => {
assert_eq!(field, "routes.input.remote");
}
other => panic!("expected ConfigInvalid, got {other:?}"),
}
}
#[test]
fn srt_listener_listen_rejects_a_hostname() {
let input = srt_input(Some("example.com:9000"), None);
let err = input
.validate()
.expect_err("hostname listen address must fail");
match err {
MultimuxError::ConfigInvalid { field, .. } => {
assert_eq!(field, "routes.input.listen");
}
other => panic!("expected ConfigInvalid, got {other:?}"),
}
}
}