A CANCEL for an INVITE that already got its 2xx is answered with 481 (0.7.x), or with a copy of the INVITE's 200 #198

//! Issue F: a CANCEL for an INVITE that already got its 2xx is answered with
//! 481 (before the ACK, 0.7.x) or with a copy of the INVITE's 200 (after the
//! ACK). RFC 3261 §9.2: the UAS answers the CANCEL itself (200, or 481 if it
//! matches nothing), with `CSeq: n CANCEL`.
//!
//! Call X: CANCEL while the 2xx is not yet ACKed.
//! Call Y: CANCEL 200 ms after the ACK.
//!
//! Usage: cargo run --release
#![allow(dead_code)] // the harness at the bottom is shared by several issues
use rsipstack::dialog::dialog::DialogState;
use std::net::SocketAddr;
use std::time::{Duration, Instant};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};

/// A CANCEL with the INVITE's branch, Call-ID, From, To and CSeq number.
fn cancel_for(peer: &Peer, uas: SocketAddr, call_id: &str) -> String {
    let me = peer.addr;
    format!(
        "CANCEL sip:bob@{uas} SIP/2.0\r\n\
         Via: SIP/2.0/UDP {me};branch=z9hG4bK-{call_id}-inv;rport\r\n\
         Max-Forwards: 70\r\n\
         From: <sip:alice@{me}>;tag=a1\r\n\
         To: <sip:bob@{uas}>\r\n\
         Call-ID: {call_id}\r\n\
         CSeq: 1 CANCEL\r\n\
         Content-Length: 0\r\n\r\n"
    )
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), BoxError> {
    println!(
        "== issue F, late CANCEL | {} | 127.0.0.1 only",
        rsipstack::VERSION
    );
    let stack = Stack::start().await?;
    let mut calls = stack.uas_loop()?;
    let uas = stack.addr;
    let mut bad = 0;
    for (call_id, ack_first) in [("f-x", false), ("f-y", true)] {
        let peer = Peer::bind().await?;
        peer.send(uas, &peer.invite(uas, call_id)).await?;
        let mut call = calls.recv().await.ok_or("no call")?;
        call.spawn_handle();
        tokio::time::sleep(Duration::from_millis(20)).await;
        call.dialog.accept(None, None)?;
        let ok = peer
            .recv_until(Duration::from_secs(2), |r| r.is_response(200, "INVITE"))
            .await
            .ok_or("no 200")?;
        if ack_first {
            peer.send(uas, &peer.ack_for(uas, &ok)).await?;
            tokio::time::sleep(Duration::from_millis(200)).await;
        }
        peer.send(uas, &cancel_for(&peer, uas, call_id)).await?;
        // Replies within 300 ms (call X's next 2xx retransmission is at 500 ms).
        let mut got = Vec::new();
        let deadline = Instant::now() + Duration::from_millis(300);
        while let Some(r) = peer
            .recv(deadline.saturating_duration_since(Instant::now()))
            .await
        {
            got.push(format!(
                "{} [CSeq: {}]",
                r.first_line(),
                header(&r.text, "CSeq").unwrap_or_default()
            ));
        }
        let label = if ack_first {
            "Y (after ACK)"
        } else {
            "X (before ACK)"
        };
        println!(
            "{label}: CANCEL answered with {got:?}; dialog state: {}",
            state_name(&call.dialog.state())
        );
        let answered_200 = got
            .iter()
            .any(|l| l.starts_with("SIP/2.0 200") && l.ends_with("CANCEL]"));
        if !answered_200 || got.iter().any(|l| l.ends_with("INVITE]")) {
            bad += 1;
        }
        if !ack_first {
            peer.send(uas, &peer.ack_for(uas, &ok)).await?;
        }
    }
    verdict(
        bad > 0,
        &format!("{bad}/2 CANCELs not answered 200 to the CANCEL"),
    )
}

// ---- harness: the stack under test and a raw UDP peer, 127.0.0.1 only ----

type BoxError = Box<dyn std::error::Error + Send + Sync>;

