1use bytes::{Bytes, BytesMut};
12
13use crate::error::{FramingError, HeaderSyntaxError, LimitKind, ParseError, StartLineError};
14use crate::message::{Header, Headers, Message, Method, Request, Response, StatusCode, Version};
15use crate::name::HeaderName;
16use crate::uri::Uri;
17
18#[derive(Debug, Clone, Copy)]
24pub struct Limits {
25 pub max_message_bytes: usize,
27 pub max_body_bytes: usize,
29 pub max_headers: usize,
31 pub max_header_bytes: usize,
33 pub max_folding_lines: usize,
35}
36
37impl Limits {
38 #[must_use]
40 pub fn datagram() -> Self {
41 Self {
42 max_message_bytes: 64 * 1024,
43 max_body_bytes: 64 * 1024,
44 max_headers: 256,
45 max_header_bytes: 8 * 1024,
46 max_folding_lines: 16,
47 }
48 }
49
50 #[must_use]
52 pub fn stream() -> Self {
53 Self {
54 max_message_bytes: 1024 * 1024,
55 max_body_bytes: 1024 * 1024,
56 ..Self::datagram()
57 }
58 }
59}
60
61impl Default for Limits {
62 fn default() -> Self {
63 Self::datagram()
64 }
65}
66
67#[allow(clippy::needless_pass_by_value)]
76pub fn parse_datagram(buf: Bytes, limits: &Limits) -> Result<Message, ParseError> {
77 if buf.len() > limits.max_message_bytes {
78 return Err(ParseError::Limit {
79 limit: LimitKind::MessageBytes,
80 value: buf.len(),
81 });
82 }
83
84 let head_end = find_header_terminator(&buf, 0)
85 .ok_or(ParseError::Framing(FramingError::NoHeaderTerminator))?;
86 let head = buf.slice(..head_end);
87 let rest = buf.slice(head_end + 4..);
88
89 let (start, headers) = parse_head(head, limits)?;
90
91 let body = if let Some(declared) = content_length(&headers)? {
92 check_body_limit(declared, limits)?;
93 let declared = usize::try_from(declared).unwrap_or(usize::MAX);
94 if declared > rest.len() {
95 return Err(ParseError::Framing(FramingError::BodyTruncated));
96 }
97 rest.slice(..declared)
98 } else {
99 check_body_limit(rest.len() as u64, limits)?;
101 rest
102 };
103
104 Ok(assemble(start, headers, body))
105}
106
107#[derive(Debug)]
113pub struct StreamParser {
114 buf: BytesMut,
115 limits: Limits,
116 state: State,
117 scanned: usize,
120 failed: bool,
121 keepalives: usize,
128}
129
130#[derive(Debug)]
131enum State {
132 Head,
133 Body {
134 pending: Box<Pending>,
135 needed: usize,
136 },
137}
138
139#[derive(Debug)]
140struct Pending {
141 start: StartLine,
142 headers: Headers,
143}
144
145impl StreamParser {
146 #[must_use]
148 pub fn new(limits: Limits) -> Self {
149 Self {
150 buf: BytesMut::new(),
151 limits,
152 state: State::Head,
153 scanned: 0,
154 failed: false,
155 keepalives: 0,
156 }
157 }
158
159 pub fn take_keepalives(&mut self) -> usize {
168 std::mem::take(&mut self.keepalives)
169 }
170
171 #[must_use]
176 pub fn pending(&self) -> usize {
177 self.buf.len()
178 }
179
180 pub fn push(&mut self, chunk: &[u8]) -> Result<Vec<Message>, ParseError> {
186 if self.failed {
187 return Err(ParseError::Framing(FramingError::NoHeaderTerminator));
188 }
189 self.buf.extend_from_slice(chunk);
190 match self.drain() {
191 Ok(messages) => Ok(messages),
192 Err(e) => {
193 self.failed = true;
194 Err(e)
195 }
196 }
197 }
198
199 fn drain(&mut self) -> Result<Vec<Message>, ParseError> {
200 let mut out = Vec::new();
201 loop {
202 match &self.state {
203 State::Head => {
204 while self.buf.starts_with(b"\r\n") {
209 let _crlf = self.buf.split_to(2);
210 self.scanned = self.scanned.saturating_sub(2);
211 self.keepalives = self.keepalives.saturating_add(1);
212 }
213 let Some(head_end) = find_header_terminator(&self.buf, self.scanned) else {
214 self.scanned = self.buf.len().saturating_sub(3);
217 if self.buf.len() > self.limits.max_message_bytes {
218 return Err(ParseError::Limit {
219 limit: LimitKind::MessageBytes,
220 value: self.buf.len(),
221 });
222 }
223 return Ok(out);
224 };
225
226 let head = self.buf.split_to(head_end).freeze();
227 let _crlfcrlf = self.buf.split_to(4);
228 self.scanned = 0;
229
230 let (start, headers) = parse_head(head, &self.limits)?;
231 let declared = content_length(&headers)?
234 .ok_or(ParseError::Framing(FramingError::ContentLengthRequired))?;
235 check_body_limit(declared, &self.limits)?;
236 let needed = usize::try_from(declared).unwrap_or(usize::MAX);
237 self.state = State::Body {
238 pending: Box::new(Pending { start, headers }),
239 needed,
240 };
241 }
242 State::Body { needed, .. } => {
243 let needed = *needed;
244 if self.buf.len() < needed {
245 return Ok(out);
246 }
247 let body = self.buf.split_to(needed).freeze();
248 let State::Body { pending, .. } =
249 std::mem::replace(&mut self.state, State::Head)
250 else {
251 unreachable!("state was just observed to be Body")
252 };
253 let Pending { start, headers } = *pending;
254 out.push(assemble(start, headers, body));
255 }
256 }
257 }
258 }
259}
260
261#[derive(Debug)]
263enum StartLine {
264 Request {
265 method: Method,
266 uri: Box<Uri>,
267 uri_span: std::ops::Range<usize>,
268 version: Version,
269 raw: Bytes,
270 },
271 Response {
272 version: Version,
273 status: StatusCode,
274 reason: Bytes,
275 raw: Bytes,
276 },
277}
278
279fn assemble(start: StartLine, headers: Headers, body: Bytes) -> Message {
280 match start {
281 StartLine::Request {
282 method,
283 uri,
284 uri_span,
285 version,
286 raw,
287 } => Message::Request(Request::from_wire(
288 method, *uri, version, raw, uri_span, headers, body,
289 )),
290 StartLine::Response {
291 version,
292 status,
293 reason,
294 raw,
295 } => Message::Response(Response::from_wire(
296 version, status, reason, raw, headers, body,
297 )),
298 }
299}
300
301fn check_body_limit(declared: u64, limits: &Limits) -> Result<(), ParseError> {
302 if declared > limits.max_body_bytes as u64 {
303 return Err(ParseError::Limit {
304 limit: LimitKind::BodyBytes,
305 value: usize::try_from(declared).unwrap_or(usize::MAX),
306 });
307 }
308 Ok(())
309}
310
311fn find_header_terminator(buf: &[u8], from: usize) -> Option<usize> {
313 buf.get(from..)
314 .and_then(|tail| tail.windows(4).position(|w| w == b"\r\n\r\n"))
315 .map(|i| i + from)
316}
317
318#[allow(clippy::needless_pass_by_value)] fn parse_head(head: Bytes, limits: &Limits) -> Result<(StartLine, Headers), ParseError> {
324 validate_line_endings(&head)?;
325
326 let lines = split_folded_lines(&head, limits)?;
327 let mut lines = lines.into_iter();
328 let (start_from, start_to) = lines.next().ok_or(StartLineError::Empty)?;
329 let start = parse_start_line(head.slice(start_from..start_to))?;
330
331 let mut headers = Headers::new();
332 for (index, (from, to)) in lines.enumerate() {
333 if headers.len() >= limits.max_headers {
334 return Err(ParseError::Limit {
335 limit: LimitKind::Headers,
336 value: headers.len() + 1,
337 });
338 }
339 let line = head.slice(from..to);
340 if line.len() > limits.max_header_bytes {
341 return Err(ParseError::Limit {
342 limit: LimitKind::HeaderBytes,
343 value: line.len(),
344 });
345 }
346 headers.push(parse_header_line(line, index + 2)?);
347 }
348
349 Ok((start, headers))
350}
351
352fn validate_line_endings(head: &[u8]) -> Result<(), ParseError> {
357 let mut i = 0;
358 while let Some(&b) = head.get(i) {
359 match b {
360 b'\r' if head.get(i + 1) == Some(&b'\n') => i += 2,
361 b'\r' | b'\n' => {
362 return Err(ParseError::HeaderSyntax {
363 line: 1 + head.get(..i).map_or(0, count_lines),
364 kind: HeaderSyntaxError::BareNewline,
365 });
366 }
367 _ => i += 1,
368 }
369 }
370 Ok(())
371}
372
373fn count_lines(prefix: &[u8]) -> usize {
374 prefix.windows(2).filter(|w| *w == b"\r\n").count()
375}
376
377fn split_folded_lines(head: &[u8], limits: &Limits) -> Result<Vec<(usize, usize)>, ParseError> {
383 let mut lines = Vec::new();
384 let mut start = 0usize;
385 let mut i = 0usize;
386 let mut folds_in_line = 0usize;
387
388 if matches!(head.first(), Some(b' ' | b'\t')) {
389 return Err(ParseError::HeaderSyntax {
390 line: 1,
391 kind: HeaderSyntaxError::LeadingFold,
392 });
393 }
394
395 while i < head.len() {
396 if head.get(i) == Some(&b'\r') && head.get(i + 1) == Some(&b'\n') {
397 if matches!(head.get(i + 2), Some(b' ' | b'\t')) {
398 folds_in_line += 1;
399 if folds_in_line > limits.max_folding_lines {
400 return Err(ParseError::Limit {
401 limit: LimitKind::FoldingLines,
402 value: folds_in_line,
403 });
404 }
405 i += 2;
406 continue;
407 }
408 lines.push((start, i));
409 i += 2;
410 start = i;
411 folds_in_line = 0;
412 } else {
413 i += 1;
414 }
415 }
416 if start < head.len() {
417 lines.push((start, head.len()));
418 }
419 Ok(lines)
420}
421
422fn parse_start_line(line: Bytes) -> Result<StartLine, ParseError> {
423 if line.is_empty() {
424 return Err(StartLineError::Empty.into());
425 }
426
427 if line
431 .get(..4)
432 .is_some_and(|prefix| prefix.eq_ignore_ascii_case(b"SIP/"))
433 {
434 return parse_status_line(line);
435 }
436
437 let mut bounds = Vec::new();
441 let mut start = 0usize;
442 for (i, &b) in line.iter().enumerate() {
443 if b == b' ' {
444 bounds.push((start, i));
445 start = i + 1;
446 }
447 }
448 bounds.push((start, line.len()));
449
450 if bounds.len() != 3 || bounds.iter().any(|(a, b)| a == b) {
451 return Err(StartLineError::RequestLineShape.into());
452 }
453 let Some(&(m0, m1)) = bounds.first() else {
454 return Err(StartLineError::RequestLineShape.into());
455 };
456 let Some(&(u0, u1)) = bounds.get(1) else {
457 return Err(StartLineError::RequestLineShape.into());
458 };
459 let Some(&(v0, v1)) = bounds.get(2) else {
460 return Err(StartLineError::RequestLineShape.into());
461 };
462
463 let method_raw = line.slice(m0..m1);
464 if !method_raw.iter().all(|&b| is_token_char(b)) {
465 return Err(StartLineError::Method.into());
466 }
467 let uri = Uri::parse(line.slice(u0..u1)).map_err(StartLineError::Uri)?;
468
469 Ok(StartLine::Request {
470 method: Method::parse(&method_raw),
471 uri: Box::new(uri),
472 uri_span: u0..u1,
473 version: Version::parse(&line.slice(v0..v1)),
474 raw: line,
475 })
476}
477
478fn parse_status_line(line: Bytes) -> Result<StartLine, ParseError> {
479 let first = line
482 .iter()
483 .position(|&b| b == b' ')
484 .ok_or(StartLineError::MissingStatusCode)?;
485 let version = Version::parse(&line.slice(..first));
486
487 let after = line.slice(first + 1..);
488 let (code_raw, reason) = match after.iter().position(|&b| b == b' ') {
489 Some(second) => (after.slice(..second), after.slice(second + 1..)),
490 None => (after.clone(), Bytes::new()),
494 };
495
496 if code_raw.len() != 3 || !code_raw.iter().all(u8::is_ascii_digit) {
497 return Err(StartLineError::StatusCode.into());
498 }
499 let value = code_raw
500 .iter()
501 .fold(0u16, |acc, &b| acc * 10 + u16::from(b - b'0'));
502 let status = StatusCode::new(value).ok_or(StartLineError::StatusCode)?;
503
504 Ok(StartLine::Response {
505 version,
506 status,
507 reason,
508 raw: line,
509 })
510}
511
512fn parse_header_line(line: Bytes, line_number: usize) -> Result<Header, ParseError> {
513 let colon = line
514 .iter()
515 .position(|&b| b == b':')
516 .ok_or(ParseError::HeaderSyntax {
517 line: line_number,
518 kind: HeaderSyntaxError::MissingColon,
519 })?;
520
521 let mut name_end = colon;
523 while name_end > 0 && matches!(line.get(name_end - 1), Some(b' ' | b'\t')) {
524 name_end -= 1;
525 }
526 let name_raw = line.slice(..name_end);
527
528 if name_raw.is_empty() {
529 return Err(ParseError::HeaderSyntax {
530 line: line_number,
531 kind: HeaderSyntaxError::EmptyName,
532 });
533 }
534 if !name_raw.iter().all(|&b| is_token_char(b)) {
535 return Err(ParseError::HeaderSyntax {
536 line: line_number,
537 kind: HeaderSyntaxError::NameNotToken,
538 });
539 }
540
541 let mut value_offset = colon + 1;
543 loop {
544 match line.get(value_offset) {
545 Some(b' ' | b'\t') => value_offset += 1,
546 Some(b'\r')
547 if line.get(value_offset + 1) == Some(&b'\n')
548 && matches!(line.get(value_offset + 2), Some(b' ' | b'\t')) =>
549 {
550 value_offset += 2;
551 }
552 _ => break,
553 }
554 }
555
556 Ok(Header::from_wire(
557 HeaderName::parse(&name_raw),
558 line,
559 value_offset,
560 ))
561}
562
563fn content_length(headers: &Headers) -> Result<Option<u64>, ParseError> {
565 let mut found: Option<u64> = None;
566 for header in headers.get_all(&HeaderName::ContentLength) {
567 if found.is_some() {
568 return Err(ParseError::Framing(FramingError::ContentLengthRepeated));
569 }
570 let value = header.value();
571 if value.is_empty() || !value.iter().all(u8::is_ascii_digit) {
575 return Err(ParseError::Framing(FramingError::ContentLengthMalformed));
576 }
577 let mut n: u64 = 0;
578 for &b in value.iter() {
579 n = n
580 .checked_mul(10)
581 .and_then(|n| n.checked_add(u64::from(b - b'0')))
582 .ok_or(ParseError::Framing(FramingError::ContentLengthMalformed))?;
583 }
584 found = Some(n);
585 }
586 Ok(found)
587}
588
589#[must_use]
591fn is_token_char(b: u8) -> bool {
592 b.is_ascii_alphanumeric()
593 || matches!(
594 b,
595 b'-' | b'.' | b'!' | b'%' | b'*' | b'_' | b'+' | b'`' | b'\'' | b'~'
596 )
597}
598
599#[cfg(test)]
600#[allow(
601 clippy::unwrap_used,
602 clippy::expect_used,
603 clippy::panic,
604 clippy::indexing_slicing
605)]
606mod tests {
607 use super::*;
608 use crate::error::LimitKind;
609
610 fn parse(text: &str) -> Result<Message, ParseError> {
611 parse_datagram(Bytes::from(text.to_owned()), &Limits::datagram())
612 }
613
614 const MINIMAL: &str = "OPTIONS sip:a@b.com SIP/2.0\r\n\
615 Via: SIP/2.0/UDP h.example.com;branch=z9hG4bKx\r\n\
616 To: <sip:a@b.com>\r\n\
617 From: <sip:c@d.net>;tag=1\r\n\
618 Call-ID: x@y\r\n\
619 CSeq: 1 OPTIONS\r\n\
620 Content-Length: 0\r\n\r\n";
621
622 #[test]
623 fn parses_a_minimal_request() {
624 let msg = parse(MINIMAL).expect("should parse");
625 let req = msg.as_request().expect("a request");
626 assert_eq!(req.method, Method::Options);
627 assert_eq!(req.version, Version::Sip20);
628 assert_eq!(req.headers.len(), 6);
629 assert!(msg.body().is_empty());
630 assert_eq!(msg.to_bytes(), Bytes::from(MINIMAL));
631 }
632
633 #[test]
634 fn parses_a_response_with_a_reason_containing_spaces() {
635 let text = "SIP/2.0 486 Busy Here Right Now\r\nContent-Length: 0\r\n\r\n";
636 let msg = parse(text).expect("should parse");
637 let res = msg.as_response().expect("a response");
638 assert_eq!(res.status.code(), 486);
639 assert_eq!(res.reason, Bytes::from_static(b"Busy Here Right Now"));
640 assert_eq!(msg.to_bytes(), Bytes::from(text));
641 }
642
643 #[test]
644 fn accepts_an_empty_reason_phrase() {
645 let text = "SIP/2.0 200 \r\nContent-Length: 0\r\n\r\n";
647 let msg = parse(text).expect("should parse");
648 assert!(msg.as_response().expect("a response").reason.is_empty());
649 assert_eq!(msg.to_bytes(), Bytes::from(text));
650 }
651
652 #[test]
653 fn rejects_bare_line_feeds() {
654 let text = "OPTIONS sip:a@b.com SIP/2.0\nContent-Length: 0\r\n\r\n";
655 assert!(matches!(
656 parse(text),
657 Err(ParseError::HeaderSyntax {
658 kind: HeaderSyntaxError::BareNewline,
659 ..
660 })
661 ));
662 }
663
664 #[test]
665 fn rejects_a_header_section_starting_with_a_fold() {
666 let text = " OPTIONS sip:a@b.com SIP/2.0\r\nContent-Length: 0\r\n\r\n";
667 assert!(matches!(
668 parse(text),
669 Err(ParseError::HeaderSyntax {
670 kind: HeaderSyntaxError::LeadingFold,
671 ..
672 })
673 ));
674 }
675
676 #[test]
677 fn content_length_faults_are_named_precisely() {
678 let with = |cl: &str, body: &str| {
679 format!("OPTIONS sip:a@b.com SIP/2.0\r\nContent-Length: {cl}\r\n\r\n{body}")
680 };
681 assert!(matches!(
682 parse(&with("-999", "")),
683 Err(ParseError::Framing(FramingError::ContentLengthMalformed))
684 ));
685 assert!(matches!(
686 parse(&with("five", "")),
687 Err(ParseError::Framing(FramingError::ContentLengthMalformed))
688 ));
689 assert!(matches!(
690 parse(&with("", "")),
691 Err(ParseError::Framing(FramingError::ContentLengthMalformed))
692 ));
693 assert!(matches!(
694 parse(&with("9999", "short")),
695 Err(ParseError::Framing(FramingError::BodyTruncated))
696 ));
697 assert!(matches!(
699 parse(&with("18446744073709551616", "")),
700 Err(ParseError::Framing(FramingError::ContentLengthMalformed))
701 ));
702 }
703
704 #[test]
705 fn repeated_content_length_is_rejected_even_when_the_values_agree() {
706 let text = "OPTIONS sip:a@b.com SIP/2.0\r\n\
707 Content-Length: 0\r\nContent-Length: 0\r\n\r\n";
708 assert!(matches!(
709 parse(text),
710 Err(ParseError::Framing(FramingError::ContentLengthRepeated))
711 ));
712 }
713
714 #[test]
715 fn a_datagram_without_content_length_takes_the_rest_as_body() {
716 let text = "OPTIONS sip:a@b.com SIP/2.0\r\nTo: <sip:a@b.com>\r\n\r\nhello";
718 let msg = parse(text).expect("should parse");
719 assert_eq!(msg.body(), &Bytes::from_static(b"hello"));
720 assert_eq!(msg.to_bytes(), Bytes::from(text));
721 }
722
723 #[test]
724 fn trailing_octets_after_the_body_are_ignored() {
725 let text = "OPTIONS sip:a@b.com SIP/2.0\r\nContent-Length: 0\r\n\r\nINVITE sip:x@y SIP/2.0\r\n\r\n";
728 let msg = parse(text).expect("should parse");
729 assert!(msg.body().is_empty());
730 let out = msg.to_bytes();
731 assert!(text.as_bytes().starts_with(&out));
732 assert!(out.len() < text.len(), "the noise must be dropped");
733 }
734
735 #[test]
736 fn limits_are_enforced_before_allocation() {
737 let limits = Limits {
738 max_body_bytes: 10,
739 ..Limits::datagram()
740 };
741 let text = "OPTIONS sip:a@b.com SIP/2.0\r\nContent-Length: 1073741824\r\n\r\n";
743 assert!(matches!(
744 parse_datagram(Bytes::from(text), &limits),
745 Err(ParseError::Limit {
746 limit: LimitKind::BodyBytes,
747 ..
748 })
749 ));
750
751 let limits = Limits {
752 max_headers: 2,
753 ..Limits::datagram()
754 };
755 let text = "OPTIONS sip:a@b.com SIP/2.0\r\nA: 1\r\nB: 2\r\nC: 3\r\n\r\n";
756 assert!(matches!(
757 parse_datagram(Bytes::from(text), &limits),
758 Err(ParseError::Limit {
759 limit: LimitKind::Headers,
760 ..
761 })
762 ));
763 }
764
765 #[test]
766 fn request_line_shape_is_strict() {
767 for text in [
768 "INVITE sip:a@b.com SIP/2.0\r\n\r\n", "INVITE sip:a@b.com SIP/2.0 \r\n\r\n", "INVITE sip:a@b.com\r\n\r\n", "INVITE <sip:a@b.com> SIP/2.0\r\n\r\n", ] {
773 assert!(
774 matches!(parse(text), Err(ParseError::StartLine(_))),
775 "{text:?} should be rejected"
776 );
777 }
778 }
779
780 #[test]
783 fn sip_version_is_recognized_case_insensitively() {
784 let text = "sip/2.0 200 OK\r\nContent-Length: 0\r\n\r\n";
785 let msg = parse(text).expect("should parse");
786 let res = msg
787 .as_response()
788 .expect("a lower-case version still marks a response");
789 assert_eq!(res.status.code(), 200);
790 assert!(res.version.is_supported());
791 assert_eq!(msg.to_bytes(), Bytes::from(text));
793
794 let text = "OPTIONS sip:a@b.com sip/2.0\r\nContent-Length: 0\r\n\r\n";
795 let msg = parse(text).expect("should parse");
796 let req = msg.as_request().expect("a request");
797 assert!(req.version.is_supported(), "sip/2.0 is SIP/2.0");
798 assert_eq!(msg.to_bytes(), Bytes::from(text));
799 }
800
801 #[test]
802 fn an_unknown_version_parses_so_the_caller_can_answer_505() {
803 let text = "OPTIONS sip:a@b.com SIP/7.0\r\nContent-Length: 0\r\n\r\n";
804 let msg = parse(text).expect("should parse");
805 let req = msg.as_request().expect("a request");
806 assert!(!req.version.is_supported());
807 assert_eq!(req.version.as_bytes(), b"SIP/7.0");
808 }
809
810 #[test]
811 fn stream_parser_requires_content_length() {
812 let mut p = StreamParser::new(Limits::stream());
813 let text = "OPTIONS sip:a@b.com SIP/2.0\r\nTo: <sip:a@b.com>\r\n\r\n";
814 assert!(matches!(
815 p.push(text.as_bytes()),
816 Err(ParseError::Framing(FramingError::ContentLengthRequired))
817 ));
818 }
819
820 #[test]
821 fn stream_parser_returns_two_messages_from_one_chunk() {
822 let mut p = StreamParser::new(Limits::stream());
823 let both = format!("{MINIMAL}{MINIMAL}");
824 let messages = p.push(both.as_bytes()).expect("should parse");
825 assert_eq!(messages.len(), 2);
826 assert_eq!(p.pending(), 0);
827 }
828
829 #[test]
830 fn stream_parser_survives_split_at_every_offset() {
831 let text = "INVITE sip:a@b.com SIP/2.0\r\n\
832 Via: SIP/2.0/TCP h.example.com;branch=z9hG4bKx\r\n\
833 Subject: folded\r\n continuation\r\n\
834 Content-Length: 5\r\n\r\nhello";
835 let bytes = text.as_bytes();
836 let whole = {
837 let mut p = StreamParser::new(Limits::stream());
838 let mut m = p.push(bytes).expect("should parse");
839 assert_eq!(m.len(), 1);
840 m.remove(0).to_bytes()
841 };
842
843 for split in 0..=bytes.len() {
844 let mut p = StreamParser::new(Limits::stream());
845 let (a, b) = bytes.split_at(split);
846 let mut got = p.push(a).expect("first half");
847 got.extend(p.push(b).expect("second half"));
848 assert_eq!(
849 got.len(),
850 1,
851 "split at {split} produced {} messages",
852 got.len()
853 );
854 assert_eq!(
855 got.first().map(Message::to_bytes),
856 Some(whole.clone()),
857 "split at {split} changed the message"
858 );
859 }
860 }
861
862 #[test]
863 fn stream_parser_handles_one_byte_at_a_time() {
864 let mut p = StreamParser::new(Limits::stream());
865 let bytes = MINIMAL.as_bytes();
866 let mut out = Vec::new();
867 for i in 0..bytes.len() {
868 let chunk = bytes.get(i..=i).expect("in range");
869 out.extend(p.push(chunk).expect("should parse"));
870 }
871 assert_eq!(out.len(), 1);
872 assert_eq!(
873 out.first().map(Message::to_bytes),
874 Some(Bytes::from(MINIMAL))
875 );
876 }
877
878 #[test]
882 fn stream_parser_ignores_crlf_before_the_start_line() {
883 let mut p = StreamParser::new(Limits::stream());
884
885 let text = format!("\r\n\r\n{MINIMAL}");
887 let messages = p
888 .push(text.as_bytes())
889 .expect("leading CRLFs are not an error");
890 assert_eq!(messages.len(), 1);
891 assert_eq!(p.pending(), 0);
892
893 assert!(p.push(b"\r\n").expect("a pong is not an error").is_empty());
895 assert_eq!(p.pending(), 0);
896 let messages = p.push(MINIMAL.as_bytes()).expect("framing must survive");
897 assert_eq!(messages.len(), 1);
898 }
899
900 #[test]
903 fn leading_crlf_is_ignored_at_every_split_point() {
904 let text = format!("\r\n\r\n\r\n{MINIMAL}");
905 let bytes = text.as_bytes();
906 for split in 0..=bytes.len() {
907 let mut p = StreamParser::new(Limits::stream());
908 let (a, b) = bytes.split_at(split);
909 let mut got = p.push(a).expect("first half");
910 got.extend(p.push(b).expect("second half"));
911 assert_eq!(
912 got.len(),
913 1,
914 "split at {split} produced {} messages",
915 got.len()
916 );
917 assert_eq!(
918 got.first().map(Message::to_bytes),
919 Some(Bytes::from(MINIMAL)),
920 "split at {split} changed the message"
921 );
922 }
923 }
924
925 #[test]
926 fn a_body_containing_a_message_is_not_a_second_message() {
927 let body = "INVITE sip:x@y SIP/2.0\r\n\r\n";
928 let text = format!(
929 "OPTIONS sip:a@b.com SIP/2.0\r\nContent-Length: {}\r\n\r\n{body}",
930 body.len()
931 );
932 let mut p = StreamParser::new(Limits::stream());
933 let messages = p.push(text.as_bytes()).expect("should parse");
934 assert_eq!(messages.len(), 1, "the body must not be reparsed");
935 assert_eq!(
936 messages.first().map(Message::body),
937 Some(&Bytes::from(body))
938 );
939 }
940
941 #[test]
942 fn a_stream_framing_error_is_permanent() {
943 let mut p = StreamParser::new(Limits::stream());
944 let text = "OPTIONS sip:a@b.com SIP/2.0\r\nContent-Length: -1\r\n\r\n";
945 assert!(p.push(text.as_bytes()).is_err());
946 assert!(p.push(MINIMAL.as_bytes()).is_err());
949 }
950}