sipx_call/
signalling.rs

1//! Confirmed INVITE dialogs with no SDP or media session.
2//!
3//! This is the narrow UAS primitive used by finite signalling and interoperability workloads. It
4//! is not a shortcut through the call invariants: the INVITE's 2xx is retransmitted until a valid
5//! ACK, every request is checked against both dialog tags and Call-ID, remote sequence numbers only
6//! advance, and BYE is a real transaction. What is absent is only offer/answer and the RTP socket.
7
8use std::time::Duration;
9
10use bytes::Bytes;
11use sipx_sip::build::{RequestBuilder, ResponseBuilder};
12use sipx_sip::headers::{CSeq, From as FromHeader, To};
13use sipx_sip::{HeaderName, Method, Request, Response, StatusCode};
14use sipx_transport::{Handle, Incoming, Target};
15use tokio::sync::mpsc;
16
17use crate::dialog::Dialog;
18use crate::error::{Error, Result};
19
20const T1: Duration = Duration::from_millis(500);
21const T2: Duration = Duration::from_secs(4);
22const TIMER_H: Duration = Duration::from_secs(32);
23
24/// One observable transition of an SDP-free confirmed dialog.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26#[non_exhaustive]
27pub enum SignallingEvent {
28    /// The ACK matched both dialog identity and the INVITE's sequence number.
29    Acknowledged,
30    /// A valid increasing BYE ended the dialog and received `200 OK`.
31    RemoteBye,
32    /// An ACK named the dialog but did not carry the INVITE's `CSeq` and method.
33    InvalidAck,
34    /// A request did not match both tags and the Call-ID. Non-ACK requests receive `481`.
35    InvalidDialog,
36    /// A request's `CSeq` was malformed, named another method or did not increase. It receives
37    /// `400` or `500` as appropriate.
38    InvalidCSeq,
39    /// A matched request used a method this signalling-only dialog does not implement. It receives
40    /// `405` with the narrow `Allow` set.
41    Unsupported,
42    /// Timer H expired before a valid ACK arrived.
43    AckTimedOut,
44    /// Sending a required dialog response failed because the endpoint stopped accepting it.
45    TransportFailed,
46}
47
48/// A prepared response and dialog whose fallible validation ran before the INVITE was claimed.
49pub(crate) struct Prepared {
50    response: Response,
51    dialog: Dialog,
52    target: Target,
53    invite_cseq: u32,
54}
55
56/// Build the bodyless 2xx and dialog without taking ownership of the invitation transaction.
57pub(crate) fn prepare(
58    endpoint: &Handle,
59    incoming: &Incoming,
60    tag: &str,
61    contact: Bytes,
62) -> Result<Prepared> {
63    if !valid_tag(tag) {
64        return Err(Error::InvalidDialogTag);
65    }
66    let Some(invite_cseq) = cseq(&incoming.request)
67        .filter(|value| value.method == Method::Invite)
68        .map(|value| value.sequence)
69    else {
70        return Err(Error::NoDialog);
71    };
72    let Some(dialog) = Dialog::from_request(&incoming.request, tag) else {
73        return Err(Error::NoDialog);
74    };
75    let Some(to) = incoming.request.headers.value(&HeaderName::To) else {
76        return Err(Error::NoDialog);
77    };
78    let to = format!("{};tag={tag}", String::from_utf8_lossy(&to));
79    let status = StatusCode::new(200).ok_or_else(|| Error::Rejected {
80        status: 200,
81        reason: "invalid success status".to_owned(),
82    })?;
83    let response = ResponseBuilder::to_request(&incoming.request, status, "OK")?
84        .set_header(&HeaderName::To, Bytes::from(to))?
85        .header(HeaderName::Contact, contact)?
86        .build();
87    let target =
88        crate::call::in_dialog_target(&dialog, Target::new(incoming.source, incoming.transport));
89    // `endpoint` is intentionally part of the preparation signature: the response's Contact is
90    // caller-selected, but all subsequent requests remain tied to this endpoint. Reading its
91    // advertised address here would silently override that explicit Contact.
92    // discard: the endpoint is a capability witness; preparation deliberately performs no I/O.
93    let _ = endpoint;
94    Ok(Prepared {
95        response,
96        dialog,
97        target,
98        invite_cseq,
99    })
100}
101
102/// Send a prepared response and transfer the reserved inbox into a confirmed signalling call.
103pub(crate) async fn establish(
104    endpoint: Handle,
105    incoming: Incoming,
106    requests: mpsc::Receiver<Incoming>,
107    prepared: Prepared,
108) -> Result<SignallingCall> {
109    endpoint
110        .respond(&incoming.key, prepared.response.clone())
111        .await?;
112    let now = tokio::time::Instant::now();
113    let retransmission = Retransmission {
114        key: incoming.key,
115        response: prepared.response,
116        interval: T1,
117        next: now + T1,
118        deadline: now + TIMER_H,
119    };
120    Ok(SignallingCall {
121        endpoint,
122        dialog: prepared.dialog,
123        target: prepared.target,
124        requests,
125        invite_cseq: prepared.invite_cseq,
126        acknowledged: false,
127        ended: false,
128        deferred_remote_bye: false,
129        retransmission: Some(retransmission),
130        last_request_elapsed: None,
131        last_response_status: None,
132    })
133}
134
135#[derive(Debug)]
136struct Retransmission {
137    key: sipx_sip::transaction::TransactionKey,
138    response: Response,
139    interval: Duration,
140    next: tokio::time::Instant,
141    deadline: tokio::time::Instant,
142}
143
144enum SignallingInput {
145    Request(Option<Box<Incoming>>),
146    Retransmit,
147}
148
149/// One confirmed INVITE dialog without SDP or media ownership.
150#[derive(Debug)]
151pub struct SignallingCall {
152    endpoint: Handle,
153    dialog: Dialog,
154    target: Target,
155    requests: mpsc::Receiver<Incoming>,
156    invite_cseq: u32,
157    acknowledged: bool,
158    ended: bool,
159    /// A valid BYE already answered while the peer's earlier ACK was still in flight.
160    deferred_remote_bye: bool,
161    retransmission: Option<Retransmission>,
162    last_request_elapsed: Option<Duration>,
163    last_response_status: Option<u16>,
164}
165
166impl SignallingCall {
167    /// The confirmed dialog, for explicit dispatcher-route release and observation.
168    #[must_use]
169    pub fn dialog(&self) -> &Dialog {
170        &self.dialog
171    }
172
173    /// Whether a valid ACK has stopped the INVITE final-response retransmission.
174    #[must_use]
175    pub const fn is_acknowledged(&self) -> bool {
176        self.acknowledged
177    }
178
179    /// Whether either side has ended this local dialog.
180    #[must_use]
181    pub const fn is_ended(&self) -> bool {
182        self.ended
183    }
184
185    /// Processing time from dequeuing the last routed request through its response handoff.
186    ///
187    /// This is responder-side service time, not end-to-end latency. It is `None` before any
188    /// request has been handled and for timer-only events.
189    #[must_use]
190    pub const fn last_request_elapsed(&self) -> Option<Duration> {
191        self.last_request_elapsed
192    }
193
194    /// Take the status successfully sent while producing the most recent event.
195    pub fn take_response_status(&mut self) -> Option<u16> {
196        self.last_response_status.take()
197    }
198
199    /// Drive one routed request or final-response timer outcome.
200    ///
201    /// Network-invalid input becomes a typed event and, when SIP defines one, a response. It never
202    /// panics or escapes as an internal error.
203    pub async fn next(&mut self) -> Option<SignallingEvent> {
204        loop {
205            if self.acknowledged && self.deferred_remote_bye {
206                self.deferred_remote_bye = false;
207                self.ended = true;
208                self.last_response_status = Some(200);
209                return Some(SignallingEvent::RemoteBye);
210            }
211            if self.ended {
212                return None;
213            }
214            let wake = self
215                .retransmission
216                .as_ref()
217                .map(|state| state.next.min(state.deadline));
218            let input = tokio::select! {
219                incoming = self.requests.recv() => SignallingInput::Request(incoming.map(Box::new)),
220                () = wait_until(wake), if wake.is_some() => SignallingInput::Retransmit,
221            };
222            match input {
223                SignallingInput::Request(Some(incoming)) => {
224                    let started = tokio::time::Instant::now();
225                    let event = self.handle(*incoming).await;
226                    self.last_request_elapsed = Some(started.elapsed());
227                    if let Some(event) = event {
228                        return Some(event);
229                    }
230                }
231                SignallingInput::Request(None) => {
232                    self.stop();
233                    return None;
234                }
235                SignallingInput::Retransmit => {
236                    if let Some(event) = self.retransmit().await {
237                        self.ended = true;
238                        return Some(event);
239                    }
240                }
241            }
242        }
243    }
244
245    /// Originate BYE and require a final response within `within`.
246    ///
247    /// The duration bounds a failure; success is the observed final response rather than elapsed
248    /// wall time.
249    pub async fn hang_up(&mut self, within: Duration) -> Result<u16> {
250        self.finish_retransmission();
251        let cseq = self.dialog.next_cseq();
252        let (local, remote) = self.dialog.local_and_remote();
253        let (uri, routes) = self.dialog.request_target();
254        let builder = RequestBuilder::new(Method::Bye, uri)
255            .header(HeaderName::To, Bytes::from(remote))?
256            .header(HeaderName::From, Bytes::from(local))?
257            .header(
258                HeaderName::CallId,
259                Bytes::from(self.dialog.id.call_id.clone()),
260            )?
261            .cseq(cseq, &Method::Bye)?
262            .max_forwards(70);
263        let bye = crate::call::add_routes(builder, &routes)?.build();
264        let mut responses = self.endpoint.send(bye, self.target.clone()).await?;
265        // Fixed duration bounds a failed teardown; the final response is the happens-before.
266        let response = tokio::time::timeout(within, responses.final_response())
267            .await
268            .map_err(|_| Error::SignallingTeardownTimeout(within))?
269            .ok_or(Error::SignallingTeardownTimeout(within))?;
270        self.ended = true;
271        if !response_matches_dialog(&response, &self.dialog, cseq) {
272            return Err(Error::InvalidDialogResponse);
273        }
274        let status = response.status.code();
275        if !response.status.is_success() {
276            return Err(Error::Rejected {
277                status,
278                reason: String::from_utf8_lossy(&response.reason).into_owned(),
279            });
280        }
281        Ok(status)
282    }
283
284    /// Stop owned retransmission work without sending BYE.
285    ///
286    /// Used only when another protocol outcome already ended the dialog or shutdown can no longer
287    /// reach the peer. A live established dialog should prefer [`Self::hang_up`].
288    pub fn stop(&mut self) {
289        self.ended = true;
290        self.finish_retransmission();
291    }
292
293    async fn handle(&mut self, incoming: Incoming) -> Option<SignallingEvent> {
294        self.last_response_status = None;
295        if !self.dialog.matches(&incoming.request) {
296            if incoming.request.method != Method::Ack
297                && respond(&self.endpoint, &incoming, 481, "Call Does Not Exist", None)
298                    .await
299                    .is_err()
300            {
301                return Some(SignallingEvent::TransportFailed);
302            }
303            if incoming.request.method != Method::Ack {
304                self.last_response_status = Some(481);
305            }
306            return Some(SignallingEvent::InvalidDialog);
307        }
308
309        match incoming.request.method {
310            Method::Ack => {
311                let valid = cseq(&incoming.request).is_some_and(|value| {
312                    value.method == Method::Ack && value.sequence == self.invite_cseq
313                });
314                if !valid {
315                    return Some(SignallingEvent::InvalidAck);
316                }
317                self.acknowledged = true;
318                self.finish_retransmission();
319                Some(SignallingEvent::Acknowledged)
320            }
321            Method::Bye => {
322                let Some(sequence) = cseq(&incoming.request)
323                    .filter(|value| value.method == Method::Bye)
324                    .map(|value| value.sequence)
325                else {
326                    if respond(&self.endpoint, &incoming, 400, "Bad Request", None)
327                        .await
328                        .is_err()
329                    {
330                        return Some(SignallingEvent::TransportFailed);
331                    }
332                    self.last_response_status = Some(400);
333                    return Some(SignallingEvent::InvalidCSeq);
334                };
335                if self
336                    .dialog
337                    .remote_cseq
338                    .is_some_and(|previous| sequence <= previous)
339                {
340                    if respond(
341                        &self.endpoint,
342                        &incoming,
343                        500,
344                        "Server Internal Error",
345                        None,
346                    )
347                    .await
348                    .is_err()
349                    {
350                        return Some(SignallingEvent::TransportFailed);
351                    }
352                    self.last_response_status = Some(500);
353                    return Some(SignallingEvent::InvalidCSeq);
354                }
355                self.dialog.record_remote_cseq(&incoming.request);
356                if respond(&self.endpoint, &incoming, 200, "OK", None)
357                    .await
358                    .is_err()
359                {
360                    Some(SignallingEvent::TransportFailed)
361                } else if self.acknowledged {
362                    self.finish_retransmission();
363                    self.ended = true;
364                    self.last_response_status = Some(200);
365                    Some(SignallingEvent::RemoteBye)
366                } else {
367                    self.deferred_remote_bye = true;
368                    None
369                }
370            }
371            _ => {
372                if respond(
373                    &self.endpoint,
374                    &incoming,
375                    405,
376                    "Method Not Allowed",
377                    Some((HeaderName::Allow, Bytes::from_static(b"ACK, BYE"))),
378                )
379                .await
380                .is_err()
381                {
382                    Some(SignallingEvent::TransportFailed)
383                } else {
384                    self.last_response_status = Some(405);
385                    Some(SignallingEvent::Unsupported)
386                }
387            }
388        }
389    }
390
391    fn finish_retransmission(&mut self) {
392        self.retransmission = None;
393    }
394
395    async fn retransmit(&mut self) -> Option<SignallingEvent> {
396        let now = tokio::time::Instant::now();
397        let state = self.retransmission.as_mut()?;
398        if now >= state.deadline {
399            self.retransmission = None;
400            return Some(SignallingEvent::AckTimedOut);
401        }
402        if self
403            .endpoint
404            .respond(&state.key, state.response.clone())
405            .await
406            .is_err()
407        {
408            let event = if tokio::time::Instant::now() >= state.deadline {
409                SignallingEvent::AckTimedOut
410            } else {
411                SignallingEvent::TransportFailed
412            };
413            self.retransmission = None;
414            return Some(event);
415        }
416        state.interval = state.interval.saturating_mul(2).min(T2);
417        state.next = tokio::time::Instant::now() + state.interval;
418        None
419    }
420}
421
422pub(crate) fn response_matches_dialog(response: &Response, dialog: &Dialog, sequence: u32) -> bool {
423    let unique_required = [
424        HeaderName::CallId,
425        HeaderName::From,
426        HeaderName::To,
427        HeaderName::CSeq,
428    ]
429    .iter()
430    .all(|name| response.headers.count(name) == 1);
431    if !unique_required {
432        return false;
433    }
434    let call_id_matches = response
435        .headers
436        .value(&HeaderName::CallId)
437        .is_some_and(|value| value.as_ref() == dialog.id.call_id.as_slice());
438    let from_matches = response
439        .headers
440        .typed::<FromHeader>()
441        .and_then(std::result::Result::ok)
442        .and_then(|value| value.tag().map(ToOwned::to_owned))
443        .is_some_and(|tag| tag == dialog.id.local_tag);
444    let to_matches = response
445        .headers
446        .typed::<To>()
447        .and_then(std::result::Result::ok)
448        .and_then(|value| value.tag().map(ToOwned::to_owned))
449        .is_some_and(|tag| tag == dialog.id.remote_tag);
450    let cseq_matches = response
451        .headers
452        .typed::<CSeq>()
453        .and_then(std::result::Result::ok)
454        .is_some_and(|value| value.method == Method::Bye && value.sequence == sequence);
455    call_id_matches && from_matches && to_matches && cseq_matches
456}
457
458async fn wait_until(deadline: Option<tokio::time::Instant>) {
459    match deadline {
460        Some(deadline) => tokio::time::sleep_until(deadline).await,
461        None => std::future::pending().await,
462    }
463}
464
465async fn respond(
466    endpoint: &Handle,
467    incoming: &Incoming,
468    status: u16,
469    reason: &'static str,
470    extra: Option<(HeaderName, Bytes)>,
471) -> Result<()> {
472    let status = StatusCode::new(status).ok_or_else(|| Error::Rejected {
473        status,
474        reason: "invalid response status".to_owned(),
475    })?;
476    let mut builder = ResponseBuilder::to_request(&incoming.request, status, reason)?;
477    if let Some((name, value)) = extra {
478        builder = builder.header(name, value)?;
479    }
480    endpoint.respond(&incoming.key, builder.build()).await?;
481    Ok(())
482}
483
484fn cseq(request: &Request) -> Option<CSeq> {
485    request
486        .headers
487        .typed::<CSeq>()
488        .and_then(std::result::Result::ok)
489}
490
491fn valid_tag(tag: &str) -> bool {
492    !tag.is_empty()
493        && tag.len() <= 128
494        && tag.bytes().all(|byte| {
495            byte.is_ascii_alphanumeric()
496                || matches!(
497                    byte,
498                    b'-' | b'.' | b'!' | b'%' | b'*' | b'_' | b'+' | b'`' | b'\'' | b'~'
499                )
500        })
501}
502
503#[cfg(test)]
504#[allow(
505    clippy::unwrap_used,
506    clippy::expect_used,
507    clippy::panic,
508    clippy::indexing_slicing
509)]
510mod tests {
511    use bytes::Bytes;
512    use sipx_sip::{HeaderName, Message, Uri};
513
514    use super::{response_matches_dialog, valid_tag};
515    use crate::{Dialog, DialogId, Role};
516
517    #[test]
518    fn dialog_tags_are_bounded_sip_tokens() {
519        assert!(valid_tag("t-0123456789abcdef"));
520        assert!(valid_tag("all.!%*_+`'~tokens"));
521        assert!(!valid_tag(""));
522        assert!(!valid_tag("space is not a token"));
523        assert!(!valid_tag(&"x".repeat(129)));
524    }
525
526    fn bye_response(call_id: &str, from_tag: &str, to_tag: &str, cseq: &str) -> sipx_sip::Response {
527        let wire = format!(
528            "SIP/2.0 200 OK\r\nCall-ID: {call_id}\r\n\
529             From: <sip:local@load.invalid>;tag={from_tag}\r\n\
530             To: <sip:remote@driver.invalid>;tag={to_tag}\r\n\
531             CSeq: {cseq}\r\nContent-Length: 0\r\n\r\n"
532        );
533        match sipx_sip::parse_datagram(Bytes::from(wire), &sipx_sip::Limits::datagram())
534            .expect("response parses")
535        {
536            Message::Response(response) => response,
537            Message::Request(_) => panic!("a response"),
538        }
539    }
540
541    #[test]
542    fn observed_bye_response_requires_every_dialog_identifier_and_exact_cseq() {
543        // Each mutation below leaves a syntactically valid final response and changes one identity
544        // coordinate, so transaction matching alone cannot make this assertion pass.
545        let dialog = Dialog {
546            role: Role::Callee,
547            id: DialogId {
548                call_id: b"observed@load.invalid".to_vec(),
549                local_tag: b"local".to_vec(),
550                remote_tag: b"remote".to_vec(),
551            },
552            local_uri: "<sip:local@load.invalid>".to_owned(),
553            remote_uri: "<sip:remote@driver.invalid>".to_owned(),
554            remote_target: Uri::parse(Bytes::from_static(b"sip:remote@driver.invalid"))
555                .expect("target URI"),
556            local_cseq: 2,
557            remote_cseq: Some(1),
558            route_set: Vec::new(),
559        };
560        assert!(response_matches_dialog(
561            &bye_response("observed@load.invalid", "local", "remote", "2 BYE"),
562            &dialog,
563            2
564        ));
565        for invalid in [
566            bye_response("wrong@load.invalid", "local", "remote", "2 BYE"),
567            bye_response("observed@load.invalid", "wrong", "remote", "2 BYE"),
568            bye_response("observed@load.invalid", "local", "wrong", "2 BYE"),
569            bye_response("observed@load.invalid", "local", "remote", "3 BYE"),
570            bye_response("observed@load.invalid", "local", "remote", "2 INVITE"),
571        ] {
572            assert!(!response_matches_dialog(&invalid, &dialog, 2));
573        }
574
575        let valid = bye_response("observed@load.invalid", "local", "remote", "2 BYE");
576        for name in [
577            HeaderName::CallId,
578            HeaderName::From,
579            HeaderName::To,
580            HeaderName::CSeq,
581        ] {
582            let mut duplicate = valid.clone();
583            let value = duplicate
584                .headers
585                .value(&name)
586                .expect("required response header")
587                .into_owned();
588            duplicate
589                .headers
590                .push(sipx_sip::Header::build(name, Bytes::from(value)).expect("duplicate header"));
591            assert!(!response_matches_dialog(&duplicate, &dialog, 2));
592        }
593    }
594}
sipx_call/signalling.rs

sipx_call/
signalling.rs