fn verdict(reproduced: bool, what: &str) -> ! {
    let tag = if reproduced {
        "REPRODUCED"
    } else {
        "not reproduced"
    };
    println!("RESULT: {tag} - {what}");
    std::process::exit(i32::from(reproduced))
}

/// The stack under test: one UDP transport on 127.0.0.1:0, default options.
struct Stack {
    endpoint: rsipstack::transaction::Endpoint,
    dialogs: std::sync::Arc<rsipstack::dialog::dialog_layer::DialogLayer>,
    addr: SocketAddr,
    cancel: tokio_util::sync::CancellationToken,
}

impl Stack {
    async fn start() -> Result<Stack, BoxError> {
        let cancel = tokio_util::sync::CancellationToken::new();
        let tl = rsipstack::transport::TransportLayer::new(cancel.child_token());
        let udp = rsipstack::transport::udp::UdpConnection::create_connection(
            "127.0.0.1:0".parse()?,
            None,
            Some(cancel.child_token()),
        )
        .await?;
        let addr = udp.get_addr().get_socketaddr()?;
        tl.add_transport(udp.into());
        let endpoint = rsipstack::EndpointBuilder::new()
            .with_user_agent("repro")
            .with_transport_layer(tl)
            .with_cancel_token(cancel.child_token())
            .build();
        let inner = endpoint.inner.clone();
        tokio::spawn(async move { inner.serve().await });
        let dialogs = std::sync::Arc::new(rsipstack::dialog::dialog_layer::DialogLayer::new(
            endpoint.inner.clone(),
        ));
        Ok(Stack {
            endpoint,
            dialogs,
            addr,
            cancel,
        })
    }
}

impl Drop for Stack {
    fn drop(&mut self) {
        self.cancel.cancel();
    }
}

/// A new server INVITE dialog, its state receiver and its (not yet handled)
/// transaction.
struct UasCall {
    dialog: rsipstack::dialog::invite_dialog::InviteDialog,
    states: UnboundedReceiver<DialogState>,
    tx: Option<rsipstack::transaction::Transaction>,
}

impl UasCall {
    fn spawn_handle(&mut self) {
        if let Some(mut tx) = self.tx.take() {
            let mut d = self.dialog.clone();
            tokio::spawn(async move { d.handle(&mut tx).await });
        }
    }
}

impl Stack {
    /// The usual UAS loop: in-dialog requests go to `match_dialog` + `handle()`;
    /// each new INVITE gets `get_or_create_server_invite` and is handed over.
    fn uas_loop(&self) -> Result<UnboundedReceiver<UasCall>, BoxError> {
        use rsipstack::sip::prelude::HeadersExt;
        let mut incoming = self.endpoint.incoming_transactions()?;
        let dl = self.dialogs.clone();
        let (call_tx, call_rx) = unbounded_channel();
        tokio::spawn(async move {
            while let Some(mut tx) = incoming.recv().await {
                let to_tag = tx
                    .original
                    .to_header()
                    .ok()
                    .and_then(|h| h.tag().ok().flatten());
                if to_tag.is_some() {
                    if let Some(mut d) = dl.match_dialog(&tx) {
                        tokio::spawn(async move { d.handle(&mut tx).await });
                    }
                } else if tx.original.method == rsipstack::sip::Method::Invite {
                    let (state_tx, states) = unbounded_channel();
                    if let Ok(dialog) = dl.get_or_create_server_invite(&tx, state_tx, None, None) {
                        let _ = call_tx.send(UasCall {
                            dialog,
                            states,
                            tx: Some(tx),
                        });
                    }
                }
            }
        });
        Ok(call_rx)
    }
}

fn state_name(s: &DialogState) -> String {
    match s {
        DialogState::Calling(_) => "Calling".into(),
        DialogState::Trying(_) => "Trying".into(),
        DialogState::Early(..) => "Early".into(),
        DialogState::WaitAck(..) => "WaitAck".into(),
        DialogState::Confirmed(..) => "Confirmed".into(),
        DialogState::Terminated(_, r) => format!("Terminated({r:?})"),
        _ => "Other".into(),
    }
}

