sccp_protocol/server/
qos.rs

1//! IP socket marking independent of the protocol layered over the socket.
2//!
3//! Policies can be applied to either a captured signaling socket or a borrowed
4//! media socket without transferring ownership of the underlying descriptor.
5
6use std::fmt;
7use std::io;
8use std::net::{IpAddr, SocketAddr};
9
10#[cfg(unix)]
11use std::os::fd::AsFd;
12#[cfg(windows)]
13use std::os::windows::io::AsSocket;
14
15use socket2::{SockRef, Socket};
16
17use crate::types::SignalingQos;
18
19/// A socket option that could not be applied.
20#[derive(Clone, Copy, Debug, Eq, PartialEq)]
21pub enum SocketQosMark {
22    Dscp,
23    SocketPriority,
24}
25
26/// One failed marking operation.
27#[derive(Debug)]
28pub struct SocketQosFailure {
29    mark: SocketQosMark,
30    source: io::Error,
31}
32
33impl SocketQosFailure {
34    pub const fn mark(&self) -> SocketQosMark {
35        self.mark
36    }
37
38    pub const fn source(&self) -> &io::Error {
39        &self.source
40    }
41}
42
43impl fmt::Display for SocketQosFailure {
44    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
45        let mark = match self.mark {
46            SocketQosMark::Dscp => "DSCP",
47            SocketQosMark::SocketPriority => "socket priority",
48        };
49        write!(formatter, "unable to apply socket {mark}: {}", self.source)
50    }
51}
52
53/// Independent results from applying every supported socket mark.
54#[derive(Debug, Default)]
55pub struct SocketQosReport {
56    failures: Vec<SocketQosFailure>,
57}
58
59impl SocketQosReport {
60    pub fn failed(mark: SocketQosMark, source: io::Error) -> Self {
61        Self {
62            failures: vec![SocketQosFailure { mark, source }],
63        }
64    }
65
66    pub fn is_complete(&self) -> bool {
67        self.failures.is_empty()
68    }
69
70    pub fn failures(&self) -> impl ExactSizeIterator<Item = &SocketQosFailure> {
71        self.failures.iter()
72    }
73
74    /// Consume the report and return each independent marking failure.
75    pub fn into_failures(self) -> impl ExactSizeIterator<Item = SocketQosFailure> {
76        self.failures.into_iter()
77    }
78
79    fn push(&mut self, mark: SocketQosMark, source: io::Error) {
80        self.failures.push(SocketQosFailure { mark, source });
81    }
82}
83
84/// Typed socket-marking values accepted by the shared platform adapter.
85pub trait SocketQosPolicy: Copy {
86    /// Six-bit differentiated-services code point.
87    fn dscp(self) -> u8;
88
89    /// Three-bit class-of-service priority.
90    fn cos(self) -> u8;
91}
92
93impl SocketQosPolicy for SignalingQos {
94    fn dscp(self) -> u8 {
95        self.dscp
96    }
97
98    fn cos(self) -> u8 {
99        self.cos
100    }
101}
102
103/// Session-owned capability for changing the underlying signaling socket.
104///
105/// Implementations attempt DSCP and socket priority independently. A partial
106/// failure is returned in the report and must not terminate the station
107/// session.
108pub trait StationSocketQos: fmt::Debug + Send + Sync {
109    fn apply(&self, qos: SignalingQos) -> SocketQosReport;
110}
111
112/// Clone of a TCP socket retained independently of its clear or TLS stream.
113#[derive(Debug)]
114pub struct SignalingSocket {
115    socket: Socket,
116    local: SocketAddr,
117}
118
119impl SignalingSocket {
120    #[cfg(unix)]
121    pub fn capture<S>(socket: &S, local: SocketAddr) -> io::Result<Self>
122    where
123        S: AsFd,
124    {
125        Ok(Self {
126            socket: SockRef::from(socket).try_clone()?,
127            local,
128        })
129    }
130
131    #[cfg(windows)]
132    pub fn capture<S>(socket: &S, local: SocketAddr) -> io::Result<Self>
133    where
134        S: AsSocket,
135    {
136        Ok(Self {
137            socket: SockRef::from(socket).try_clone()?,
138            local,
139        })
140    }
141}
142
143impl StationSocketQos for SignalingSocket {
144    fn apply(&self, qos: SignalingQos) -> SocketQosReport {
145        apply_socket_marks(&self.socket, self.local.ip(), qos)
146    }
147}
148
149/// Apply typed marking policy to a borrowed socket without taking ownership.
150///
151/// This is suitable for sockets owned by a larger media or signaling object.
152/// The descriptor remains open and owned by its original RAII guard.
153#[cfg(unix)]
154pub fn apply_socket_qos<S, Q>(socket: &S, qos: Q) -> io::Result<SocketQosReport>
155where
156    S: AsFd,
157    Q: SocketQosPolicy,
158{
159    apply_borrowed_socket_qos(SockRef::from(socket), qos)
160}
161
162/// Windows form of [`apply_socket_qos`].
163#[cfg(windows)]
164pub fn apply_socket_qos<S, Q>(socket: &S, qos: Q) -> io::Result<SocketQosReport>
165where
166    S: AsSocket,
167    Q: SocketQosPolicy,
168{
169    apply_borrowed_socket_qos(SockRef::from(socket), qos)
170}
171
172fn apply_borrowed_socket_qos(
173    socket: SockRef<'_>,
174    qos: impl SocketQosPolicy,
175) -> io::Result<SocketQosReport> {
176    let local = socket.local_addr()?.as_socket().ok_or_else(|| {
177        io::Error::new(
178            io::ErrorKind::Unsupported,
179            "socket address family does not support IP QoS",
180        )
181    })?;
182    Ok(apply_socket_marks(&socket, local.ip(), qos))
183}
184
185fn apply_socket_marks(
186    socket: &Socket,
187    local_address: IpAddr,
188    qos: impl SocketQosPolicy,
189) -> SocketQosReport {
190    let mut report = SocketQosReport::default();
191    let dscp = qos.dscp();
192    let dscp_result = if dscp <= 63 {
193        apply_dscp(socket, local_address, dscp)
194    } else {
195        Err(invalid_qos_value("DSCP", dscp, 63))
196    };
197    if let Err(source) = dscp_result {
198        report.push(SocketQosMark::Dscp, source);
199    }
200    let cos = qos.cos();
201    let priority_result = if cos <= 7 {
202        apply_socket_priority(socket, cos)
203    } else {
204        Err(invalid_qos_value("COS", cos, 7))
205    };
206    if let Err(source) = priority_result {
207        report.push(SocketQosMark::SocketPriority, source);
208    }
209    report
210}
211
212fn invalid_qos_value(name: &str, value: u8, maximum: u8) -> io::Error {
213    io::Error::new(
214        io::ErrorKind::InvalidInput,
215        format!("{name} {value} exceeds {maximum}"),
216    )
217}
218
219fn apply_dscp(socket: &Socket, address: IpAddr, dscp: u8) -> io::Result<()> {
220    let traffic_class = u32::from(dscp) << 2;
221    match address {
222        IpAddr::V4(_) => apply_ipv4_traffic_class(socket, traffic_class),
223        IpAddr::V6(_) => apply_ipv6_traffic_class(socket, traffic_class),
224    }
225}
226
227#[cfg(not(any(
228    target_os = "fuchsia",
229    target_os = "redox",
230    target_os = "solaris",
231    target_os = "haiku",
232    target_os = "wasi",
233)))]
234fn apply_ipv4_traffic_class(socket: &Socket, traffic_class: u32) -> io::Result<()> {
235    socket.set_tos_v4(traffic_class)
236}
237
238#[cfg(any(
239    target_os = "fuchsia",
240    target_os = "redox",
241    target_os = "solaris",
242    target_os = "haiku",
243    target_os = "wasi",
244))]
245fn apply_ipv4_traffic_class(_socket: &Socket, _traffic_class: u32) -> io::Result<()> {
246    Err(io::Error::new(
247        io::ErrorKind::Unsupported,
248        "IPv4 DSCP marking is unavailable on this platform",
249    ))
250}
251
252#[cfg(any(
253    target_os = "android",
254    target_os = "dragonfly",
255    target_os = "freebsd",
256    target_os = "fuchsia",
257    target_os = "linux",
258    target_os = "macos",
259    target_os = "netbsd",
260    target_os = "openbsd",
261    target_os = "cygwin",
262    target_os = "illumos",
263))]
264fn apply_ipv6_traffic_class(socket: &Socket, traffic_class: u32) -> io::Result<()> {
265    socket.set_tclass_v6(traffic_class)
266}
267
268#[cfg(not(any(
269    target_os = "android",
270    target_os = "dragonfly",
271    target_os = "freebsd",
272    target_os = "fuchsia",
273    target_os = "linux",
274    target_os = "macos",
275    target_os = "netbsd",
276    target_os = "openbsd",
277    target_os = "cygwin",
278    target_os = "illumos",
279)))]
280fn apply_ipv6_traffic_class(_socket: &Socket, _traffic_class: u32) -> io::Result<()> {
281    Err(io::Error::new(
282        io::ErrorKind::Unsupported,
283        "IPv6 DSCP marking is unavailable on this platform",
284    ))
285}
286
287#[cfg(any(target_os = "linux", target_os = "android", target_os = "fuchsia"))]
288fn apply_socket_priority(socket: &Socket, cos: u8) -> io::Result<()> {
289    socket.set_priority(u32::from(cos))
290}
291
292#[cfg(not(any(target_os = "linux", target_os = "android", target_os = "fuchsia")))]
293fn apply_socket_priority(_socket: &Socket, _cos: u8) -> io::Result<()> {
294    Err(io::Error::new(
295        io::ErrorKind::Unsupported,
296        "socket priority marking is unavailable on this platform",
297    ))
298}
299
300#[cfg(test)]
301mod tests {
302    use std::net::{Ipv4Addr, TcpListener, UdpSocket};
303
304    use super::*;
305
306    #[test]
307    fn dscp_is_shifted_into_the_ipv4_traffic_class() {
308        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
309        let socket = SignalingSocket::capture(&listener, listener.local_addr().unwrap()).unwrap();
310
311        let report = socket.apply(SignalingQos::new(26, 0));
312
313        assert!(
314            report
315                .failures()
316                .all(|failure| failure.mark() == SocketQosMark::SocketPriority)
317        );
318        assert_eq!(socket.socket.tos_v4().unwrap(), 26 << 2);
319    }
320
321    #[test]
322    fn borrowed_udp_sockets_are_marked_independently_without_ownership_transfer() {
323        let rtp = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
324        let rtcp = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
325
326        let rtp_report = apply_socket_qos(&rtp, SignalingQos::new(46, 6)).unwrap();
327        let rtcp_report = apply_socket_qos(&rtcp, SignalingQos::new(46, 6)).unwrap();
328
329        assert!(rtp_report.failures().all(platform_priority_failure));
330        assert!(rtcp_report.failures().all(platform_priority_failure));
331        assert_eq!(SockRef::from(&rtp).tos_v4().unwrap(), 46 << 2);
332        assert_eq!(SockRef::from(&rtcp).tos_v4().unwrap(), 46 << 2);
333        assert!(rtp.local_addr().is_ok());
334        assert!(rtcp.local_addr().is_ok());
335    }
336
337    #[test]
338    fn public_adapter_rejects_out_of_range_marks_independently() {
339        let socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
340
341        let report = apply_socket_qos(&socket, SignalingQos::new(64, 8)).unwrap();
342        let failures = report.failures().collect::<Vec<_>>();
343
344        assert_eq!(failures.len(), 2);
345        assert_eq!(failures[0].mark(), SocketQosMark::Dscp);
346        assert_eq!(failures[0].source().kind(), io::ErrorKind::InvalidInput);
347        assert_eq!(failures[1].mark(), SocketQosMark::SocketPriority);
348        assert_eq!(failures[1].source().kind(), io::ErrorKind::InvalidInput);
349        assert_eq!(SockRef::from(&socket).tos_v4().unwrap(), 0);
350        assert!(socket.local_addr().is_ok());
351    }
352
353    fn platform_priority_failure(failure: &SocketQosFailure) -> bool {
354        failure.mark() == SocketQosMark::SocketPriority
355            && failure.source().kind() == io::ErrorKind::Unsupported
356    }
357
358    #[cfg(any(target_os = "linux", target_os = "android", target_os = "fuchsia"))]
359    #[test]
360    fn cos_is_applied_as_socket_priority_when_supported() {
361        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
362        let socket = SignalingSocket::capture(&listener, listener.local_addr().unwrap()).unwrap();
363
364        assert!(socket.apply(SignalingQos::new(0, 5)).is_complete());
365        assert_eq!(socket.socket.priority().unwrap(), 5);
366    }
367
368    #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "fuchsia")))]
369    #[test]
370    fn unsupported_socket_priority_is_reported_after_dscp_succeeds() {
371        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
372        let socket = SignalingSocket::capture(&listener, listener.local_addr().unwrap()).unwrap();
373
374        let report = socket.apply(SignalingQos::new(46, 6));
375
376        let failure = report.failures().next().unwrap();
377        assert_eq!(failure.mark(), SocketQosMark::SocketPriority);
378        assert_eq!(failure.source().kind(), io::ErrorKind::Unsupported);
379        assert_eq!(socket.socket.tos_v4().unwrap(), 46 << 2);
380    }
381}
sccp_protocol/server/qos.rs

sccp_protocol/server/
qos.rs
