netget/server/h323/
call.rs

1//! One H.225.0 call-signalling leg: the Q.931 messages of a call on its TCP connection, the
2//! H.225.0 UUIE each carries, the call reference and identifiers, tunnelled H.245, and the
3//! RTP/RTCP sockets this side advertises. Shared by the endpoint server (the called side) and
4//! the client (the calling side).
5
6use super::h225::{self, UserInformation};
7use super::h245::{self, Codec, H245Session, Media};
8use super::q931::{self, ie, msg, Q931};
9use anyhow::{bail, Context, Result};
10use serde_json::{json, Value};
11use std::net::{IpAddr, SocketAddr};
12use std::sync::atomic::{AtomicU64, Ordering};
13use std::sync::Arc;
14use tokio::io::AsyncWriteExt;
15use tokio::net::tcp::OwnedWriteHalf;
16use tokio::net::UdpSocket;
17
18/// The Q.850 causes NetGet writes.
19pub mod cause {
20    pub const UNALLOCATED_NUMBER: u8 = 1;
21    pub const NORMAL_CLEARING: u8 = 16;
22    pub const USER_BUSY: u8 = 17;
23    pub const NO_ANSWER: u8 = 19;
24    pub const CALL_REJECTED: u8 = 21;
25    pub const NORMAL_UNSPECIFIED: u8 = 31;
26    pub const TEMPORARY_FAILURE: u8 = 41;
27    pub const CONGESTION: u8 = 42;
28    pub const INVALID_CALL_REFERENCE: u8 = 81;
29    pub const INCOMPATIBLE_DESTINATION: u8 = 88;
30    pub const PROTOCOL_ERROR: u8 = 111;
31}
32
33/// The `ReleaseCompleteReason` alternatives the model may name.
34pub const RELEASE_REASONS: &[&str] = &[
35    "noBandwidth",
36    "gatekeeperResources",
37    "unreachableDestination",
38    "destinationRejection",
39    "invalidRevision",
40    "noPermission",
41    "unreachableGatekeeper",
42    "gatewayResources",
43    "badFormatAddress",
44    "adaptiveBusy",
45    "inConf",
46    "undefinedReason",
47    "facilityCallDeflection",
48    "securityDenied",
49    "calledPartyNotRegistered",
50    "callerNotRegistered",
51    "newConnectionNeeded",
52];
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55pub enum Role {
56    /// This side received the Setup.
57    Callee,
58    /// This side sent the Setup.
59    Caller,
60}
61
62#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub enum CallState {
64    /// Setup sent or received, nothing else yet.
65    Offered,
66    Proceeding,
67    Alerting,
68    Connected,
69    Released,
70}
71
72impl CallState {
73    pub fn name(self) -> &'static str {
74        match self {
75            CallState::Offered => "offered",
76            CallState::Proceeding => "proceeding",
77            CallState::Alerting => "alerting",
78            CallState::Connected => "connected",
79            CallState::Released => "released",
80        }
81    }
82}
83
84/// The RTP and RTCP sockets a call advertises. NetGet sends no media; it counts the packets
85/// the peer sends, which is what says whether the addresses it negotiated were used.
86pub struct MediaSockets {
87    pub rtp: SocketAddr,
88    pub rtcp: SocketAddr,
89    pub packets: Arc<AtomicU64>,
90    task: tokio::task::JoinHandle<()>,
91}
92
93impl MediaSockets {
94    /// Bind an RTP/RTCP pair (an even port and the next) on `ip`.
95    pub async fn bind(ip: IpAddr) -> Result<Self> {
96        let mut last_err = None;
97        for _ in 0..16 {
98            let first = UdpSocket::bind(SocketAddr::new(ip, 0)).await?;
99            let port = first.local_addr()?.port();
100            let even = port & !1;
101            let pair = if port == even {
102                UdpSocket::bind(SocketAddr::new(ip, even.wrapping_add(1)))
103                    .await
104                    .map(|second| (first, second))
105            } else {
106                drop(first);
107                match UdpSocket::bind(SocketAddr::new(ip, even)).await {
108                    Ok(a) => UdpSocket::bind(SocketAddr::new(ip, even + 1))
109                        .await
110                        .map(|b| (a, b)),
111                    Err(e) => Err(e),
112                }
113            };
114            match pair {
115                Ok((rtp, rtcp)) => {
116                    let packets = Arc::new(AtomicU64::new(0));
117                    let rtp_addr = rtp.local_addr()?;
118                    let rtcp_addr = rtcp.local_addr()?;
119                    let count = packets.clone();
120                    let task = tokio::spawn(async move {
121                        let mut a = [0u8; 2048];
122                        let mut b = [0u8; 2048];
123                        loop {
124                            tokio::select! {
125                                r = rtp.recv_from(&mut a) => if r.is_err() { break },
126                                r = rtcp.recv_from(&mut b) => if r.is_err() { break },
127                            }
128                            count.fetch_add(1, Ordering::Relaxed);
129                        }
130                    });
131                    return Ok(Self {
132                        rtp: rtp_addr,
133                        rtcp: rtcp_addr,
134                        packets,
135                        task,
136                    });
137                }
138                Err(e) => last_err = Some(e),
139            }
140        }
141        Err(last_err
142            .map(anyhow::Error::from)
143            .unwrap_or_else(|| anyhow::anyhow!("no RTP port pair")))
144        .context("binding RTP/RTCP")
145    }
146
147    pub fn received(&self) -> u64 {
148        self.packets.load(Ordering::Relaxed)
149    }
150}
151
152impl Drop for MediaSockets {
153    fn drop(&mut self) {
154        self.task.abort();
155    }
156}
157
158/// What an incoming message meant for the call.
159#[derive(Debug, Clone)]
160pub enum Incoming {
161    Proceeding,
162    Alerting,
163    Connected,
164    Released {
165        cause: Option<u8>,
166        reason: Option<String>,
167    },
168    Forwarded {
169        address: Option<SocketAddr>,
170        aliases: Vec<String>,
171    },
172    /// A Facility, Progress, Status, Information or Notify that changes nothing NetGet tracks.
173    Other(&'static str),
174    /// A second Setup on a connection whose call is already up.
175    UnexpectedSetup,
176    /// A message for another call: its call reference value, or its call reference flag
177    /// (Q.931 4.3: set by the side that did not allocate the reference), is not this call's.
178    WrongCallReference {
179        crv: u16,
180        from_destination: bool,
181    },
182}
183
184pub struct CallLeg {
185    writer: OwnedWriteHalf,
186    pub crv: u16,
187    pub role: Role,
188    pub call_id: String,
189    pub conference_id: String,
190    /// The peer said `h245Tunnelling: TRUE`; NetGet always does.
191    pub peer_tunnelling: bool,
192    pub h245: H245Session,
193    pub state: CallState,
194    pub media: MediaSockets,
195    /// Agreed through fastStart, when it was.
196    pub fast_start: Option<Media>,
197    /// Codecs this side accepts, in order of preference.
198    pub codecs: Vec<Codec>,
199    pub bytes_sent: u64,
200    /// The H.225.0 message names this side sent, in order.
201    pub sent: Vec<&'static str>,
202    /// fastStart proposals from the caller (callee side only).
203    pub offered_fast_start: Vec<Vec<u8>>,
204    /// The call reached Connect at some point (it may since have been released).
205    pub ever_connected: bool,
206    pending_h245: Vec<Vec<u8>>,
207}
208
209impl CallLeg {
210    #[allow(clippy::too_many_arguments)]
211    pub fn new(
212        writer: OwnedWriteHalf,
213        crv: u16,
214        role: Role,
215        call_id: String,
216        conference_id: String,
217        peer_tunnelling: bool,
218        media: MediaSockets,
219        codecs: Vec<Codec>,
220    ) -> Self {
221        let h245 = H245Session::new(codecs.clone(), media.rtp, media.rtcp);
222        Self {
223            writer,
224            crv,
225            role,
226            call_id,
227            conference_id,
228            peer_tunnelling,
229            h245,
230            state: CallState::Offered,
231            media,
232            fast_start: None,
233            codecs,
234            bytes_sent: 0,
235            sent: Vec::new(),
236            offered_fast_start: Vec::new(),
237            ever_connected: false,
238            pending_h245: Vec::new(),
239        }
240    }
241
242    pub fn tunnelling(&self) -> bool {
243        self.peer_tunnelling
244    }
245
246    /// Queue an H.245 message to go out tunnelled with the next H.225.0 message (or
247    /// [`CallLeg::flush_h245`]).
248    pub fn queue_h245(&mut self, m: Vec<u8>) {
249        self.pending_h245.push(m);
250    }
251
252    fn call_identifier(&self) -> Value {
253        json!({"guid": self.call_id})
254    }
255
256    /// Write one Q.931 message carrying an H.225.0 UUIE (and any H.245 waiting to go).
257    pub async fn send(
258        &mut self,
259        message_type: u8,
260        kind: &str,
261        body: Value,
262        extra_ies: Vec<(u8, Vec<u8>)>,
263    ) -> Result<()> {
264        let h245 = if self.tunnelling() {
265            std::mem::take(&mut self.pending_h245)
266        } else {
267            Vec::new()
268        };
269        let uu = h225::encode_user_information(kind, body, true, &h245)?;
270        let mut m = Q931::new(message_type, self.crv, self.role == Role::Callee);
271        for (id, contents) in extra_ies {
272            m = m.with_ie(id, contents);
273        }
274        let bytes = q931::tpkt(&m.with_user_user(&uu).encode()?)?;
275        self.writer.write_all(&bytes).await?;
276        self.bytes_sent += bytes.len() as u64;
277        self.sent.push(q931::message_name(message_type));
278        Ok(())
279    }
280
281    /// Send whatever H.245 is waiting, in a Facility with an empty body (H.225.0 v4).
282    pub async fn flush_h245(&mut self) -> Result<()> {
283        if self.pending_h245.is_empty() || !self.tunnelling() {
284            self.pending_h245.clear();
285            return Ok(());
286        }
287        self.send(msg::FACILITY, "empty", Value::Null, Vec::new())
288            .await
289    }
290
291    fn destination_body(&self) -> Value {
292        json!({
293            "protocolIdentifier": h225::PROTOCOL_ID,
294            "destinationInfo": h225::endpoint_type("terminal"),
295            "callIdentifier": self.call_identifier(),
296            "multipleCalls": false,
297            "maintainConnection": false,
298        })
299    }
300
301    pub async fn call_proceeding(&mut self) -> Result<()> {
302        if self.state != CallState::Offered {
303            bail!("CallProceeding after {}", self.state.name());
304        }
305        let body = self.destination_body();
306        self.send(msg::CALL_PROCEEDING, "callProceeding", body, Vec::new())
307            .await?;
308        self.state = CallState::Proceeding;
309        Ok(())
310    }
311
312    pub async fn alerting(&mut self) -> Result<()> {
313        if !matches!(self.state, CallState::Offered | CallState::Proceeding) {
314            bail!("Alerting after {}", self.state.name());
315        }
316        let body = self.destination_body();
317        self.send(msg::ALERTING, "alerting", body, Vec::new())
318            .await?;
319        self.state = CallState::Alerting;
320        Ok(())
321    }
322
323    /// Answer the call: Connect, with a fastStart answer when the caller offered G.711, and
324    /// this side's TerminalCapabilitySet and MasterSlaveDetermination tunnelled alongside.
325    pub async fn connect(&mut self, prefer: &[Codec], display: Option<&str>) -> Result<()> {
326        if !matches!(
327            self.state,
328            CallState::Offered | CallState::Proceeding | CallState::Alerting
329        ) {
330            bail!("Connect after {}", self.state.name());
331        }
332        let mut body = self.destination_body();
333        body["conferenceID"] = json!(self.conference_id);
334        if !self.offered_fast_start.is_empty() {
335            let order: Vec<Codec> = if prefer.is_empty() {
336                self.codecs.clone()
337            } else {
338                prefer.to_vec()
339            };
340            match h245::answer_fast_start(
341                &self.offered_fast_start,
342                &order,
343                self.media.rtp,
344                self.media.rtcp,
345            )? {
346                Some((answer, media)) => {
347                    body["fastStart"] =
348                        Value::Array(answer.iter().map(|a| json!(hex::encode(a))).collect());
349                    self.fast_start = Some(media);
350                }
351                None => body["fastConnectRefused"] = Value::Null,
352            }
353        }
354        if let Some(d) = display {
355            body["displayName"] = json!([{"name": d}]);
356        }
357        if self.tunnelling() && !self.h245.started() {
358            let opening = self.h245.start()?;
359            self.pending_h245.extend(opening);
360        }
361        let mut ies = Vec::new();
362        if let Some(d) = display {
363            ies.push((ie::DISPLAY, d.as_bytes().to_vec()));
364        }
365        self.send(msg::CONNECT, "connect", body, ies).await?;
366        self.state = CallState::Connected;
367        self.ever_connected = true;
368        Ok(())
369    }
370
371    /// Release Complete with a Q.850 cause and, optionally, an H.225.0 reason.
372    pub async fn release(&mut self, cause: u8, reason: Option<&str>) -> Result<()> {
373        if self.state == CallState::Released {
374            return Ok(());
375        }
376        if self.h245.started() && !self.h245.ended && self.tunnelling() {
377            let end = self.h245.end_session()?;
378            self.pending_h245.push(end);
379        }
380        let mut body = json!({
381            "protocolIdentifier": h225::PROTOCOL_ID,
382            "callIdentifier": self.call_identifier(),
383        });
384        if let Some(r) = reason {
385            if !RELEASE_REASONS.contains(&r) {
386                bail!("release reason {r:?} is not one of {RELEASE_REASONS:?}");
387            }
388            body["reason"] = json!({ r: null });
389        }
390        self.send(
391            msg::RELEASE_COMPLETE,
392            "releaseComplete",
393            body,
394            vec![(ie::CAUSE, q931::cause_ie(cause))],
395        )
396        .await?;
397        self.state = CallState::Released;
398        let _ = self.writer.shutdown().await;
399        Ok(())
400    }
401
402    /// Deflect the call: Facility (callForwarded) naming where to call instead, then Release
403    /// Complete (facilityCallDeflection).
404    pub async fn forward(
405        &mut self,
406        address: Option<SocketAddr>,
407        aliases: Vec<Value>,
408    ) -> Result<()> {
409        if address.is_none() && aliases.is_empty() {
410            bail!("forwarding needs an address or an alias");
411        }
412        let mut body = json!({
413            "protocolIdentifier": h225::PROTOCOL_ID,
414            "reason": {"callForwarded": null},
415            "conferenceID": self.conference_id,
416            "callIdentifier": self.call_identifier(),
417            "multipleCalls": false,
418            "maintainConnection": false,
419        });
420        if let Some(a) = address {
421            body["alternativeAddress"] = h225::transport(a);
422        }
423        if !aliases.is_empty() {
424            body["alternativeAliasAddress"] = Value::Array(aliases);
425        }
426        self.send(msg::FACILITY, "facility", body, Vec::new())
427            .await?;
428        self.release(cause::NORMAL_CLEARING, Some("facilityCallDeflection"))
429            .await
430    }
431
432    /// Handle one message from the peer.
433    pub async fn on_incoming(&mut self, m: &Q931) -> Result<Incoming> {
434        // The peer's messages carry the flag set when the peer is the called side.
435        let expected_flag = self.role == Role::Caller;
436        if m.crv != self.crv || m.from_destination != expected_flag {
437            return Ok(Incoming::WrongCallReference {
438                crv: m.crv,
439                from_destination: m.from_destination,
440            });
441        }
442        let ui = match m.user_user()? {
443            Some(raw) => Some(h225::decode_user_information(raw)?),
444            None => None,
445        };
446        if let Some(ui) = &ui {
447            if ui.h245_tunnelling() {
448                self.peer_tunnelling = true;
449            }
450            for tunnelled in ui.h245_control() {
451                match self.h245.on_message(&tunnelled) {
452                    Ok(replies) => self.pending_h245.extend(replies),
453                    Err(e) => {
454                        tracing::debug!("H.245 message not handled: {e:#}");
455                    }
456                }
457            }
458        }
459        let incoming = self.classify(m, ui.as_ref())?;
460        if !matches!(incoming, Incoming::Released { .. }) && self.state != CallState::Released {
461            self.flush_h245().await?;
462        }
463        Ok(incoming)
464    }
465
466    fn classify(&mut self, m: &Q931, ui: Option<&UserInformation>) -> Result<Incoming> {
467        Ok(match m.message_type {
468            msg::CALL_PROCEEDING => {
469                self.take_fast_start(ui);
470                if self.state == CallState::Offered {
471                    self.state = CallState::Proceeding;
472                }
473                Incoming::Proceeding
474            }
475            msg::ALERTING => {
476                self.take_fast_start(ui);
477                if matches!(self.state, CallState::Offered | CallState::Proceeding) {
478                    self.state = CallState::Alerting;
479                }
480                Incoming::Alerting
481            }
482            msg::CONNECT => {
483                self.take_fast_start(ui);
484                self.state = CallState::Connected;
485                self.ever_connected = true;
486                Incoming::Connected
487            }
488            msg::PROGRESS => {
489                self.take_fast_start(ui);
490                Incoming::Other("Progress")
491            }
492            msg::RELEASE_COMPLETE => {
493                self.state = CallState::Released;
494                Incoming::Released {
495                    cause: m.cause(),
496                    reason: ui.map(|u| h225::choice_name(u.body.get("reason"))),
497                }
498            }
499            msg::FACILITY => {
500                let reason = ui.map(|u| h225::choice_name(u.body.get("reason")));
501                if reason.as_deref() == Some("callForwarded")
502                    || reason.as_deref() == Some("routeCallToGatekeeper")
503                {
504                    let u = ui.expect("reason came from the UUIE");
505                    Incoming::Forwarded {
506                        address: u.body.get("alternativeAddress").and_then(h225::socket_addr),
507                        aliases: h225::aliases_text(u.body.get("alternativeAliasAddress")),
508                    }
509                } else {
510                    Incoming::Other("Facility")
511                }
512            }
513            msg::SETUP => Incoming::UnexpectedSetup,
514            msg::STATUS => Incoming::Other("Status"),
515            msg::STATUS_ENQUIRY => Incoming::Other("StatusEnquiry"),
516            msg::INFORMATION => Incoming::Other("Information"),
517            msg::NOTIFY => Incoming::Other("Notify"),
518            _ => Incoming::Other("Unknown"),
519        })
520    }
521
522    fn take_fast_start(&mut self, ui: Option<&UserInformation>) {
523        if self.role != Role::Caller || self.fast_start.is_some() {
524            return;
525        }
526        if let Some(ui) = ui {
527            let answer = ui.fast_start();
528            if !answer.is_empty() {
529                self.fast_start = Some(h245::read_fast_start_answer(&answer));
530            }
531        }
532    }
533
534    /// The call's media as it stands: fastStart's agreement, else H.245's.
535    pub fn media_json(&self) -> Value {
536        let m = self
537            .fast_start
538            .clone()
539            .unwrap_or_else(|| self.h245.media.clone());
540        let mut v = m.to_json();
541        v["negotiated_by"] = json!(if self.fast_start.is_some() {
542            "fastStart"
543        } else if self.h245.started() {
544            "h245"
545        } else {
546            "none"
547        });
548        v["local_rtp"] = json!(self.media.rtp.to_string());
549        v["rtp_packets_received"] = json!(self.media.received());
550        v
551    }
552
553    pub fn summary(&self) -> Value {
554        json!({
555            "crv": self.crv,
556            "call_id": self.call_id,
557            "conference_id": self.conference_id,
558            "state": self.state.name(),
559            "h245_tunnelling": self.tunnelling(),
560            "h245": self.h245.summary(),
561            "media": self.media_json(),
562            "sent": self.sent,
563        })
564    }
565}
566
567/// Read Q.931 messages from a call-signalling connection into a channel, so the call's loop
568/// can select over them alongside the model and the operator. Ends at EOF or the first error,
569/// which it sends as `Err`.
570pub async fn read_loop<R: tokio::io::AsyncRead + Unpin>(
571    mut reader: R,
572    tx: tokio::sync::mpsc::Sender<Result<Q931>>,
573) {
574    loop {
575        match q931::read_tpkt(&mut reader).await {
576            Ok(None) => break,
577            Ok(Some(payload)) if payload.is_empty() => continue,
578            Ok(Some(payload)) => {
579                let parsed = Q931::parse(&payload);
580                let stop = parsed.is_err();
581                if tx.send(parsed).await.is_err() || stop {
582                    break;
583                }
584            }
585            Err(e) => {
586                let _ = tx.send(Err(e)).await;
587                break;
588            }
589        }
590    }
591}
592
593/// The model's choice of codecs: a name or a list of names, defaulting to every codec.
594pub fn codecs_from(v: Option<&Value>) -> Result<Vec<Codec>> {
595    match v {
596        None | Some(Value::Null) => Ok(h245::ALL_CODECS.to_vec()),
597        Some(Value::String(s)) => Ok(vec![Codec::parse(s)?]),
598        Some(Value::Array(a)) => {
599            let mut out = Vec::new();
600            for c in a.iter().take(4) {
601                let s = c.as_str().context("a codec is a string")?;
602                let c = Codec::parse(s)?;
603                if !out.contains(&c) {
604                    out.push(c);
605                }
606            }
607            if out.is_empty() {
608                bail!("an empty codec list");
609            }
610            Ok(out)
611        }
612        Some(other) => bail!("codec {other} is not a name or list of names"),
613    }
614}
netget/server/h323/call.rs

netget/server/h323/
call.rs