/// The far end: a raw UDP socket that writes and reads SIP text by hand.
struct Peer {
    sock: tokio::net::UdpSocket,
    addr: SocketAddr,
}

/// One datagram received by the peer.
struct Rx {
    text: String,
    from: SocketAddr,
}

impl Rx {
    fn first_line(&self) -> &str {
        self.text.lines().next().unwrap_or("")
    }
    fn is_request(&self, method: &str) -> bool {
        self.first_line().starts_with(&format!("{method} "))
    }
    fn is_response(&self, code: u16, cseq_method: &str) -> bool {
        self.first_line().starts_with(&format!("SIP/2.0 {code} "))
            && header(&self.text, "CSeq").is_some_and(|c| c.ends_with(cseq_method))
    }
}

/// The value of the first header `name` (full name, case-insensitive).
fn header(text: &str, name: &str) -> Option<String> {
    let prefix = format!("{}:", name.to_ascii_lowercase());
    text.split("\r\n")
        .find(|l| l.to_ascii_lowercase().starts_with(&prefix))
        .map(|l| l[prefix.len()..].trim().to_string())
}

impl Peer {
    async fn bind() -> Result<Peer, BoxError> {
        let sock = tokio::net::UdpSocket::bind("127.0.0.1:0").await?;
        let addr = sock.local_addr()?;
        Ok(Peer { sock, addr })
    }

    async fn send(&self, to: SocketAddr, text: &str) -> Result<(), BoxError> {
        self.sock.send_to(text.as_bytes(), to).await?;
        Ok(())
    }

    async fn recv(&self, wait: Duration) -> Option<Rx> {
        let mut buf = vec![0u8; 65536];
        let (n, from) = tokio::time::timeout(wait, self.sock.recv_from(&mut buf))
            .await
            .ok()?
            .ok()?;
        Some(Rx {
            text: String::from_utf8_lossy(&buf[..n]).into_owned(),
            from,
        })
    }

    /// The first datagram matching `pred` within `wait`; others are dropped.
    async fn recv_until(&self, wait: Duration, pred: impl Fn(&Rx) -> bool) -> Option<Rx> {
        let deadline = Instant::now() + wait;
        loop {
            let rx = self
                .recv(deadline.saturating_duration_since(Instant::now()))
                .await?;
            if pred(&rx) {
                return Some(rx);
            }
        }
    }
}

impl Peer {
    /// An out-of-dialog INVITE from this peer to `uas`.
    fn invite(&self, uas: SocketAddr, call_id: &str) -> String {
        let me = self.addr;
        format!(
            "INVITE sip:bob@{uas} SIP/2.0\r\n\
             Via: SIP/2.0/UDP {me};branch=z9hG4bK-{call_id}-inv;rport\r\n\
             Max-Forwards: 70\r\n\
             From: <sip:alice@{me}>;tag=a1\r\n\
             To: <sip:bob@{uas}>\r\n\
             Call-ID: {call_id}\r\n\
             CSeq: 1 INVITE\r\n\
             Contact: <sip:alice@{me}>\r\n\
             Content-Length: 0\r\n\r\n"
        )
    }

    /// The ACK for a 2xx (new branch, To copied from the 2xx).
    fn ack_for(&self, uas: SocketAddr, ok: &Rx) -> String {
        let me = self.addr;
        let to = header(&ok.text, "To").unwrap_or_default();
        let call_id = header(&ok.text, "Call-ID").unwrap_or_default();
        format!(
            "ACK sip:bob@{uas} SIP/2.0\r\n\
             Via: SIP/2.0/UDP {me};branch=z9hG4bK-{call_id}-ack;rport\r\n\
             Max-Forwards: 70\r\n\
             From: <sip:alice@{me}>;tag=a1\r\n\
             To: {to}\r\n\
             Call-ID: {call_id}\r\n\
             CSeq: 1 ACK\r\n\
             Content-Length: 0\r\n\r\n"
        )
    }
}