use broadcast_common::{Parse, Serialize};
use crate::RtmpError;
type Result<T> = core::result::Result<T, RtmpError>;
pub mod marker {
pub const NUMBER: u8 = 0x00;
pub const BOOLEAN: u8 = 0x01;
pub const STRING: u8 = 0x02;
pub const OBJECT: u8 = 0x03;
pub const NULL: u8 = 0x05;
pub const UNDEFINED: u8 = 0x06;
pub const ECMA_ARRAY: u8 = 0x08;
pub const OBJECT_END: u8 = 0x09;
pub const STRICT_ARRAY: u8 = 0x0A;
pub const DATE: u8 = 0x0B;
pub const LONG_STRING: u8 = 0x0C;
}
pub const MAX_AMF0_DEPTH: usize = 32;
const MARKER_LEN: usize = 1;
const NUMBER_LEN: usize = 8;
const BOOLEAN_LEN: usize = 1;
const U16_LEN: usize = 2;
const U32_LEN: usize = 4;
const DATE_RESERVED_LEN: usize = 2;
const OBJECT_END_LEN: usize = 3;
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum Amf0Value {
Number(f64),
Boolean(bool),
String(String),
Object(Vec<(String, Amf0Value)>),
Null,
Undefined,
EcmaArray(Vec<(String, Amf0Value)>),
StrictArray(Vec<Amf0Value>),
Date(f64),
LongString(String),
}
fn buffer_too_short(need: usize, have: usize, what: &'static str) -> RtmpError {
RtmpError::BufferTooShort { need, have, what }
}
fn read_utf8_short(bytes: &[u8], what: &'static str) -> Result<(String, usize)> {
if bytes.len() < U16_LEN {
return Err(buffer_too_short(U16_LEN, bytes.len(), what));
}
let len = u16::from_be_bytes([bytes[0], bytes[1]]) as usize;
let total = U16_LEN
.checked_add(len)
.ok_or(RtmpError::Malformed { what })?;
if bytes.len() < total {
return Err(buffer_too_short(total, bytes.len(), what));
}
let s = String::from_utf8(bytes[U16_LEN..total].to_vec())
.map_err(|_| RtmpError::Malformed { what })?;
Ok((s, total))
}
fn read_utf8_long(bytes: &[u8], what: &'static str) -> Result<(String, usize)> {
if bytes.len() < U32_LEN {
return Err(buffer_too_short(U32_LEN, bytes.len(), what));
}
let len = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize;
let total = U32_LEN
.checked_add(len)
.ok_or(RtmpError::Malformed { what })?;
if bytes.len() < total {
return Err(buffer_too_short(total, bytes.len(), what));
}
let s = String::from_utf8(bytes[U32_LEN..total].to_vec())
.map_err(|_| RtmpError::Malformed { what })?;
Ok((s, total))
}
fn parse_pairs(bytes: &[u8], depth: usize) -> Result<(Vec<(String, Amf0Value)>, usize)> {
let mut consumed = 0;
let mut pairs = Vec::new();
loop {
let (key, key_len) = read_utf8_short(&bytes[consumed..], "amf0 object key")?;
let after_key = consumed + key_len;
if key.is_empty() {
if bytes.len() < after_key + MARKER_LEN {
return Err(buffer_too_short(
after_key + MARKER_LEN,
bytes.len(),
"amf0 object-end marker",
));
}
if bytes[after_key] == marker::OBJECT_END {
return Ok((pairs, after_key + MARKER_LEN));
}
}
let value = parse_value(&bytes[after_key..], depth)?;
let value_len = value.serialized_len();
pairs.push((key, value));
consumed = after_key + value_len;
}
}
fn parse_value(bytes: &[u8], depth: usize) -> Result<Amf0Value> {
if bytes.is_empty() {
return Err(buffer_too_short(MARKER_LEN, 0, "amf0 value marker"));
}
let body = &bytes[MARKER_LEN..];
match bytes[0] {
marker::NUMBER => {
if body.len() < NUMBER_LEN {
return Err(buffer_too_short(NUMBER_LEN, body.len(), "amf0 number"));
}
let mut b = [0u8; NUMBER_LEN];
b.copy_from_slice(&body[..NUMBER_LEN]);
Ok(Amf0Value::Number(f64::from_be_bytes(b)))
}
marker::BOOLEAN => {
if body.is_empty() {
return Err(buffer_too_short(BOOLEAN_LEN, 0, "amf0 boolean"));
}
Ok(Amf0Value::Boolean(body[0] != 0))
}
marker::STRING => {
let (s, _) = read_utf8_short(body, "amf0 string")?;
Ok(Amf0Value::String(s))
}
marker::OBJECT => {
if depth >= MAX_AMF0_DEPTH {
return Err(RtmpError::Unsupported {
what: "amf0 nesting depth exceeded",
});
}
let (pairs, _) = parse_pairs(body, depth + 1)?;
Ok(Amf0Value::Object(pairs))
}
marker::NULL => Ok(Amf0Value::Null),
marker::UNDEFINED => Ok(Amf0Value::Undefined),
marker::ECMA_ARRAY => {
if depth >= MAX_AMF0_DEPTH {
return Err(RtmpError::Unsupported {
what: "amf0 nesting depth exceeded",
});
}
if body.len() < U32_LEN {
return Err(buffer_too_short(
U32_LEN,
body.len(),
"amf0 ecma array count",
));
}
let (pairs, _) = parse_pairs(&body[U32_LEN..], depth + 1)?;
Ok(Amf0Value::EcmaArray(pairs))
}
marker::STRICT_ARRAY => {
if depth >= MAX_AMF0_DEPTH {
return Err(RtmpError::Unsupported {
what: "amf0 nesting depth exceeded",
});
}
if body.len() < U32_LEN {
return Err(buffer_too_short(
U32_LEN,
body.len(),
"amf0 strict array count",
));
}
let count = u32::from_be_bytes([body[0], body[1], body[2], body[3]]);
let mut rest = &body[U32_LEN..];
let mut values = Vec::new();
for _ in 0..count {
let value = parse_value(rest, depth + 1)?;
let consumed = value.serialized_len();
values.push(value);
rest = &rest[consumed..];
}
Ok(Amf0Value::StrictArray(values))
}
marker::DATE => {
if body.len() < NUMBER_LEN + DATE_RESERVED_LEN {
return Err(buffer_too_short(
NUMBER_LEN + DATE_RESERVED_LEN,
body.len(),
"amf0 date",
));
}
let mut b = [0u8; NUMBER_LEN];
b.copy_from_slice(&body[..NUMBER_LEN]);
let tz = u16::from_be_bytes([body[NUMBER_LEN], body[NUMBER_LEN + 1]]);
if tz != 0 {
return Err(RtmpError::Malformed {
what: "amf0 date reserved time zone (must be 0x0000)",
});
}
Ok(Amf0Value::Date(f64::from_be_bytes(b)))
}
marker::LONG_STRING => {
let (s, _) = read_utf8_long(body, "amf0 long string")?;
Ok(Amf0Value::LongString(s))
}
_ => Err(RtmpError::Unsupported {
what: "amf0 value marker (reserved, legacy, or amf3-switch)",
}),
}
}
fn pairs_body_len(pairs: &[(String, Amf0Value)]) -> usize {
pairs
.iter()
.map(|(k, v)| U16_LEN + k.len() + v.serialized_len())
.sum::<usize>()
+ OBJECT_END_LEN
}
fn write_pairs(pairs: &[(String, Amf0Value)], buf: &mut [u8]) -> Result<usize> {
let mut offset = 0;
for (k, v) in pairs {
let key_total = U16_LEN + k.len();
if buf.len() < offset + key_total {
return Err(buffer_too_short(
offset + key_total,
buf.len(),
"amf0 object key output",
));
}
buf[offset..offset + U16_LEN].copy_from_slice(&(k.len() as u16).to_be_bytes());
buf[offset + U16_LEN..offset + key_total].copy_from_slice(k.as_bytes());
offset += key_total;
offset += v.serialize_into(&mut buf[offset..])?;
}
if buf.len() < offset + OBJECT_END_LEN {
return Err(buffer_too_short(
offset + OBJECT_END_LEN,
buf.len(),
"amf0 object-end output",
));
}
buf[offset] = 0;
buf[offset + 1] = 0;
buf[offset + 2] = marker::OBJECT_END;
Ok(offset + OBJECT_END_LEN)
}
impl<'a> Parse<'a> for Amf0Value {
type Error = RtmpError;
fn parse(bytes: &'a [u8]) -> Result<Self> {
parse_value(bytes, 0)
}
}
impl Serialize for Amf0Value {
type Error = RtmpError;
fn serialized_len(&self) -> usize {
MARKER_LEN
+ match self {
Amf0Value::Number(_) | Amf0Value::Date(_) => NUMBER_LEN,
Amf0Value::Boolean(_) => BOOLEAN_LEN,
Amf0Value::String(s) => U16_LEN + s.len(),
Amf0Value::LongString(s) => U32_LEN + s.len(),
Amf0Value::Object(pairs) => pairs_body_len(pairs),
Amf0Value::Null | Amf0Value::Undefined => 0,
Amf0Value::EcmaArray(pairs) => U32_LEN + pairs_body_len(pairs),
Amf0Value::StrictArray(values) => {
U32_LEN + values.iter().map(Serialize::serialized_len).sum::<usize>()
}
}
+ match self {
Amf0Value::Date(_) => DATE_RESERVED_LEN,
_ => 0,
}
}
fn serialize_into(&self, buf: &mut [u8]) -> Result<usize> {
let written = self.serialized_len();
if buf.len() < written {
return Err(buffer_too_short(written, buf.len(), "amf0 value output"));
}
let (marker_byte, body) = buf[..written].split_at_mut(MARKER_LEN);
match self {
Amf0Value::Number(v) => {
marker_byte[0] = marker::NUMBER;
body[..NUMBER_LEN].copy_from_slice(&v.to_be_bytes());
}
Amf0Value::Boolean(v) => {
marker_byte[0] = marker::BOOLEAN;
body[0] = u8::from(*v);
}
Amf0Value::String(s) => {
if s.len() > usize::from(u16::MAX) {
return Err(RtmpError::Unsupported {
what: "amf0 string exceeds u16 length (use long string)",
});
}
marker_byte[0] = marker::STRING;
body[..U16_LEN].copy_from_slice(&(s.len() as u16).to_be_bytes());
body[U16_LEN..].copy_from_slice(s.as_bytes());
}
Amf0Value::LongString(s) => {
marker_byte[0] = marker::LONG_STRING;
body[..U32_LEN].copy_from_slice(&(s.len() as u32).to_be_bytes());
body[U32_LEN..].copy_from_slice(s.as_bytes());
}
Amf0Value::Object(pairs) => {
marker_byte[0] = marker::OBJECT;
write_pairs(pairs, body)?;
}
Amf0Value::Null => marker_byte[0] = marker::NULL,
Amf0Value::Undefined => marker_byte[0] = marker::UNDEFINED,
Amf0Value::EcmaArray(pairs) => {
marker_byte[0] = marker::ECMA_ARRAY;
body[..U32_LEN].copy_from_slice(&(pairs.len() as u32).to_be_bytes());
write_pairs(pairs, &mut body[U32_LEN..])?;
}
Amf0Value::StrictArray(values) => {
marker_byte[0] = marker::STRICT_ARRAY;
body[..U32_LEN].copy_from_slice(&(values.len() as u32).to_be_bytes());
let mut offset = U32_LEN;
for v in values {
offset += v.serialize_into(&mut body[offset..])?;
}
}
Amf0Value::Date(v) => {
marker_byte[0] = marker::DATE;
body[..NUMBER_LEN].copy_from_slice(&v.to_be_bytes());
body[NUMBER_LEN..NUMBER_LEN + DATE_RESERVED_LEN].copy_from_slice(&[0, 0]);
}
}
Ok(written)
}
}
#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct Command {
pub name: String,
pub transaction_id: f64,
pub arguments: Vec<Amf0Value>,
}
impl Command {
pub fn parse(payload: &[u8]) -> Result<Self> {
let name_value = Amf0Value::parse(payload)?;
let mut offset = name_value.serialized_len();
let name = match name_value {
Amf0Value::String(s) => s,
_ => {
return Err(RtmpError::Malformed {
what: "rtmp command name (expected amf0 string)",
});
}
};
let txn_value = Amf0Value::parse(&payload[offset..])?;
offset += txn_value.serialized_len();
let transaction_id = match txn_value {
Amf0Value::Number(n) => n,
_ => {
return Err(RtmpError::Malformed {
what: "rtmp command transaction id (expected amf0 number)",
});
}
};
let mut arguments = Vec::new();
while offset < payload.len() {
let value = Amf0Value::parse(&payload[offset..])?;
offset += value.serialized_len();
arguments.push(value);
}
Ok(Command {
name,
transaction_id,
arguments,
})
}
#[must_use]
pub fn to_body(&self) -> Vec<u8> {
let mut out = Amf0Value::String(self.name.clone()).to_bytes();
out.extend(Amf0Value::Number(self.transaction_id).to_bytes());
for arg in &self.arguments {
out.extend(arg.to_bytes());
}
out
}
}
#[cfg(test)]
mod tests {
use super::*;
fn round_trip(v: &Amf0Value) {
let bytes = v.to_bytes();
assert_eq!(bytes.len(), v.serialized_len());
let parsed = Amf0Value::parse(&bytes).expect("parse");
assert_eq!(&parsed, v);
assert_eq!(parsed.to_bytes(), bytes);
}
#[test]
fn number_round_trips() {
round_trip(&Amf0Value::Number(0.0));
round_trip(&Amf0Value::Number(-1.5));
round_trip(&Amf0Value::Number(1_000_000.25));
}
#[test]
fn boolean_round_trips() {
round_trip(&Amf0Value::Boolean(true));
round_trip(&Amf0Value::Boolean(false));
}
#[test]
fn string_round_trips_including_empty_and_multibyte() {
round_trip(&Amf0Value::String(String::new()));
round_trip(&Amf0Value::String("live".to_string()));
round_trip(&Amf0Value::String("héllo wörld 日本語".to_string()));
}
#[test]
fn null_and_undefined_round_trip() {
round_trip(&Amf0Value::Null);
round_trip(&Amf0Value::Undefined);
}
#[test]
fn date_round_trips() {
round_trip(&Amf0Value::Date(1_700_000_000_000.0));
}
#[test]
fn long_string_round_trips() {
round_trip(&Amf0Value::LongString("x".repeat(70_000)));
}
#[test]
fn object_round_trips_including_nested_object() {
round_trip(&Amf0Value::Object(vec![]));
round_trip(&Amf0Value::Object(vec![
("app".to_string(), Amf0Value::String("live".to_string())),
("audioSampleRate".to_string(), Amf0Value::Number(44100.0)),
("live".to_string(), Amf0Value::Boolean(true)),
]));
round_trip(&Amf0Value::Object(vec![(
"capabilities".to_string(),
Amf0Value::Object(vec![("videoCodecs".to_string(), Amf0Value::Number(252.0))]),
)]));
}
#[test]
fn ecma_array_round_trips() {
round_trip(&Amf0Value::EcmaArray(vec![]));
round_trip(&Amf0Value::EcmaArray(vec![
("duration".to_string(), Amf0Value::Number(0.0)),
("width".to_string(), Amf0Value::Number(1920.0)),
]));
}
#[test]
fn strict_array_round_trips() {
round_trip(&Amf0Value::StrictArray(vec![]));
round_trip(&Amf0Value::StrictArray(vec![
Amf0Value::Number(1.0),
Amf0Value::String("two".to_string()),
Amf0Value::Boolean(false),
Amf0Value::Object(vec![("k".to_string(), Amf0Value::Null)]),
]));
}
#[test]
fn ecma_array_count_is_informational_not_cross_checked() {
let mut bytes = vec![marker::ECMA_ARRAY];
bytes.extend_from_slice(&999u32.to_be_bytes()); bytes.extend_from_slice(&1u16.to_be_bytes());
bytes.extend_from_slice(b"k");
bytes.push(marker::NULL);
bytes.extend_from_slice(&[0, 0, marker::OBJECT_END]);
let parsed = Amf0Value::parse(&bytes).expect("parse");
assert_eq!(
parsed,
Amf0Value::EcmaArray(vec![("k".to_string(), Amf0Value::Null)])
);
}
#[test]
fn depth_guard_rejects_pathological_nesting_without_stack_overflow() {
let mut inner = vec![marker::NULL];
for _ in 0..(MAX_AMF0_DEPTH * 4) {
let mut wrapped = vec![marker::OBJECT];
wrapped.extend_from_slice(&1u16.to_be_bytes());
wrapped.push(b'a');
wrapped.extend_from_slice(&inner);
wrapped.extend_from_slice(&[0, 0, marker::OBJECT_END]);
inner = wrapped;
}
let result = Amf0Value::parse(&inner);
assert!(matches!(result, Err(RtmpError::Unsupported { .. })));
}
#[test]
fn depth_guard_allows_nesting_at_the_limit() {
let mut inner = vec![marker::NULL];
for _ in 0..(MAX_AMF0_DEPTH - 1) {
let mut wrapped = vec![marker::OBJECT];
wrapped.extend_from_slice(&1u16.to_be_bytes());
wrapped.push(b'a');
wrapped.extend_from_slice(&inner);
wrapped.extend_from_slice(&[0, 0, marker::OBJECT_END]);
inner = wrapped;
}
assert!(Amf0Value::parse(&inner).is_ok());
}
#[test]
fn dropping_object_end_marker_is_rejected() {
let full = Amf0Value::Object(vec![("k".to_string(), Amf0Value::Null)]).to_bytes();
let truncated = &full[..full.len() - 3];
assert!(Amf0Value::parse(truncated).is_err());
}
#[test]
fn mis_sized_string_length_is_rejected() {
let mut bytes = vec![marker::STRING];
bytes.extend_from_slice(&100u16.to_be_bytes()); bytes.extend_from_slice(b"short"); assert!(matches!(
Amf0Value::parse(&bytes),
Err(RtmpError::BufferTooShort { .. })
));
}
#[test]
fn invalid_utf8_string_is_malformed() {
let mut bytes = vec![marker::STRING];
bytes.extend_from_slice(&2u16.to_be_bytes());
bytes.extend_from_slice(&[0xFF, 0xFE]); assert!(matches!(
Amf0Value::parse(&bytes),
Err(RtmpError::Malformed { .. })
));
}
#[test]
fn unsupported_marker_is_rejected_not_panicking() {
assert!(matches!(
Amf0Value::parse(&[0x11]), Err(RtmpError::Unsupported { .. })
));
assert!(matches!(
Amf0Value::parse(&[0x07]), Err(RtmpError::Unsupported { .. })
));
}
#[test]
fn date_rejects_nonzero_reserved_timezone() {
let mut bytes = vec![marker::DATE];
bytes.extend_from_slice(&0.0f64.to_be_bytes());
bytes.extend_from_slice(&1u16.to_be_bytes()); assert!(matches!(
Amf0Value::parse(&bytes),
Err(RtmpError::Malformed { .. })
));
}
#[test]
fn empty_buffer_and_truncated_marker_are_buffer_too_short_not_panics() {
assert!(matches!(
Amf0Value::parse(&[]),
Err(RtmpError::BufferTooShort { .. })
));
assert!(matches!(
Amf0Value::parse(&[marker::NUMBER, 0, 0, 0]),
Err(RtmpError::BufferTooShort { .. })
));
}
fn connect_command() -> Command {
Command {
name: "connect".to_string(),
transaction_id: 1.0,
arguments: vec![Amf0Value::Object(vec![
("app".to_string(), Amf0Value::String("live".to_string())),
(
"flashVer".to_string(),
Amf0Value::String("FMLE/3.0".to_string()),
),
(
"tcUrl".to_string(),
Amf0Value::String("rtmp://example.test/live".to_string()),
),
("fpad".to_string(), Amf0Value::Boolean(false)),
])],
}
}
fn publish_command() -> Command {
Command {
name: "publish".to_string(),
transaction_id: 5.0,
arguments: vec![
Amf0Value::Null,
Amf0Value::String("stream_key_123".to_string()),
Amf0Value::String("live".to_string()),
],
}
}
#[test]
fn connect_command_round_trips_byte_identically() {
let cmd = connect_command();
let bytes = cmd.to_body();
let parsed = Command::parse(&bytes).expect("parse connect");
assert_eq!(parsed, cmd);
assert_eq!(parsed.to_body(), bytes);
}
#[test]
fn publish_command_round_trips_byte_identically() {
let cmd = publish_command();
let bytes = cmd.to_body();
let parsed = Command::parse(&bytes).expect("parse publish");
assert_eq!(parsed, cmd);
assert_eq!(parsed.to_body(), bytes);
}
#[test]
fn command_name_must_be_string() {
let bytes = Amf0Value::Number(1.0).to_bytes();
assert!(matches!(
Command::parse(&bytes),
Err(RtmpError::Malformed { .. })
));
}
#[test]
fn command_transaction_id_must_be_number() {
let mut bytes = Amf0Value::String("connect".to_string()).to_bytes();
bytes.extend(Amf0Value::String("not a number".to_string()).to_bytes());
assert!(matches!(
Command::parse(&bytes),
Err(RtmpError::Malformed { .. })
));
}
#[test]
fn long_string_length_overflowing_usize_is_rejected_not_wrapped() {
let mut bytes = vec![marker::LONG_STRING];
bytes.extend_from_slice(&(u32::MAX - 1).to_be_bytes());
bytes.extend_from_slice(b"short");
let err = Amf0Value::parse(&bytes).unwrap_err();
assert!(matches!(
err,
RtmpError::Malformed { .. } | RtmpError::BufferTooShort { .. }
));
}
#[cfg(feature = "serde")]
#[test]
fn amf0_value_and_command_serde_round_trip() {
let value = Amf0Value::Object(vec![
("app".to_string(), Amf0Value::String("live".to_string())),
("live".to_string(), Amf0Value::Boolean(true)),
("duration".to_string(), Amf0Value::Number(0.0)),
(
"items".to_string(),
Amf0Value::StrictArray(vec![Amf0Value::Null, Amf0Value::Undefined]),
),
]);
let json = serde_json::to_string(&value).expect("serialize Amf0Value");
let back: Amf0Value = serde_json::from_str(&json).expect("deserialize Amf0Value");
assert_eq!(back, value);
let cmd = publish_command();
let json = serde_json::to_string(&cmd).expect("serialize Command");
let back: Command = serde_json::from_str(&json).expect("deserialize Command");
assert_eq!(back, cmd);
}
}