diff --git a/src/transaction/endpoint.rs b/src/transaction/endpoint.rs index 7a508846..8731a0d5 100644 --- a/src/transaction/endpoint.rs +++ b/src/transaction/endpoint.rs @@ -398,7 +398,17 @@ impl EndpointInner { // check is the termination of an existing transaction let last_message = self.finished_transactions.get(&key).and_then(|v| v.clone()); - if let Some(last_message) = last_message { + // A CANCEL shares the INVITE's key but is answered on its own + // (RFC 3261 §9.2): 200 once the INVITE had a final response, + // else 481 below (dropped unanswered by its TU). + if req.method() == &crate::sip::Method::Cancel { + if let Some(SipMessage::Response(r)) = &last_message { + if r.status_code.kind() != crate::sip::StatusCodeKind::Provisional { + let status = crate::sip::StatusCode::OK; + return self.reply_cancel(req, status, Some(r), &connection).await; + } + } + } else if let Some(last_message) = last_message { // ACK for a completed ServerInvite transaction: absorb it silently // and clean up the waiting_ack entry. if req.method() == &crate::sip::Method::Ack { @@ -506,24 +516,8 @@ impl EndpointInner { match request.method { crate::sip::Method::Cancel => { - let resp = self.make_response( - &request, - crate::sip::StatusCode::CallTransactionDoesNotExist, - None, - ); - let resp = if let Some(ref inspector) = self.message_inspector { - inspector.before_send(resp.into(), None) - } else { - resp.into() - }; - - let dest = if !connection.is_reliable() { - self.get_destination_from_request(&request).await - } else { - None - }; - connection.send(resp, dest.as_ref()).await?; - return Ok(()); + let status = crate::sip::StatusCode::CallTransactionDoesNotExist; + return self.reply_cancel(&request, status, None, &connection).await; } crate::sip::Method::Ack => return Ok(()), _ => {} @@ -536,6 +530,28 @@ impl EndpointInner { Ok(()) } + /// Answer a CANCEL that no live transaction handles. + async fn reply_cancel( + &self, + req: &crate::sip::Request, + status: crate::sip::StatusCode, + invite_response: Option<&crate::sip::Response>, + connection: &SipConnection, + ) -> Result<()> { + let resp = self.make_cancel_response(req, status, invite_response); + let resp = if let Some(ref inspector) = self.message_inspector { + inspector.before_send(resp.into(), None) + } else { + resp.into() + }; + let dest = if !connection.is_reliable() { + self.get_destination_from_request(req).await + } else { + None + }; + connection.send(resp, dest.as_ref()).await + } + pub fn attach_transaction(&self, key: &TransactionKey, tu_sender: TransactionEventSender) { trace!(%key, "attach transaction"); self.transactions.insert(key.clone(), tu_sender); diff --git a/src/transaction/message.rs b/src/transaction/message.rs index 35ae3fbd..b7f6a4b5 100644 --- a/src/transaction/message.rs +++ b/src/transaction/message.rs @@ -253,6 +253,25 @@ impl EndpointInner { } } + /// A response to a CANCEL, with the To header (and tag) of the INVITE's + /// last response when there is one (RFC 3261 §9.2). + pub(crate) fn make_cancel_response( + &self, + cancel: &Request, + status_code: StatusCode, + invite_response: Option<&Response>, + ) -> Response { + let mut resp = self.make_response(cancel, status_code, None); + if let Some(to) = invite_response.and_then(|r| r.to_header().ok()) { + for header in resp.headers.iter_mut() { + if let Header::To(t) = header { + *t = to.clone(); + } + } + } + resp + } + // make ack from response, for ack to non-200 reponse, should pass the original invite pub fn make_ack(&self, invite: &Request, resp: &Response) -> Result { let mut headers = resp.headers.clone(); diff --git a/src/transaction/tests/mod.rs b/src/transaction/tests/mod.rs index 78d53053..f98beebf 100644 --- a/src/transaction/tests/mod.rs +++ b/src/transaction/tests/mod.rs @@ -14,6 +14,7 @@ mod test_endpoint; mod test_forked_2xx_ack; mod test_provisional_responses; mod test_server; +mod test_server_cancel; mod test_server_invite_ack; mod test_server_invite_drop; mod test_stream_reconnect; diff --git a/src/transaction/tests/test_server_cancel.rs b/src/transaction/tests/test_server_cancel.rs new file mode 100644 index 00000000..deace8be --- /dev/null +++ b/src/transaction/tests/test_server_cancel.rs @@ -0,0 +1,254 @@ +//! A CANCEL matching a server INVITE (RFC 3261 §9.2, issue #198): the UAS +//! answers the CANCEL itself, with `CSeq: n CANCEL` and the To tag of the +//! INVITE's response, 200 while the INVITE transaction exists and 481 when +//! nothing matches. Once the INVITE has a final response the CANCEL has no +//! effect on it. + +use crate::sip::StatusCode; +use crate::transaction::{transaction::Transaction, TransactionState}; +use crate::transport::{tcp_listener::TcpListenerConnection, udp::UdpConnection, TransportLayer}; +use crate::EndpointBuilder; +use std::net::SocketAddr; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::time::timeout; +use tokio_util::sync::CancellationToken; + +/// The far end: a raw UDP socket or TCP stream writing SIP text by hand. +enum Peer { + Udp(tokio::net::UdpSocket, SocketAddr), + Tcp(tokio::net::TcpStream), +} + +/// A response seen by the peer: " " and its To tag. +type Seen = (String, Option); + +impl Peer { + /// A `method` request with Via branch `branch` (the case name, or + /// `-ack` for a 2xx ACK) on the case's Call-ID. + fn request(&self, method: &str, branch: &str, to_tag: Option<&str>) -> String { + let (me, proto) = match self { + Peer::Udp(sock, _) => (sock.local_addr().unwrap(), "UDP"), + Peer::Tcp(stream) => (stream.local_addr().unwrap(), "TCP"), + }; + let call_id = branch.trim_end_matches("-ack"); + let to_tag = to_tag.map(|t| format!(";tag={t}")).unwrap_or_default(); + format!( + "{method} sip:bob@127.0.0.1 SIP/2.0\r\n\ + Via: SIP/2.0/{proto} {me};branch=z9hG4bK-{branch};rport\r\n\ + Max-Forwards: 70\r\n\ + From: ;tag=a1\r\n\ + To: {to_tag}\r\n\ + Call-ID: cancel-{call_id}\r\n\ + CSeq: 1 {method}\r\n\ + Content-Length: 0\r\n\r\n" + ) + } + + async fn send(&mut self, text: &str) { + match self { + Peer::Udp(sock, server) => { + sock.send_to(text.as_bytes(), *server).await.unwrap(); + } + Peer::Tcp(stream) => stream.write_all(text.as_bytes()).await.unwrap(), + } + } + + /// The responses received within `wait`. + async fn responses(&mut self, wait: Duration) -> Vec { + let mut text = String::new(); + let mut buf = vec![0u8; 65536]; + let deadline = tokio::time::Instant::now() + wait; + loop { + let read = match self { + Peer::Udp(sock, _) => timeout_at(deadline, sock.recv_from(&mut buf)) + .await + .map(|(n, _)| n), + Peer::Tcp(stream) => timeout_at(deadline, stream.read(&mut buf)).await, + }; + match read { + Some(n) if n > 0 => text.push_str(&String::from_utf8_lossy(&buf[..n])), + _ => break, + } + } + // Every message here has an empty body. + text.split("\r\n\r\n") + .filter(|m| !m.is_empty()) + .map(|m| { + let header = |name: &str| m.lines().find_map(|l| l.strip_prefix(name)); + let code = m.split(' ').nth(1).unwrap_or_default(); + let cseq = header("CSeq: ").unwrap_or_default(); + let tag = header("To: ").and_then(|to| to.split_once(";tag=").map(|t| t.1)); + (format!("{code} {cseq}"), tag.map(str::to_string)) + }) + .collect() + } +} + +async fn timeout_at( + deadline: tokio::time::Instant, + f: impl std::future::Future>, +) -> Option { + tokio::time::timeout_at(deadline, f).await.ok()?.ok() +} + +/// Runs `case` against a server endpoint, sending the same CANCEL twice (a +/// retransmission). Returns the responses the peer received from the first +/// CANCEL on, and the To tag of the INVITE's final response, if any, sent +/// before it. +async fn cancel_answers(case: &str, tcp: bool) -> (Vec, Option) { + let token = CancellationToken::new(); + let tl = TransportLayer::new(token.child_token()); + let mut peer = if tcp { + let port = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + tl.add_transport(TcpListenerConnection::new(addr, None).await.unwrap().into()); + tl.serve_listens().await.unwrap(); + Peer::Tcp(tokio::net::TcpStream::connect(addr).await.unwrap()) + } else { + let conn = UdpConnection::create_connection("127.0.0.1:0".parse().unwrap(), None, None) + .await + .unwrap(); + let addr = conn.get_addr().get_socketaddr().unwrap(); + tl.add_transport(conn.into()); + let sock = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); + Peer::Udp(sock, addr) + }; + let endpoint = EndpointBuilder::new() + .with_transport_layer(tl) + .with_cancel_token(token.child_token()) + .build(); + let mut incoming = endpoint.incoming_transactions().unwrap(); + let inner = endpoint.inner.clone(); + tokio::spawn(async move { inner.serve().await }); + + let mut tx: Option = None; + let mut final_tag = None; + if case != "no_invite" { + peer.send(&peer.request("INVITE", case, None)).await; + let mut t = incoming.recv().await.unwrap(); + let status = match case { + "proceeding" | "dropped_proceeding" => StatusCode::Ringing, + "accepted" | "acked" => StatusCode::OK, + _ => StatusCode::TemporarilyUnavailable, + }; + // A final reply adds a To tag (`reply_with`). + t.reply(status).await.unwrap(); + let first = peer.responses(Duration::from_millis(100)).await; + if !matches!(case, "proceeding" | "dropped_proceeding") { + final_tag = first[0].1.clone(); + assert!(final_tag.is_some(), "{case}: a final response has a To tag"); + } + if matches!(case, "acked" | "confirmed") { + let branch = if case == "acked" { + format!("{case}-ack") + } else { + case.into() + }; + let ack = peer.request("ACK", &branch, final_tag.as_deref()); + peer.send(&ack).await; + let ack = timeout(Duration::from_secs(1), t.receive()).await.unwrap(); + assert!(ack.is_some(), "{case}: the ACK must reach the TU"); + } + if case == "acked" { + // The ACK ended the transaction; only its cached 2xx remains. + assert_eq!(t.state, TransactionState::Terminated); + } else if !case.starts_with("dropped") { + tx = Some(t); + } + } + let state = tx.as_ref().map(|t| t.state.clone()); + + // The CANCEL has the INVITE's (untagged) To (RFC 3261 §9.1). + let cancel = peer.request("CANCEL", case, None); + peer.send(&cancel).await; + if case == "proceeding" { + let t = tx.as_mut().unwrap(); + let msg = timeout(Duration::from_secs(1), t.receive()).await.unwrap(); + let msg = msg.unwrap().to_string(); + assert!( + msg.starts_with("CANCEL "), + "proceeding: CANCEL must reach the TU" + ); + t.reply(StatusCode::RequestTerminated).await.unwrap(); + } + peer.send(&cancel).await; + if let Some(t) = tx.as_mut() { + // Drive the transaction for the CANCEL (and Timer G over UDP). + let delivered = timeout(Duration::from_millis(700), t.receive()).await; + assert!( + delivered.is_err(), + "{case}: a CANCEL after a final response must not reach the TU" + ); + if case != "proceeding" { + let now = Some(t.state.clone()); + assert_eq!(now, state, "{case}: the CANCEL must not change the INVITE"); + } + } + let got = peer.responses(Duration::from_millis(300)).await; + if case == "accepted" { + // The 2xx still awaits its ACK, which still ends the transaction. + let t = tx.as_mut().unwrap(); + let ack = peer.request("ACK", "accepted-ack", final_tag.as_deref()); + peer.send(&ack).await; + let ack = timeout(Duration::from_secs(1), t.receive()).await.unwrap(); + assert!( + ack.is_some(), + "accepted: the ACK must reach the TU after the CANCEL" + ); + assert_eq!(t.state, TransactionState::Terminated); + } + token.cancel(); + (got, final_tag) +} + +#[tokio::test] +async fn test_cancel_answers_by_invite_state() { + for tcp in [false, true] { + for (case, cancel_answer) in [ + ("accepted", "200"), + ("acked", "200"), + ("proceeding", "200"), + ("completed", "200"), + ("confirmed", "200"), + ("dropped_completed", "200"), + ("dropped_proceeding", "481"), + ("no_invite", "481"), + ] { + let (got, final_tag) = cancel_answers(case, tcp).await; + let row = format!("{case} (tcp: {tcp}): {got:?}"); + let (cancels, invites): (Vec<_>, Vec<_>) = + got.iter().partition(|r| r.0.ends_with(" CANCEL")); + let codes: Vec<_> = cancels.iter().map(|r| r.0.as_str()).collect(); + let expected = format!("{cancel_answer} 1 CANCEL"); + assert_eq!(codes, [expected.as_str(); 2], "{row}"); + // Responses to the INVITE: the 487 to a cancelled INVITE, the + // non-2xx final retransmitted over UDP (Timer G) and the 2xx + // retransmitted until its ACK. Never one answering a CANCEL. + let expected_invite = match case { + "proceeding" => Some("487 1 INVITE"), + "completed" if !tcp => Some("480 1 INVITE"), + "accepted" => Some("200 1 INVITE"), + _ => None, + }; + let first_invite = invites.first().map(|r| r.0.as_str()); + assert_eq!(first_invite, expected_invite, "{row}"); + assert!(invites + .iter() + .all(|r| Some(r.0.as_str()) == expected_invite)); + // The 200 to the CANCEL has the To tag of the INVITE's last + // response (§9.2): in Proceeding the 180's (none here) for the + // first CANCEL and the 487's for its retransmission. + if case == "proceeding" { + assert_eq!(cancels[0].1, None, "{row}"); + assert_eq!(cancels[1].1, invites[0].1, "{row}"); + } else if final_tag.is_some() { + assert!(cancels.iter().all(|r| r.1 == final_tag), "{row}"); + } + } + } +} diff --git a/src/transaction/transaction.rs b/src/transaction/transaction.rs index 793f1783..b53252c1 100644 --- a/src/transaction/transaction.rs +++ b/src/transaction/transaction.rs @@ -863,7 +863,10 @@ impl Transaction { TransactionState::Proceeding | TransactionState::Trying => true, // RFC 3261 Section 9.2: after a final response, CANCEL has no // effect on the original request, its responses, or session state. - TransactionState::Completed => false, + // A 2xx keeps the transaction in Accepted (RFC 6026 §7.1). + TransactionState::Completed + | TransactionState::Confirmed + | TransactionState::Accepted => false, _ => { if let Some(connection) = &self.connection { let resp = self.endpoint_inner.make_response( @@ -885,9 +888,11 @@ impl Transaction { }; if let Some(connection) = &self.connection { - let resp = self - .endpoint_inner - .make_response(&req, StatusCode::OK, None); + let resp = self.endpoint_inner.make_cancel_response( + &req, + StatusCode::OK, + self.last_response.as_ref(), + ); let resp = if let Some(ref inspector) = self.endpoint_inner.message_inspector { inspector.before_send(resp.into(), self.destination.as_ref()) } else {