Merge pull request #30 from slyoldfox/main

Consume the full stream to make sure SIP messages with bodies are fully parsed

Author: shenjinti

src/dialog/invitation.rs

@@ -247,7 +247,7 @@ impl DialogLayer {
         };
         let recipient = to.uri.clone();
 
-        let form = rsip::typed::From {
+        let from = rsip::typed::From {
             display_name: None,
             uri: opt.caller.clone(),
             params: vec![],
@@ -257,7 +257,7 @@ impl DialogLayer {
         let via = self.endpoint.get_via(None, None)?;
         let mut request =
             self.endpoint
-                .make_request(rsip::Method::Invite, recipient, via, form, to, last_seq);
+                .make_request(rsip::Method::Invite, recipient, via, from, to, last_seq);
 
         let contact = rsip::typed::Contact {
             display_name: None,

src/dialog/registration.rs

@@ -358,7 +358,7 @@ impl Registration {
             });
         }
 
-        let form = rsip::typed::From {
+        let from = rsip::typed::From {
             display_name: None,
             uri: to.uri.clone(),
             params: vec![],
@@ -396,7 +396,7 @@ impl Registration {
             rsip::Method::Register,
             server,
             via,
-            form,
+            from,
             to,
             self.last_seq,
         );

src/transport/stream.rs

@@ -62,31 +62,37 @@ impl Decoder for SipCodec {
             return Ok(Some(SipCodecType::KeepaliveResponse));
         }
 
-        if let Some(end_pos) = src
-            .windows(KEEPALIVE_REQUEST.len())
-            .position(|window| window == KEEPALIVE_REQUEST)
-        {
-            let msg_end = end_pos + KEEPALIVE_REQUEST.len();
-            let msg_data = &src[..msg_end];
-            match SipMessage::try_from(msg_data) {
-                Ok(msg) => {
-                    src.advance(msg_end);
-                    Ok(Some(SipCodecType::Message(msg)))
-                }
-                Err(e) => {
-                    src.advance(msg_end);
-                    Err(crate::Error::Error(format!(
-                        "Failed to parse SIP message: {}",
-                        e
-                    )))
+        if let Some(headers_end) = src.windows(4).position(|w| w == b"\r\n\r\n") {
+            let headers = &src[..headers_end + 4]; // include CRLFCRLF
+
+            // --- Inline get_content_length logic ---
+            let headers_str = std::str::from_utf8(headers)
+                .map_err(|e| crate::Error::Error(format!("Invalid UTF-8 in headers: {}", e)))?;
+            let mut content_length = 0;
+            for line in headers_str.lines() {
+                if let Some(rest) = line.strip_prefix("Content-Length:") {
+                    content_length = rest
+                        .trim()
+                        .parse::<usize>()
+                        .map_err(|e| crate::Error::Error(format!("Invalid Content-Length: {}", e)))?;
+                    break;
                 }
             }
-        } else {
-            if src.len() > MAX_SIP_MESSAGE_SIZE {
-                return Err(crate::Error::Error("SIP message too large".to_string()));
+            // --- End inline Content-Length logic ---
+
+            let total_len = headers_end + 4 + content_length;
+
+            if src.len() >= total_len {
+                let msg_data = src.split_to(total_len); // consume full message
+                let msg = SipMessage::try_from(&msg_data[..])?;
+                return Ok(Some(SipCodecType::Message(msg)));
             }
-            Ok(None)
         }
+
+        if src.len() > MAX_SIP_MESSAGE_SIZE {
+            return Err(crate::Error::Error("SIP message too large".to_string()));
+        }
+        Ok(None)
     }
 }
 
@@ -194,7 +200,8 @@ where
                             }
                             Err(e) => {
                                 warn!("Error decoding message from {}: {:?}", remote_addr, e);
-                                // Continue processing despite decode errors
+                                buffer.clear(); // reset buffer so we don't loop forever on bad data
+                                break;          // or `return Err(e.into())` to close the connection
                             }
                         }
                     }

src/transport/tests/test_stream_encoding.rs

@@ -262,3 +262,69 @@ fn test_sip_codec_size_limit() {
     let result = codec.decode(&mut buffer);
     assert!(result.is_err(), "Should error on oversized message");
 }
+
+/// Test SipCodec handling of multiple messages in one buffer (TCP packet coalescing)
+#[test]
+fn test_sip_codec_multiple_messages_with_bodies() {
+    let mut codec = SipCodec::new();
+    let mut buffer = BytesMut::new();
+
+    let message1 = "REGISTER sip:example.com SIP/2.0\r\n\
+                    Via: SIP/2.0/TCP 127.0.0.1:5060;branch=z9hG4bK-test1\r\n\
+                    From: <sip:[email protected]>;tag=test1\r\n\
+                    To: <sip:[email protected]>\r\n\
+                    Call-ID: test-call-id-1\r\n\
+                    CSeq: 1 REGISTER\r\n\
+                    Contact: <sip:[email protected]:5060>\r\n\
+                    Max-Forwards: 70\r\n\
+                    Content-Length: 11\r\n\r\nHello world";
+
+    let message2 = "REGISTER sip:example.com SIP/2.0\r\n\
+                    Via: SIP/2.0/TCP 127.0.0.1:5060;branch=z9hG4bK-test2\r\n\
+                    From: <sip:[email protected]>;tag=test2\r\n\
+                    To: <sip:[email protected]>\r\n\
+                    Call-ID: test-call-id-2\r\n\
+                    CSeq: 1 REGISTER\r\n\
+                    Contact: <sip:[email protected]:5060>\r\n\
+                    Max-Forwards: 70\r\n\
+                    Content-Length: 6\r\n\r\nfoobar";
+
+    // Add both messages to buffer
+    buffer.extend_from_slice(message1.as_bytes());
+    buffer.extend_from_slice(message2.as_bytes());
+
+    // First decode
+    let result1 = codec
+        .decode(&mut buffer)
+        .expect("first decode should succeed");
+    assert!(result1.is_some(), "Should decode first message");
+
+    let msg1 = result1.unwrap();
+    match msg1 {
+        crate::transport::stream::SipCodecType::Message(SipMessage::Request(req)) => {
+            assert_eq!(req.call_id_header().unwrap().value(), "test-call-id-1");
+        }
+        _ => panic!("Expected request message"),
+    }
+
+    // Second decode
+    let result2 = codec
+        .decode(&mut buffer)
+        .expect("second decode should succeed");
+    assert!(result2.is_some(), "Should decode second message");
+
+    let msg2 = result2.unwrap();
+    match msg2 {
+        crate::transport::stream::SipCodecType::Message(SipMessage::Request(req)) => {
+            assert_eq!(req.call_id_header().unwrap().value(), "test-call-id-2");
+        }
+        _ => panic!("Expected request message"),
+    }
+
+    // Buffer should be empty after consuming both messages
+    assert_eq!(
+        buffer.len(),
+        0,
+        "Buffer should be empty after consuming all messages"
+    );
+}