1use broadcast_common::{Parse, Serialize};
31
32use crate::RtmpError;
33
34type Result<T> = core::result::Result<T, RtmpError>;
35
36pub const RTMP_VERSION: u8 = 3;
38
39pub const HANDSHAKE_PACKET_LEN: usize = 1536;
41
42const VERSION_LEN: usize = 1;
44const FIELD_LEN: usize = 4;
46const RANDOM_LEN: usize = HANDSHAKE_PACKET_LEN - FIELD_LEN - FIELD_LEN;
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub struct Version(pub u8);
54
55impl<'a> Parse<'a> for Version {
56 type Error = RtmpError;
57
58 fn parse(bytes: &'a [u8]) -> Result<Self> {
59 if bytes.len() < VERSION_LEN {
60 return Err(RtmpError::BufferTooShort {
61 need: VERSION_LEN,
62 have: bytes.len(),
63 what: "C0/S0 version",
64 });
65 }
66 Ok(Version(bytes[0]))
67 }
68}
69
70impl Serialize for Version {
71 type Error = RtmpError;
72
73 fn serialized_len(&self) -> usize {
74 VERSION_LEN
75 }
76
77 fn serialize_into(&self, buf: &mut [u8]) -> Result<usize> {
78 if buf.len() < VERSION_LEN {
79 return Err(RtmpError::BufferTooShort {
80 need: VERSION_LEN,
81 have: buf.len(),
82 what: "C0/S0 version output",
83 });
84 }
85 buf[0] = self.0;
86 Ok(VERSION_LEN)
87 }
88}
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub struct HandshakePacket {
94 pub time: u32,
97 pub zero: u32,
99 pub random: [u8; RANDOM_LEN],
102}
103
104impl<'a> Parse<'a> for HandshakePacket {
105 type Error = RtmpError;
106
107 fn parse(bytes: &'a [u8]) -> Result<Self> {
108 if bytes.len() < HANDSHAKE_PACKET_LEN {
109 return Err(RtmpError::BufferTooShort {
110 need: HANDSHAKE_PACKET_LEN,
111 have: bytes.len(),
112 what: "C1/S1",
113 });
114 }
115 let time = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
116 let zero = u32::from_be_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
117 let mut random = [0u8; RANDOM_LEN];
118 random.copy_from_slice(&bytes[2 * FIELD_LEN..HANDSHAKE_PACKET_LEN]);
119 Ok(HandshakePacket { time, zero, random })
120 }
121}
122
123impl Serialize for HandshakePacket {
124 type Error = RtmpError;
125
126 fn serialized_len(&self) -> usize {
127 HANDSHAKE_PACKET_LEN
128 }
129
130 fn serialize_into(&self, buf: &mut [u8]) -> Result<usize> {
131 if buf.len() < HANDSHAKE_PACKET_LEN {
132 return Err(RtmpError::BufferTooShort {
133 need: HANDSHAKE_PACKET_LEN,
134 have: buf.len(),
135 what: "C1/S1 output",
136 });
137 }
138 buf[0..FIELD_LEN].copy_from_slice(&self.time.to_be_bytes());
139 buf[FIELD_LEN..2 * FIELD_LEN].copy_from_slice(&self.zero.to_be_bytes());
140 buf[2 * FIELD_LEN..HANDSHAKE_PACKET_LEN].copy_from_slice(&self.random);
141 Ok(HANDSHAKE_PACKET_LEN)
142 }
143}
144
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
148pub struct EchoPacket {
149 pub time: u32,
152 pub time2: u32,
155 pub random_echo: [u8; RANDOM_LEN],
158}
159
160impl<'a> Parse<'a> for EchoPacket {
161 type Error = RtmpError;
162
163 fn parse(bytes: &'a [u8]) -> Result<Self> {
164 if bytes.len() < HANDSHAKE_PACKET_LEN {
165 return Err(RtmpError::BufferTooShort {
166 need: HANDSHAKE_PACKET_LEN,
167 have: bytes.len(),
168 what: "C2/S2",
169 });
170 }
171 let time = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
172 let time2 = u32::from_be_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]);
173 let mut random_echo = [0u8; RANDOM_LEN];
174 random_echo.copy_from_slice(&bytes[2 * FIELD_LEN..HANDSHAKE_PACKET_LEN]);
175 Ok(EchoPacket {
176 time,
177 time2,
178 random_echo,
179 })
180 }
181}
182
183impl Serialize for EchoPacket {
184 type Error = RtmpError;
185
186 fn serialized_len(&self) -> usize {
187 HANDSHAKE_PACKET_LEN
188 }
189
190 fn serialize_into(&self, buf: &mut [u8]) -> Result<usize> {
191 if buf.len() < HANDSHAKE_PACKET_LEN {
192 return Err(RtmpError::BufferTooShort {
193 need: HANDSHAKE_PACKET_LEN,
194 have: buf.len(),
195 what: "C2/S2 output",
196 });
197 }
198 buf[0..FIELD_LEN].copy_from_slice(&self.time.to_be_bytes());
199 buf[FIELD_LEN..2 * FIELD_LEN].copy_from_slice(&self.time2.to_be_bytes());
200 buf[2 * FIELD_LEN..HANDSHAKE_PACKET_LEN].copy_from_slice(&self.random_echo);
201 Ok(HANDSHAKE_PACKET_LEN)
202 }
203}
204
205#[must_use]
212pub fn default_random_fill() -> [u8; RANDOM_LEN] {
213 let mut random = [0u8; RANDOM_LEN];
214 for (i, b) in random.iter_mut().enumerate() {
215 *b = (i & 0xFF) as u8;
216 }
217 random
218}
219
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
224enum HandshakeState {
225 WaitC0C1,
227 WaitC2,
230 Done,
232}
233
234#[derive(Debug)]
241pub struct Handshake {
242 state: HandshakeState,
243 local_time: u32,
245 local_random: [u8; RANDOM_LEN],
247 read_time: u32,
251}
252
253impl Default for Handshake {
254 fn default() -> Self {
255 Self::new()
256 }
257}
258
259impl Handshake {
260 #[must_use]
265 pub fn new() -> Self {
266 Self::with_time_and_random(0, default_random_fill(), 0)
267 }
268
269 #[must_use]
272 pub fn with_time_and_random(
273 local_time: u32,
274 local_random: [u8; RANDOM_LEN],
275 read_time: u32,
276 ) -> Self {
277 Self {
278 state: HandshakeState::WaitC0C1,
279 local_time,
280 local_random,
281 read_time,
282 }
283 }
284
285 #[must_use]
287 pub fn is_done(&self) -> bool {
288 self.state == HandshakeState::Done
289 }
290
291 pub fn read(&mut self, input: &[u8]) -> Result<(Vec<u8>, usize, bool)> {
310 match self.state {
311 HandshakeState::WaitC0C1 => {
312 let need = VERSION_LEN + HANDSHAKE_PACKET_LEN;
313 if input.len() < need {
314 return Err(RtmpError::BufferTooShort {
315 need,
316 have: input.len(),
317 what: "C0+C1",
318 });
319 }
320 let _c0 = Version::parse(&input[..VERSION_LEN])?;
321 let c1 = HandshakePacket::parse(&input[VERSION_LEN..need])?;
322
323 let s0 = Version(RTMP_VERSION);
324 let s1 = HandshakePacket {
325 time: self.local_time,
326 zero: 0,
327 random: self.local_random,
328 };
329 let s2 = EchoPacket {
330 time: c1.time,
331 time2: self.read_time,
332 random_echo: c1.random,
333 };
334
335 let reply_len = VERSION_LEN + HANDSHAKE_PACKET_LEN + HANDSHAKE_PACKET_LEN;
336 let mut reply = vec![0u8; reply_len];
337 s0.serialize_into(&mut reply[..VERSION_LEN])?;
338 s1.serialize_into(&mut reply[VERSION_LEN..VERSION_LEN + HANDSHAKE_PACKET_LEN])?;
339 s2.serialize_into(&mut reply[VERSION_LEN + HANDSHAKE_PACKET_LEN..])?;
340
341 self.state = HandshakeState::WaitC2;
342 Ok((reply, need, false))
343 }
344 HandshakeState::WaitC2 => {
345 if input.len() < HANDSHAKE_PACKET_LEN {
346 return Err(RtmpError::BufferTooShort {
347 need: HANDSHAKE_PACKET_LEN,
348 have: input.len(),
349 what: "C2",
350 });
351 }
352 let _c2 = EchoPacket::parse(&input[..HANDSHAKE_PACKET_LEN])?;
353 self.state = HandshakeState::Done;
354 Ok((Vec::new(), HANDSHAKE_PACKET_LEN, true))
355 }
356 HandshakeState::Done => Ok((Vec::new(), 0, true)),
357 }
358 }
359}
360
361#[cfg(test)]
362mod tests {
363 use super::*;
364
365 fn patterned_random(seed: u8) -> [u8; RANDOM_LEN] {
366 let mut r = [0u8; RANDOM_LEN];
367 for (i, b) in r.iter_mut().enumerate() {
368 *b = seed.wrapping_add((i & 0xFF) as u8);
369 }
370 r
371 }
372
373 #[test]
376 fn version_round_trip_build_serialize_parse() {
377 let v = Version(RTMP_VERSION);
378 let mut buf = [0u8; VERSION_LEN];
379 let n = v.serialize_into(&mut buf).unwrap();
380 assert_eq!(n, VERSION_LEN);
381 let parsed = Version::parse(&buf).unwrap();
382 assert_eq!(parsed, v);
383 }
384
385 #[test]
386 fn version_round_trip_parse_serialize_byte_identical() {
387 let bytes = [3u8];
388 let v = Version::parse(&bytes).unwrap();
389 let mut buf = [0u8; VERSION_LEN];
390 v.serialize_into(&mut buf).unwrap();
391 assert_eq!(buf, bytes);
392 }
393
394 #[test]
395 fn version_short_input_is_buffer_too_short() {
396 let bytes: [u8; 0] = [];
397 assert!(matches!(
398 Version::parse(&bytes),
399 Err(RtmpError::BufferTooShort {
400 need: VERSION_LEN,
401 have: 0,
402 ..
403 })
404 ));
405 }
406
407 #[test]
410 fn handshake_packet_round_trip_build_serialize_parse() {
411 let hp = HandshakePacket {
412 time: 0x1122_3344,
413 zero: 0,
414 random: patterned_random(0xAB),
415 };
416 let mut buf = [0u8; HANDSHAKE_PACKET_LEN];
417 let n = hp.serialize_into(&mut buf).unwrap();
418 assert_eq!(n, HANDSHAKE_PACKET_LEN);
419 let parsed = HandshakePacket::parse(&buf).unwrap();
420 assert_eq!(parsed, hp);
421 }
422
423 #[test]
424 fn handshake_packet_round_trip_parse_serialize_byte_identical() {
425 let mut bytes = [0u8; HANDSHAKE_PACKET_LEN];
426 bytes[0..4].copy_from_slice(&0xDEAD_BEEFu32.to_be_bytes());
427 for (i, b) in bytes[8..].iter_mut().enumerate() {
429 *b = (i as u8).wrapping_mul(7);
430 }
431 let hp = HandshakePacket::parse(&bytes).unwrap();
432 assert_eq!(hp.time, 0xDEAD_BEEF);
433 assert_eq!(hp.zero, 0);
434 let mut buf = [0u8; HANDSHAKE_PACKET_LEN];
435 hp.serialize_into(&mut buf).unwrap();
436 assert_eq!(buf, bytes, "C1/S1 byte-identical round trip");
437 }
438
439 #[test]
440 fn handshake_packet_short_input_is_buffer_too_short() {
441 let bytes = [0u8; HANDSHAKE_PACKET_LEN - 1];
442 assert!(matches!(
443 HandshakePacket::parse(&bytes),
444 Err(RtmpError::BufferTooShort {
445 need: HANDSHAKE_PACKET_LEN,
446 have,
447 ..
448 }) if have == HANDSHAKE_PACKET_LEN - 1
449 ));
450 }
451
452 #[test]
455 fn echo_packet_round_trip_build_serialize_parse() {
456 let ep = EchoPacket {
457 time: 0x0102_0304,
458 time2: 0x0506_0708,
459 random_echo: patterned_random(0x5A),
460 };
461 let mut buf = [0u8; HANDSHAKE_PACKET_LEN];
462 let n = ep.serialize_into(&mut buf).unwrap();
463 assert_eq!(n, HANDSHAKE_PACKET_LEN);
464 let parsed = EchoPacket::parse(&buf).unwrap();
465 assert_eq!(parsed, ep);
466 }
467
468 #[test]
469 fn echo_packet_round_trip_parse_serialize_byte_identical() {
470 let mut bytes = [0u8; HANDSHAKE_PACKET_LEN];
471 bytes[0..4].copy_from_slice(&0x1111_2222u32.to_be_bytes());
472 bytes[4..8].copy_from_slice(&0x3333_4444u32.to_be_bytes());
473 for (i, b) in bytes[8..].iter_mut().enumerate() {
474 *b = (i as u8) ^ 0x5A;
475 }
476 let ep = EchoPacket::parse(&bytes).unwrap();
477 let mut buf = [0u8; HANDSHAKE_PACKET_LEN];
478 ep.serialize_into(&mut buf).unwrap();
479 assert_eq!(buf, bytes, "C2/S2 byte-identical round trip");
480 }
481
482 #[test]
483 fn echo_packet_short_input_is_buffer_too_short() {
484 let bytes = [0u8; 10];
485 assert!(matches!(
486 EchoPacket::parse(&bytes),
487 Err(RtmpError::BufferTooShort {
488 need: HANDSHAKE_PACKET_LEN,
489 have: 10,
490 ..
491 })
492 ));
493 }
494
495 fn build_c0_c1(client_time: u32, client_random: [u8; RANDOM_LEN]) -> Vec<u8> {
498 let mut v = vec![0u8; VERSION_LEN + HANDSHAKE_PACKET_LEN];
499 v[0] = RTMP_VERSION;
500 let c1 = HandshakePacket {
501 time: client_time,
502 zero: 0,
503 random: client_random,
504 };
505 c1.serialize_into(&mut v[VERSION_LEN..]).unwrap();
506 v
507 }
508
509 fn build_c2(time: u32, time2: u32, random_echo: [u8; RANDOM_LEN]) -> Vec<u8> {
510 let c2 = EchoPacket {
511 time,
512 time2,
513 random_echo,
514 };
515 let mut v = vec![0u8; HANDSHAKE_PACKET_LEN];
516 c2.serialize_into(&mut v).unwrap();
517 v
518 }
519
520 #[test]
521 fn fsm_happy_path_reaches_done_and_s2_echoes_c1() {
522 let client_time = 0x0000_1234;
523 let client_random = patterned_random(0x77);
524
525 let mut hs = Handshake::new();
526 assert!(!hs.is_done());
527
528 let c0_c1 = build_c0_c1(client_time, client_random);
529 let (reply, consumed, done) = hs.read(&c0_c1).unwrap();
530
531 assert_eq!(consumed, VERSION_LEN + HANDSHAKE_PACKET_LEN);
532 assert!(!done);
533 assert!(!hs.is_done());
534 assert_eq!(
535 reply.len(),
536 VERSION_LEN + HANDSHAKE_PACKET_LEN + HANDSHAKE_PACKET_LEN,
537 "S0+S1+S2 must total 3073 bytes"
538 );
539
540 let s0 = Version::parse(&reply[..VERSION_LEN]).unwrap();
542 assert_eq!(s0.0, RTMP_VERSION);
543
544 let s1_start = VERSION_LEN;
546 let s1_end = s1_start + HANDSHAKE_PACKET_LEN;
547 let _s1 = HandshakePacket::parse(&reply[s1_start..s1_end]).unwrap();
548
549 let s2 = EchoPacket::parse(&reply[s1_end..]).unwrap();
551 assert_eq!(s2.time, client_time, "S2.time must echo C1.time");
552 assert_eq!(
553 s2.random_echo, client_random,
554 "S2.random_echo must equal C1's random bytes verbatim"
555 );
556
557 let s1 = HandshakePacket::parse(&reply[s1_start..s1_end]).unwrap();
560 let c2 = build_c2(s1.time, 0, s1.random);
561 let (reply2, consumed2, done2) = hs.read(&c2).unwrap();
562 assert_eq!(consumed2, HANDSHAKE_PACKET_LEN);
563 assert!(done2);
564 assert!(hs.is_done());
565 assert!(
566 reply2.is_empty(),
567 "C2 receipt produces no further reply bytes"
568 );
569 }
570
571 #[test]
572 fn fsm_partial_c1_is_buffer_too_short() {
573 let mut hs = Handshake::new();
574 let full = build_c0_c1(0, patterned_random(1));
576 let partial = &full[..full.len() - 1];
577 let err = hs.read(partial).unwrap_err();
578 assert!(matches!(err, RtmpError::BufferTooShort { .. }));
579 assert!(!hs.is_done(), "state must not advance on a short read");
580 }
581
582 #[test]
583 fn fsm_partial_c2_is_buffer_too_short() {
584 let mut hs = Handshake::new();
585 let c0_c1 = build_c0_c1(0, patterned_random(2));
586 let (_reply, _consumed, done) = hs.read(&c0_c1).unwrap();
587 assert!(!done);
588
589 let full_c2 = build_c2(0, 0, patterned_random(2));
590 let partial_c2 = &full_c2[..full_c2.len() - 1];
591 let err = hs.read(partial_c2).unwrap_err();
592 assert!(matches!(err, RtmpError::BufferTooShort { .. }));
593 assert!(
594 !hs.is_done(),
595 "state must not advance to Done on a short C2 read"
596 );
597 }
598
599 #[test]
600 fn fsm_done_is_idempotent_noop() {
601 let mut hs = Handshake::new();
602 let c0_c1 = build_c0_c1(0, patterned_random(3));
603 hs.read(&c0_c1).unwrap();
604 let full_c2 = build_c2(0, 0, patterned_random(3));
605 hs.read(&full_c2).unwrap();
606 assert!(hs.is_done());
607
608 let (reply, consumed, done) = hs.read(&[]).unwrap();
609 assert!(reply.is_empty());
610 assert_eq!(consumed, 0);
611 assert!(done);
612 }
613
614 #[test]
620 fn handshake_packet_serialize_writes_full_random_tail() {
621 let hp = HandshakePacket {
622 time: 1,
623 zero: 0,
624 random: [0xFFu8; RANDOM_LEN],
625 };
626 let mut buf = [0u8; HANDSHAKE_PACKET_LEN];
627 hp.serialize_into(&mut buf).unwrap();
628 assert!(
629 buf[8..].iter().all(|&b| b == 0xFF),
630 "every random byte must be written, not just a prefix"
631 );
632 }
633
634 #[test]
635 fn echo_packet_random_echo_must_differ_from_local_pattern_to_catch_non_echo_bugs() {
636 assert_ne!(patterned_random(0x77), default_random_fill());
640 }
641}