From 2e86cd0a3297126d81b1ab483fcae91937a79cf7 Mon Sep 17 00:00:00 2001 From: Tyson George Date: Fri, 25 Sep 2026 14:30:09 -0400 Subject: [PATCH] fix: build the Via of in-dialog requests from the transport they are sent on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DialogInner::make_request built the Via of every in-dialog request (BYE, re-INVITE, UPDATE, INFO, PRACK, ...) with get_via(None), i.e. from the endpoint's first listener. On an endpoint that binds UDP first and TCP as well, a dialog established over TCP therefore sent its BYE and re-INVITE over the TCP flow (affinity) or towards a ;transport=tcp route or remote target, but with a "Via: SIP/2.0/UDP" header. RFC 3261 §18.1.1 requires the Via to name the transport the request is sent over; peers may drop the mismatched request or answer it over UDP. The initial INVITE already picks the listener that matches the target transport (make_invite_request), and REGISTER does the same since 0c2dcc1. Do the same for in-dialog requests when the caller pins no address: take the transport of the affinity connection, else of the outbound proxy, else of the first route, else of the remote target, and use the matching listener. UDP, an unknown transport, or a target locator (which decides the transport at send time) keeps the previous behaviour. When a server dialog's request falls back to the dial-back address, its Via is rebuilt (same branch) for that address's transport. --- src/dialog/dialog.rs | 65 +++++ src/dialog/tests/mod.rs | 1 + src/dialog/tests/test_in_dialog_via.rs | 388 +++++++++++++++++++++++++ src/transaction/endpoint.rs | 2 +- 4 files changed, 455 insertions(+), 1 deletion(-) create mode 100644 src/dialog/tests/test_in_dialog_via.rs diff --git a/src/dialog/dialog.rs b/src/dialog/dialog.rs index 00ba6211..c314e169 100644 --- a/src/dialog/dialog.rs +++ b/src/dialog/dialog.rs @@ -895,10 +895,47 @@ impl DialogInner { headers: Option>, body: Option>, ) -> Result { + let addr = addr.or_else(|| self.via_addr_for_send_transport()); let via = self.endpoint_inner.get_via(addr, branch)?; self.make_request_with_vias(method, cseq, vec![via], headers, body) } + /// The listener address whose transport matches the one an in-dialog + /// request will be sent over: the reliable flow reused by affinity, else + /// the outbound proxy, else the transport of the first route, else of the + /// remote target. The Via must name the transport the request is sent on + /// (RFC 3261 §18.1.1); `make_invite_request` does the same for the + /// initial INVITE. `None` keeps the endpoint's default (first) listener, + /// also when a target locator decides the transport at send time. + fn via_addr_for_send_transport(&self) -> Option { + use crate::sip::uri::ParamsExt; + + let transport = match self.resolve_affinity_connection() { + Some(connection) => connection.get_addr().r#type, + None if self.endpoint_inner.locator.is_some() => None, + None => match self.endpoint_inner.transport_layer.outbound.as_ref() { + Some(outbound) => outbound.r#type, + None => { + let route = self.route_set.lock().first().cloned(); + match route { + Some(route) => route + .typed() + .ok() + .and_then(|route| route.uri.transport().cloned()), + None => self.remote_uri.lock().transport().cloned(), + } + } + }, + } + .filter(|t| *t != crate::sip::Transport::Udp)?; + + self.endpoint_inner + .transport_layer + .get_addrs() + .into_iter() + .find(|a| a.r#type == Some(transport)) + } + pub(super) fn make_response( &self, request: &Request, @@ -1149,6 +1186,34 @@ impl DialogInner { "no usable connection, retrying via recorded flow address" ); tx.destination = Some(fallback_addr.clone()); + // The dial-back can leave on another transport than the one + // the Via was built for; keep the Via matching it. + let transport = fallback_addr.r#type.unwrap_or_default(); + let listener = self + .endpoint_inner + .transport_layer + .get_addrs() + .into_iter() + .find(|a| a.r#type.unwrap_or_default() == transport); + let branch = tx + .original + .top_via_header() + .and_then(|via| via.typed()) + .ok() + .filter(|via| via.transport != transport) + .and_then(|via| { + via.params + .into_iter() + .find(|p| matches!(p, Param::Branch(_))) + }); + if let (Some(listener), Some(branch)) = (listener, branch) { + if let (Ok(new_via), Ok(via)) = ( + self.endpoint_inner.get_via(Some(listener), Some(branch)), + tx.original.via_header_mut(), + ) { + via.update_first_value(|_| Ok(new_via.into())).ok(); + } + } if let Err(e) = tx.send().await { warn!( id = self.id.lock().to_string(), diff --git a/src/dialog/tests/mod.rs b/src/dialog/tests/mod.rs index 3659bb23..de51169c 100644 --- a/src/dialog/tests/mod.rs +++ b/src/dialog/tests/mod.rs @@ -3,6 +3,7 @@ mod test_client_dialog; mod test_connection_affinity; mod test_dialog_layer; mod test_dialog_states; +mod test_in_dialog_via; mod test_prack; mod test_proxy_headers; mod test_refer; diff --git a/src/dialog/tests/test_in_dialog_via.rs b/src/dialog/tests/test_in_dialog_via.rs new file mode 100644 index 00000000..20d1f145 --- /dev/null +++ b/src/dialog/tests/test_in_dialog_via.rs @@ -0,0 +1,388 @@ +//! In-dialog requests sent over TCP must carry a TCP Via (RFC 3261 §18.1.1: +//! the Via transport is the transport the request is sent over). The initial +//! INVITE already picks the listener that matches the target transport +//! (`make_invite_request`); BYE / re-INVITE / UPDATE built by the dialog used +//! the endpoint's first listener, which is UDP on a UDP+TCP endpoint. +use crate::dialog::{ + dialog_layer::DialogLayer, invitation::InviteOption, invite_dialog::InviteDialog, +}; +use crate::sip::{ + prelude::{HeadersExt, ToTypedHeader}, + Method, Request, SipMessage, Transport, Uri, +}; +use crate::transport::{tcp_listener::TcpListenerConnection, udp::UdpConnection, TransportLayer}; +use crate::EndpointBuilder; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::mpsc::unbounded_channel; +use tokio_util::sync::CancellationToken; + +const PEER_TAG: &str = "peer-tag"; + +/// Read SIP messages (no bodies) off a TCP stream until one is a request +/// with `method`. +async fn recv_request(stream: &mut TcpStream, buf: &mut Vec, method: Method) -> Request { + loop { + if let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + let raw: Vec = buf.drain(..end + 4).collect(); + let text = String::from_utf8(raw).expect("non utf-8 SIP message"); + if let Ok(SipMessage::Request(req)) = SipMessage::try_from(text.as_str()) { + if req.method == method { + return req; + } + } + continue; + } + let mut chunk = [0u8; 4096]; + let n = tokio::time::timeout(Duration::from_secs(2), stream.read(&mut chunk)) + .await + .unwrap_or_else(|_| panic!("timeout waiting for {method}")) + .expect("read failed"); + assert!(n > 0, "connection closed while waiting for {method}"); + buf.extend_from_slice(&chunk[..n]); + } +} + +fn response(req: &Request, code: u16, reason: &str, extra: &str) -> String { + let to = req.to_header().unwrap().value().to_string(); + let to = if to.contains(";tag=") { + to + } else { + format!("{to};tag={PEER_TAG}") + }; + format!( + "SIP/2.0 {code} {reason}\r\n\ + Via: {}\r\n\ + From: {}\r\n\ + To: {to}\r\n\ + Call-ID: {}\r\n\ + CSeq: {}\r\n\ + {extra}\ + Content-Length: 0\r\n\r\n", + req.via_header().unwrap().value(), + req.from_header().unwrap().value(), + req.call_id_header().unwrap().value(), + req.cseq_header().unwrap().value(), + ) +} + +fn via_transport(req: &Request) -> Transport { + req.top_via_header().unwrap().typed().unwrap().transport +} + +/// A UAC endpoint listening on UDP (first) and TCP establishes a dialog with +/// a TCP peer. `record_route` makes the peer insert a Record-Route, so the +/// in-dialog requests follow the route set instead of the INVITE's flow. +async fn establish( + token: &CancellationToken, + record_route: bool, +) -> crate::Result<(InviteDialog, TcpStream, Vec)> { + let peer = TcpListener::bind("127.0.0.1:0").await?; + let peer_addr = peer.local_addr()?; + + let transport_layer = TransportLayer::new(token.child_token()); + let udp = UdpConnection::create_connection( + "127.0.0.1:0".parse().unwrap(), + None, + Some(token.child_token()), + ) + .await?; + transport_layer.add_transport(udp.into()); + let tcp_port = std::net::TcpListener::bind("127.0.0.1:0")? + .local_addr()? + .port(); + let tcp = TcpListenerConnection::new(format!("127.0.0.1:{tcp_port}").parse()?, None).await?; + transport_layer.add_transport(tcp.into()); + let endpoint = EndpointBuilder::new() + .with_user_agent("rsipstack-test") + .with_transport_layer(transport_layer) + .with_cancel_token(token.child_token()) + .build(); + let endpoint_inner = endpoint.inner.clone(); + tokio::spawn(async move { + let _ = endpoint_inner.serve().await; + }); + let dialog_layer = DialogLayer::new(endpoint.inner.clone()); + + let (state_sender, _state_receiver) = unbounded_channel(); + let invite_option = InviteOption { + caller: Uri::try_from("sip:alice@example.com")?, + callee: Uri::try_from(format!("sip:bob@{peer_addr};transport=tcp").as_str())?, + contact: Uri::try_from(format!("sip:alice@127.0.0.1:{tcp_port};transport=tcp").as_str())?, + ..Default::default() + }; + let invite = tokio::spawn(async move { + dialog_layer + .do_invite(invite_option, state_sender) + .await + .map(|(dialog, resp)| (dialog, resp.map(|r| r.status_code))) + }); + + let (mut stream, _) = tokio::time::timeout(Duration::from_secs(2), peer.accept()) + .await + .expect("timeout waiting for the INVITE connection")?; + let mut buf = Vec::new(); + let req = recv_request(&mut stream, &mut buf, Method::Invite).await; + assert_eq!( + via_transport(&req), + Transport::Tcp, + "the initial INVITE over TCP carries a TCP Via" + ); + let mut extra = format!("Contact: \r\n"); + if record_route { + extra.push_str(&format!( + "Record-Route: \r\n" + )); + } + stream + .write_all(response(&req, 200, "OK", &extra).as_bytes()) + .await?; + recv_request(&mut stream, &mut buf, Method::Ack).await; + + let (dialog, status) = tokio::time::timeout(Duration::from_secs(2), invite) + .await + .expect("do_invite timed out") + .expect("do_invite panicked")?; + assert_eq!(status, Some(crate::sip::StatusCode::OK)); + Ok((dialog, stream, buf)) +} + +async fn assert_in_dialog_requests_use_tcp_via(record_route: bool) -> crate::Result<()> { + let token = CancellationToken::new(); + let (dialog, mut stream, mut buf) = establish(&token, record_route).await?; + // A 2xx to a re-INVITE carries a Contact (RFC 3261 §12.1.1, §13.3.1.4). + let contact = format!( + "Contact: \r\n", + stream.local_addr()? + ); + + let reinvite = { + let dialog = dialog.clone(); + tokio::spawn(async move { dialog.reinvite(None, None).await }) + }; + let req = recv_request(&mut stream, &mut buf, Method::Invite).await; + assert_eq!( + via_transport(&req), + Transport::Tcp, + "a re-INVITE sent over TCP must carry a TCP Via, got: {}", + req.via_header().unwrap().value() + ); + stream + .write_all(response(&req, 200, "OK", &contact).as_bytes()) + .await?; + recv_request(&mut stream, &mut buf, Method::Ack).await; + tokio::time::timeout(Duration::from_secs(2), reinvite) + .await + .expect("reinvite timed out") + .expect("reinvite panicked")?; + + let bye = { + let dialog = dialog.clone(); + tokio::spawn(async move { dialog.bye().await }) + }; + let req = recv_request(&mut stream, &mut buf, Method::Bye).await; + assert_eq!( + via_transport(&req), + Transport::Tcp, + "a BYE sent over TCP must carry a TCP Via, got: {}", + req.via_header().unwrap().value() + ); + stream + .write_all(response(&req, 200, "OK", "").as_bytes()) + .await?; + tokio::time::timeout(Duration::from_secs(2), bye) + .await + .expect("bye timed out") + .expect("bye panicked")?; + + token.cancel(); + Ok(()) +} + +#[tokio::test] +async fn test_in_dialog_requests_over_tcp_flow_carry_tcp_via() -> crate::Result<()> { + assert_in_dialog_requests_use_tcp_via(false).await +} + +#[tokio::test] +async fn test_in_dialog_requests_over_tcp_route_carry_tcp_via() -> crate::Result<()> { + assert_in_dialog_requests_use_tcp_via(true).await +} + +/// Maps every target to a fixed UDP address, as a locator that decides the +/// transport at send time would. +struct UdpLocator(crate::transport::SipAddr); + +#[async_trait::async_trait] +impl crate::transaction::endpoint::TargetLocator for UdpLocator { + async fn locate(&self, _uri: &Uri) -> crate::Result { + Ok(self.0.clone()) + } +} + +#[tokio::test] +async fn test_in_dialog_via_is_left_to_the_locator() -> crate::Result<()> { + use crate::dialog::{dialog::DialogInner, DialogId}; + use crate::sip::headers::{CSeq, CallId, Contact, From, MaxForwards, To, Via}; + use crate::transaction::key::TransactionRole; + + let token = CancellationToken::new(); + let transport_layer = TransportLayer::new(token.child_token()); + let udp = UdpConnection::create_connection( + "127.0.0.1:0".parse().unwrap(), + None, + Some(token.child_token()), + ) + .await?; + let udp_addr = udp.get_addr().clone(); + transport_layer.add_transport(udp.into()); + let tcp = TcpListenerConnection::new("127.0.0.1:5999".parse()?, None).await?; + transport_layer.add_transport(tcp.into()); + let mut builder = EndpointBuilder::new(); + builder + .with_transport_layer(transport_layer) + .with_cancel_token(token.child_token()) + .with_target_locator(Box::new(UdpLocator(udp_addr))); + let endpoint = builder.build(); + + let initial = Request { + method: Method::Invite, + uri: Uri::try_from("sip:bob@192.0.2.10:5060;transport=tcp")?, + headers: vec![ + Via::new("SIP/2.0/UDP 127.0.0.1:5060;branch=z9hG4bKlocator").into(), + CSeq::new("1 INVITE").into(), + From::new(";tag=alice-tag").into(), + To::new(";tag=bob-tag").into(), + CallId::new("via-locator").into(), + Contact::new("").into(), + MaxForwards::new("70").into(), + ] + .into(), + version: crate::sip::Version::V2, + body: vec![], + }; + let id = DialogId::try_from((&initial, TransactionRole::Client))?; + let (tu_tx, _tu_rx) = unbounded_channel(); + let (state_tx, _state_rx) = unbounded_channel(); + let inner = DialogInner::new( + TransactionRole::Client, + id, + initial, + endpoint.inner.clone(), + state_tx, + None, + Some(Uri::try_from("sip:alice@127.0.0.1:5060")?), + tu_tx, + )?; + *inner.remote_uri.lock() = Uri::try_from("sip:bob@192.0.2.10:5060;transport=tcp")?; + + let bye = inner.make_request(Method::Bye, None, None, None, None, None)?; + assert_eq!( + via_transport(&bye), + Transport::Udp, + "with a target locator the transport is decided at send time; keep the default Via" + ); + token.cancel(); + Ok(()) +} + +#[tokio::test] +async fn test_dialback_retry_keeps_the_via_on_its_transport() -> crate::Result<()> { + use crate::dialog::{ + dialog::{DialogInner, DialogState}, + DialogId, + }; + use crate::sip::headers::{CSeq, CallId, Contact, From, MaxForwards, To, Via}; + use crate::transaction::key::TransactionRole; + + // A server dialog received over UDP whose Contact names TCP: when the TCP + // target cannot be reached, the BYE is dialed back over the recorded UDP + // source, and its Via must say UDP. + let token = CancellationToken::new(); + let transport_layer = TransportLayer::new(token.child_token()); + let udp = UdpConnection::create_connection( + "127.0.0.1:0".parse().unwrap(), + None, + Some(token.child_token()), + ) + .await?; + transport_layer.add_transport(udp.into()); + let tcp = TcpListenerConnection::new("127.0.0.1:5999".parse()?, None).await?; + transport_layer.add_transport(tcp.into()); + let endpoint = EndpointBuilder::new() + .with_transport_layer(transport_layer) + .with_cancel_token(token.child_token()) + .build(); + + let probe = tokio::net::UdpSocket::bind("127.0.0.1:0").await?; + let probe_port = probe.local_addr()?.port(); + let closed_port = std::net::TcpListener::bind("127.0.0.1:0")? + .local_addr()? + .port(); + let initial = Request { + method: Method::Invite, + uri: Uri::try_from("sip:bob@127.0.0.1:5060")?, + headers: vec![ + Via::new( + format!("SIP/2.0/UDP caller.invalid:5060;branch=z9hG4bKdialback;received=127.0.0.1;rport={probe_port}") + .as_str(), + ) + .into(), + CSeq::new("1 INVITE").into(), + From::new(";tag=alice-tag").into(), + To::new("").into(), + CallId::new("via-dialback").into(), + Contact::new(format!("").as_str()) + .into(), + MaxForwards::new("70").into(), + ] + .into(), + version: crate::sip::Version::V2, + body: vec![], + }; + let mut id = DialogId::try_from((&initial, TransactionRole::Server))?; + id.local_tag = "local-tag".into(); + let (tu_tx, _tu_rx) = unbounded_channel(); + let (state_tx, _state_rx) = unbounded_channel(); + let inner = DialogInner::new( + TransactionRole::Server, + id.clone(), + initial, + endpoint.inner.clone(), + state_tx, + None, + Some(Uri::try_from("sip:bob@127.0.0.1:5060")?), + tu_tx, + )?; + *inner.remote_uri.lock() = + Uri::try_from(format!("sip:alice@127.0.0.1:{closed_port};transport=tcp").as_str())?; + inner.transition(DialogState::Confirmed(id, Default::default()))?; + let dialog = InviteDialog::from_inner(std::sync::Arc::new(inner)); + tokio::spawn(async move { dialog.bye().await }); + + let mut buf = [0u8; 4096]; + loop { + let (len, _) = tokio::time::timeout(Duration::from_secs(3), probe.recv_from(&mut buf)) + .await + .expect("the BYE must be dialed back to the recorded UDP source")?; + let text = String::from_utf8_lossy(&buf[..len]).to_string(); + if let Ok(SipMessage::Request(req)) = SipMessage::try_from(text.as_str()) { + if req.method == Method::Bye { + assert_eq!( + via_transport(&req), + Transport::Udp, + "a BYE dialed back over UDP must carry a UDP Via, got: {}", + req.via_header().unwrap().value() + ); + assert!( + !req.via_header().unwrap().value().contains("alias"), + "a UDP Via carries no TCP alias parameter" + ); + break; + } + } + } + token.cancel(); + Ok(()) +} diff --git a/src/transaction/endpoint.rs b/src/transaction/endpoint.rs index fdf1462d..d568f10e 100644 --- a/src/transaction/endpoint.rs +++ b/src/transaction/endpoint.rs @@ -117,7 +117,7 @@ pub struct EndpointInner { #[allow(dead_code)] timer_interval: Duration, pub(super) message_inspector: Option>, - pub(super) locator: Option>, + pub(crate) locator: Option>, pub(super) transport_inspector: Option>, pub option: EndpointOption, }