fidl_fuchsia_net_sockets_ext/
lib.rs

1// Copyright 2025 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Extensions for the fuchsia.sockets FIDL library.
6
7#![warn(
8    missing_docs,
9    unreachable_patterns,
10    clippy::useless_conversion,
11    clippy::redundant_clone,
12    clippy::precedence
13)]
14
15use fidl_fuchsia_net as fnet;
16use fidl_fuchsia_net_ext::{IntoExt, Marks};
17use fidl_fuchsia_net_matchers as fnet_matchers;
18use fidl_fuchsia_net_matchers_ext as fnet_matchers_ext;
19use fidl_fuchsia_net_sockets as fnet_sockets;
20use fidl_fuchsia_net_tcp as fnet_tcp;
21use fidl_fuchsia_net_udp as fnet_udp;
22use futures::{Stream, TryStreamExt as _};
23use net_types::ip::{self, GenericOverIp, Ip, IpInvariant, Ipv4, Ipv6};
24use thiserror::Error;
25
26/// An extension type for [`fnet_sockets::IpSocketMatcher`].
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub enum IpSocketMatcher {
29    /// Matches against the IP version of the socket.
30    Family(ip::IpVersion),
31    /// Matches against the source address of the socket.
32    SrcAddr(fnet_matchers_ext::BoundAddress),
33    /// Matches against the destination address of the socket.
34    DstAddr(fnet_matchers_ext::BoundAddress),
35    /// Matches against transport protocol fields of the socket.
36    Proto(fnet_matchers_ext::SocketTransportProtocol),
37    /// Matches against the (bound, i.e. SO_BINDTODEVICE) interface of the
38    /// socket.
39    BoundInterface(fnet_matchers_ext::BoundInterface),
40    /// Matches against the cookie of the socket (i.e. SO_COOKIE)
41    Cookie(fnet_matchers::SocketCookie),
42    /// Matches against one mark of the socket.
43    Mark(fnet_matchers_ext::MarkInDomain),
44}
45
46/// Errors returned by the conversion from [`fnet_sockets::IpSocketMatcher`]
47/// to [`IpSocketMatcher`].
48#[derive(Debug, PartialEq, Error)]
49pub enum IpSocketMatcherError {
50    /// A union type was unknown.
51    #[error("got unexpected union variant: {0}")]
52    UnknownUnionVariant(u64),
53    /// An error was encountered when converting one of the address matchers.
54    #[error("address matcher conversion failure: {0}")]
55    Address(fnet_matchers_ext::BoundAddressError),
56    /// An error was encountered when converting the transport protocol
57    /// matcher.
58    #[error("protocol matcher conversion failure: {0}")]
59    TransportProtocol(fnet_matchers_ext::SocketTransportProtocolError),
60    /// An error was encountered while converting the interface matcher.
61    #[error("bound interface matcher conversion failure: {0}")]
62    BoundInterface(fnet_matchers_ext::BoundInterfaceError),
63    /// An error was encountered when converting one of the mark matchers.
64    #[error("mark matcher conversion failure: {0}")]
65    Mark(fnet_matchers_ext::MarkInDomainError),
66}
67
68impl TryFrom<fnet_sockets::IpSocketMatcher> for IpSocketMatcher {
69    type Error = IpSocketMatcherError;
70
71    fn try_from(matcher: fnet_sockets::IpSocketMatcher) -> Result<Self, Self::Error> {
72        match matcher {
73            fnet_sockets::IpSocketMatcher::Family(ip_version) => {
74                Ok(Self::Family(ip_version.into_ext()))
75            }
76            fnet_sockets::IpSocketMatcher::SrcAddr(addr) => {
77                Ok(Self::SrcAddr(addr.try_into().map_err(|e| IpSocketMatcherError::Address(e))?))
78            }
79            fnet_sockets::IpSocketMatcher::DstAddr(addr) => {
80                Ok(Self::DstAddr(addr.try_into().map_err(|e| IpSocketMatcherError::Address(e))?))
81            }
82            fnet_sockets::IpSocketMatcher::Proto(proto) => Ok(Self::Proto(
83                proto.try_into().map_err(|e| IpSocketMatcherError::TransportProtocol(e))?,
84            )),
85            fnet_sockets::IpSocketMatcher::BoundInterface(bound_interface) => {
86                Ok(Self::BoundInterface(
87                    bound_interface
88                        .try_into()
89                        .map_err(|e| IpSocketMatcherError::BoundInterface(e))?,
90                ))
91            }
92            fnet_sockets::IpSocketMatcher::Cookie(cookie) => Ok(Self::Cookie(cookie)),
93            fnet_sockets::IpSocketMatcher::Mark(mark) => {
94                Ok(Self::Mark(mark.try_into().map_err(|e| IpSocketMatcherError::Mark(e))?))
95            }
96            fnet_sockets::IpSocketMatcher::__SourceBreaking { unknown_ordinal } => {
97                Err(IpSocketMatcherError::UnknownUnionVariant(unknown_ordinal))
98            }
99        }
100    }
101}
102
103impl From<IpSocketMatcher> for fnet_sockets::IpSocketMatcher {
104    fn from(value: IpSocketMatcher) -> Self {
105        match value {
106            IpSocketMatcher::Family(ip_version) => {
107                fnet_sockets::IpSocketMatcher::Family(ip_version.into_ext())
108            }
109            IpSocketMatcher::SrcAddr(address) => {
110                fnet_sockets::IpSocketMatcher::SrcAddr(address.into())
111            }
112            IpSocketMatcher::DstAddr(address) => {
113                fnet_sockets::IpSocketMatcher::DstAddr(address.into())
114            }
115            IpSocketMatcher::Proto(socket_transport_protocol) => {
116                fnet_sockets::IpSocketMatcher::Proto(socket_transport_protocol.into())
117            }
118            IpSocketMatcher::BoundInterface(mark) => {
119                fnet_sockets::IpSocketMatcher::BoundInterface(mark.into())
120            }
121            IpSocketMatcher::Cookie(socket_cookie) => {
122                fnet_sockets::IpSocketMatcher::Cookie(socket_cookie)
123            }
124            IpSocketMatcher::Mark(mark) => fnet_sockets::IpSocketMatcher::Mark(mark.into()),
125        }
126    }
127}
128
129/// Extension type for [`fnet_sockets::IpSocketState`].
130#[derive(Debug, PartialEq, Eq, Clone)]
131pub enum IpSocketState {
132    /// IPv4 socket state.
133    V4(IpSocketStateSpecific<Ipv4>),
134    /// IPv6 socket state.
135    V6(IpSocketStateSpecific<Ipv6>),
136}
137
138/// Error type for [`IpSocketState`] conversion.
139#[derive(Debug, Error, PartialEq)]
140pub enum IpSocketStateError {
141    /// Missing a required field.
142    #[error("missing field: {0}")]
143    MissingField(&'static str),
144    /// The socket address version does not match the expected version.
145    #[error("version mismatch")]
146    VersionMismatch,
147    /// The transport state is invalid.
148    #[error("transport state error: {0}")]
149    Transport(IpSocketTransportStateError),
150}
151
152impl TryFrom<fnet_sockets::IpSocketState> for IpSocketState {
153    type Error = IpSocketStateError;
154
155    fn try_from(value: fnet_sockets::IpSocketState) -> Result<Self, Self::Error> {
156        fn convert_address<I: Ip>(addr: fnet::IpAddress) -> Result<I::Addr, IpSocketStateError> {
157            I::map_ip::<_, Option<I::Addr>>(
158                IpInvariant(addr.into_ext()),
159                |IpInvariant(addr)| match addr {
160                    net_types::ip::IpAddr::V4(addr) => Some(addr),
161                    _ => None,
162                },
163                |IpInvariant(addr)| match addr {
164                    net_types::ip::IpAddr::V6(addr) => Some(addr),
165                    _ => None,
166                },
167            )
168            .ok_or(IpSocketStateError::VersionMismatch)
169        }
170
171        fn to_ip_socket_specific<I: Ip>(
172            src_addr: Option<fnet::IpAddress>,
173            dst_addr: Option<fnet::IpAddress>,
174            cookie: u64,
175            marks: fnet::Marks,
176            transport: fnet_sockets::IpSocketTransportState,
177        ) -> Result<IpSocketStateSpecific<I>, IpSocketStateError> {
178            let src_addr: Option<I::Addr> = src_addr.map(convert_address::<I>).transpose()?;
179            let dst_addr: Option<I::Addr> = dst_addr.map(convert_address::<I>).transpose()?;
180
181            Ok(IpSocketStateSpecific {
182                src_addr,
183                dst_addr,
184                cookie,
185                marks: marks.into(),
186                transport: transport.try_into().map_err(IpSocketStateError::Transport)?,
187            })
188        }
189
190        let fnet_sockets::IpSocketState {
191            family,
192            src_addr,
193            dst_addr,
194            cookie,
195            marks,
196            transport,
197            __source_breaking,
198        } = value;
199
200        let family = family.ok_or(IpSocketStateError::MissingField("family"))?;
201        let cookie = cookie.ok_or(IpSocketStateError::MissingField("cookie"))?;
202        let marks = marks.ok_or(IpSocketStateError::MissingField("marks"))?;
203        let transport = transport.ok_or(IpSocketStateError::MissingField("transport"))?;
204
205        match family {
206            fnet::IpVersion::V4 => Ok(IpSocketState::V4(to_ip_socket_specific(
207                src_addr, dst_addr, cookie, marks, transport,
208            )?)),
209            fnet::IpVersion::V6 => Ok(IpSocketState::V6(to_ip_socket_specific(
210                src_addr, dst_addr, cookie, marks, transport,
211            )?)),
212        }
213    }
214}
215
216impl From<IpSocketState> for fnet_sockets::IpSocketState {
217    fn from(state: IpSocketState) -> Self {
218        match state {
219            IpSocketState::V4(state) => state.into(),
220            IpSocketState::V6(state) => state.into(),
221        }
222    }
223}
224
225/// Lowest-level socket state information that ensures all fields are for the
226/// same IP version.
227#[derive(Debug, PartialEq, Eq, Clone, GenericOverIp)]
228#[generic_over_ip(I, Ip)]
229pub struct IpSocketStateSpecific<I: Ip> {
230    /// The source address of the socket.
231    pub src_addr: Option<I::Addr>,
232    /// The destination address of the socket.
233    pub dst_addr: Option<I::Addr>,
234    /// The cookie of the socket.
235    pub cookie: u64,
236    /// The marks of the socket.
237    pub marks: Marks,
238    /// The transport state of the socket.
239    pub transport: IpSocketTransportState,
240}
241
242impl<I: Ip> From<IpSocketStateSpecific<I>> for fnet_sockets::IpSocketState {
243    fn from(value: IpSocketStateSpecific<I>) -> Self {
244        let IpSocketStateSpecific { src_addr, dst_addr, cookie, marks, transport } = value;
245
246        fnet_sockets::IpSocketState {
247            family: Some(I::VERSION.into_ext()),
248            src_addr: src_addr.map(|a| net_types::ip::IpAddr::from(a).into_ext()),
249            dst_addr: dst_addr.map(|a| net_types::ip::IpAddr::from(a).into_ext()),
250            cookie: Some(cookie),
251            marks: Some(marks.into()),
252            transport: Some(transport.into()),
253            __source_breaking: fidl::marker::SourceBreaking,
254        }
255    }
256}
257
258/// Extension type for [`fnet_sockets::IpSocketTransportState`].
259#[derive(Debug, PartialEq, Eq, Clone)]
260pub enum IpSocketTransportState {
261    /// TCP socket state.
262    Tcp(IpSocketTcpState),
263    /// UDP socket state.
264    Udp(IpSocketUdpState),
265}
266
267/// Error type for [`IpSocketTransportState`] conversion.
268#[derive(Debug, PartialEq, Error)]
269pub enum IpSocketTransportStateError {
270    /// Error converting a TCP socket state.
271    #[error("tcp validation error: {0}")]
272    Tcp(IpSocketTcpStateError),
273    /// Error converting a UDP socket state.
274    #[error("udp validation error: {0}")]
275    Udp(IpSocketUdpStateError),
276    /// A union type was unknown.
277    #[error("got unexpected union variant: {0}")]
278    UnknownUnionVariant(u64),
279}
280
281impl TryFrom<fnet_sockets::IpSocketTransportState> for IpSocketTransportState {
282    type Error = IpSocketTransportStateError;
283
284    fn try_from(value: fnet_sockets::IpSocketTransportState) -> Result<Self, Self::Error> {
285        match value {
286            fnet_sockets::IpSocketTransportState::Tcp(tcp) => Ok(IpSocketTransportState::Tcp(
287                tcp.try_into().map_err(IpSocketTransportStateError::Tcp)?,
288            )),
289            fnet_sockets::IpSocketTransportState::Udp(udp) => Ok(IpSocketTransportState::Udp(
290                udp.try_into().map_err(IpSocketTransportStateError::Udp)?,
291            )),
292            fnet_sockets::IpSocketTransportState::__SourceBreaking { unknown_ordinal } => {
293                Err(IpSocketTransportStateError::UnknownUnionVariant(unknown_ordinal))
294            }
295        }
296    }
297}
298
299impl From<IpSocketTransportState> for fnet_sockets::IpSocketTransportState {
300    fn from(state: IpSocketTransportState) -> Self {
301        match state {
302            IpSocketTransportState::Tcp(tcp) => {
303                fnet_sockets::IpSocketTransportState::Tcp(tcp.into())
304            }
305            IpSocketTransportState::Udp(udp) => {
306                fnet_sockets::IpSocketTransportState::Udp(udp.into())
307            }
308        }
309    }
310}
311
312/// Extension type for [`fnet_sockets::IpSocketTcpState`].
313#[derive(Debug, PartialEq, Eq, Clone)]
314pub struct IpSocketTcpState {
315    /// The source port of the socket.
316    pub src_port: Option<u16>,
317    /// The destination port of the socket.
318    pub dst_port: Option<u16>,
319    /// The TCP state machine state for the socket.
320    pub state: fnet_tcp::State,
321    /// Extended TCP information if the TCP_INFO extension was requested.
322    pub tcp_info: Option<TcpInfo>,
323}
324
325/// Error type for [`IpSocketTcpState`] conversion.
326#[derive(Debug, PartialEq, Error)]
327pub enum IpSocketTcpStateError {
328    /// Missing a required field.
329    #[error("missing field: {0}")]
330    MissingField(&'static str),
331    /// Error converting a [`TcpInfo`].
332    #[error("tcp info error: {0}")]
333    TcpInfo(TcpInfoError),
334}
335
336impl TryFrom<fnet_sockets::IpSocketTcpState> for IpSocketTcpState {
337    type Error = IpSocketTcpStateError;
338
339    fn try_from(value: fnet_sockets::IpSocketTcpState) -> Result<Self, Self::Error> {
340        let fnet_sockets::IpSocketTcpState {
341            src_port,
342            dst_port,
343            state,
344            tcp_info,
345            __source_breaking,
346        } = value;
347
348        let state = state.ok_or(IpSocketTcpStateError::MissingField("state"))?;
349
350        Ok(IpSocketTcpState {
351            src_port,
352            dst_port,
353            state,
354            tcp_info: tcp_info
355                .map(|t| t.try_into())
356                .transpose()
357                .map_err(|e| IpSocketTcpStateError::TcpInfo(e))?,
358        })
359    }
360}
361
362impl From<IpSocketTcpState> for fnet_sockets::IpSocketTcpState {
363    fn from(state: IpSocketTcpState) -> Self {
364        let IpSocketTcpState { src_port, dst_port, state, tcp_info } = state;
365        fnet_sockets::IpSocketTcpState {
366            src_port,
367            dst_port,
368            state: Some(state),
369            tcp_info: tcp_info.map(Into::into),
370            __source_breaking: fidl::marker::SourceBreaking,
371        }
372    }
373}
374
375/// Extension type for [`fnet_tcp::Info`].
376#[derive(Debug, PartialEq, Eq, Clone)]
377pub struct TcpInfo {
378    /// The state of the TCP connection.
379    pub state: fnet_tcp::State,
380    /// The congestion control state of the TCP connection.
381    pub ca_state: fnet_tcp::CongestionControlState,
382    /// The retransmission timeout of the TCP connection in microseconds.
383    pub rto_usec: Option<u32>,
384    /// The time since the most recent data was sent on the connection in milliseconds.
385    pub tcpi_last_data_sent_msec: Option<u32>,
386    /// The time since the most recent ACK was received in milliseconds.
387    pub tcpi_last_ack_recv_msec: Option<u32>,
388    /// The estimated smoothed roundtrip time in microseconds.
389    pub rtt_usec: Option<u32>,
390    /// The smoothed mean deviation of the roundtrip time in microseconds.
391    pub rtt_var_usec: Option<u32>,
392    /// The sending slow start threshold in segments.
393    pub snd_ssthresh: u32,
394    /// The current sending congestion window in segments.
395    pub snd_cwnd: u32,
396    /// The total number of retransmissions.
397    pub tcpi_total_retrans: u32,
398    /// The total number of segments sent.
399    pub tcpi_segs_out: u64,
400    /// The total number of segments received.
401    pub tcpi_segs_in: u64,
402    /// Whether reordering has been seen on the connection.
403    pub reorder_seen: bool,
404    /// The send MSS for this endpoint.
405    pub tcpi_snd_mss: Option<u32>,
406    /// The receive MSS for this endpoint.
407    pub tcpi_rcv_mss: Option<u32>,
408}
409
410/// Error type for [`TcpInfo`] conversion.
411#[derive(Debug, PartialEq, Error)]
412pub enum TcpInfoError {
413    /// Missing a required field.
414    #[error("missing field: {0}")]
415    MissingField(&'static str),
416}
417
418impl TryFrom<fnet_tcp::Info> for TcpInfo {
419    type Error = TcpInfoError;
420
421    fn try_from(value: fnet_tcp::Info) -> Result<Self, Self::Error> {
422        let fnet_tcp::Info {
423            state,
424            ca_state,
425            rto_usec,
426            tcpi_last_data_sent_msec,
427            tcpi_last_ack_recv_msec,
428            rtt_usec,
429            rtt_var_usec,
430            snd_ssthresh,
431            snd_cwnd,
432            tcpi_total_retrans,
433            tcpi_segs_out,
434            tcpi_segs_in,
435            reorder_seen,
436            tcpi_snd_mss,
437            tcpi_rcv_mss,
438            __source_breaking,
439        } = value;
440
441        Ok(TcpInfo {
442            state: state.ok_or(TcpInfoError::MissingField("state"))?,
443            ca_state: ca_state.ok_or(TcpInfoError::MissingField("ca_state"))?,
444            rto_usec,
445            tcpi_last_data_sent_msec,
446            tcpi_last_ack_recv_msec,
447            rtt_usec,
448            rtt_var_usec,
449            snd_ssthresh: snd_ssthresh.ok_or(TcpInfoError::MissingField("snd_ssthresh"))?,
450            snd_cwnd: snd_cwnd.ok_or(TcpInfoError::MissingField("snd_cwnd"))?,
451            tcpi_total_retrans: tcpi_total_retrans
452                .ok_or(TcpInfoError::MissingField("tcpi_total_retrans"))?,
453            tcpi_segs_out: tcpi_segs_out.ok_or(TcpInfoError::MissingField("tcpi_segs_out"))?,
454            tcpi_segs_in: tcpi_segs_in.ok_or(TcpInfoError::MissingField("tcpi_segs_in"))?,
455            reorder_seen: reorder_seen.ok_or(TcpInfoError::MissingField("reorder_seen"))?,
456            tcpi_snd_mss,
457            tcpi_rcv_mss,
458        })
459    }
460}
461
462impl From<TcpInfo> for fnet_tcp::Info {
463    fn from(info: TcpInfo) -> Self {
464        let TcpInfo {
465            state,
466            ca_state,
467            rto_usec,
468            tcpi_last_data_sent_msec,
469            tcpi_last_ack_recv_msec,
470            rtt_usec,
471            rtt_var_usec,
472            snd_ssthresh,
473            snd_cwnd,
474            tcpi_total_retrans,
475            tcpi_segs_out,
476            tcpi_segs_in,
477            reorder_seen,
478            tcpi_snd_mss,
479            tcpi_rcv_mss,
480        } = info;
481        fnet_tcp::Info {
482            state: Some(state),
483            ca_state: Some(ca_state),
484            rto_usec: rto_usec,
485            tcpi_last_data_sent_msec,
486            tcpi_last_ack_recv_msec,
487            rtt_usec: rtt_usec,
488            rtt_var_usec: rtt_var_usec,
489            snd_ssthresh: Some(snd_ssthresh),
490            snd_cwnd: Some(snd_cwnd),
491            tcpi_total_retrans: Some(tcpi_total_retrans),
492            tcpi_segs_out: Some(tcpi_segs_out),
493            tcpi_segs_in: Some(tcpi_segs_in),
494            reorder_seen: Some(reorder_seen),
495            tcpi_snd_mss,
496            tcpi_rcv_mss,
497            __source_breaking: fidl::marker::SourceBreaking,
498        }
499    }
500}
501
502/// Extension type for [`fnet_sockets::IpSocketUdpState`].
503#[derive(Debug, PartialEq, Eq, Clone)]
504pub struct IpSocketUdpState {
505    /// The source port of the socket.
506    pub src_port: Option<u16>,
507    /// The destination port of the socket.
508    pub dst_port: Option<u16>,
509    /// The UDP pseudo-state machine state for the socket.
510    pub state: fnet_udp::State,
511}
512
513/// Error type for [`IpSocketUdpState`] conversion.
514#[derive(Debug, PartialEq, Error)]
515pub enum IpSocketUdpStateError {
516    /// Missing a required field.
517    #[error("missing field: {0}")]
518    MissingField(&'static str),
519}
520
521impl TryFrom<fnet_sockets::IpSocketUdpState> for IpSocketUdpState {
522    type Error = IpSocketUdpStateError;
523
524    fn try_from(value: fnet_sockets::IpSocketUdpState) -> Result<Self, Self::Error> {
525        let fnet_sockets::IpSocketUdpState { src_port, dst_port, state, __source_breaking } = value;
526
527        let state = state.ok_or(IpSocketUdpStateError::MissingField("state"))?;
528
529        Ok(IpSocketUdpState { src_port, dst_port, state })
530    }
531}
532
533impl From<IpSocketUdpState> for fnet_sockets::IpSocketUdpState {
534    fn from(state: IpSocketUdpState) -> Self {
535        let IpSocketUdpState { src_port, dst_port, state } = state;
536        fnet_sockets::IpSocketUdpState {
537            src_port,
538            dst_port,
539            state: Some(state),
540            __source_breaking: fidl::marker::SourceBreaking,
541        }
542    }
543}
544
545/// Errors returned by [`iterate_ip`]
546#[derive(Debug, Error)]
547pub enum IterateIpError {
548    /// The specified matcher was the first invalid one.
549    #[error("invalid matcher at position {0}")]
550    InvalidMatcher(usize),
551    /// An unknown response was received on the call to `Diagnostics.IterateIp`
552    #[error("unknown ordinal on Diagnostics.IterateIp call: {0}")]
553    UnknownOrdinal(u64),
554    /// A low-level FIDL error was encountered on the call to
555    /// `Diagnostics.IterateIp`.
556    #[error("fidl error during Diagnostics.IterateIp call: {0}")]
557    Fidl(fidl::Error),
558}
559
560impl From<fidl::Error> for IterateIpError {
561    fn from(e: fidl::Error) -> Self {
562        IterateIpError::Fidl(e)
563    }
564}
565
566/// Errors returned by the stream returned from [`iterate_ip`].
567#[derive(Debug, Error)]
568pub enum IpIteratorError {
569    /// The netstack returned an empty batch of sockets
570    #[error("received empty batch of sockets")]
571    EmptyBatch,
572    /// A low-level FIDL error was encountered on the call to
573    /// `Diagnostics.IterateIp`.
574    #[error("fidl error during Diagnostics.IterateIp call: {0}")]
575    Fidl(fidl::Error),
576    /// An error was encountered while converting a socket state.
577    #[error("error converting socket state: {0}")]
578    Conversion(IpSocketStateError),
579}
580
581impl From<fidl::Error> for IpIteratorError {
582    fn from(e: fidl::Error) -> Self {
583        IpIteratorError::Fidl(e)
584    }
585}
586
587/// Send a request to `Diagnostics.IterateIp` and drive the resulting
588/// `IpIterator`.
589///
590/// `IpIterator` returns a series of batches of sockets matching the query, the
591/// returned stream flattens those batches into individual sockets. If an error
592/// is encuontered during iteration, it is returned and iteration halts.
593//
594// TODO(https://github.com/rust-lang/rust/issues/130043): Remove types from the
595// precise capturing clause on the stream.
596pub async fn iterate_ip<M, I>(
597    diagnostics: &fnet_sockets::DiagnosticsProxy,
598    extensions: fnet_sockets::Extensions,
599    matchers: M,
600) -> Result<impl Stream<Item = Result<IpSocketState, IpIteratorError>> + use<M, I>, IterateIpError>
601where
602    M: IntoIterator<Item = I>,
603    I: Into<fnet_sockets::IpSocketMatcher>,
604{
605    let (proxy, server_end) = fidl::endpoints::create_proxy::<fnet_sockets::IpIteratorMarker>();
606    match diagnostics
607        .iterate_ip(
608            server_end,
609            extensions,
610            &matchers.into_iter().map(Into::into).collect::<Vec<_>>()[..],
611        )
612        .await?
613    {
614        fnet_sockets::IterateIpResult::Ok(_empty) => Ok(()),
615        fnet_sockets::IterateIpResult::InvalidMatcher(fnet_sockets::InvalidMatcher { index }) => {
616            Err(IterateIpError::InvalidMatcher(index as usize))
617        }
618        fnet_sockets::IterateIpResult::__SourceBreaking { unknown_ordinal } => {
619            Err(IterateIpError::UnknownOrdinal(unknown_ordinal))
620        }
621    }?;
622
623    Ok(futures::stream::try_unfold((proxy, true), |(proxy, has_more)| async move {
624        if !has_more {
625            return Ok(None);
626        }
627
628        let (batch, has_more) = proxy.next().await?;
629        if batch.is_empty() && has_more {
630            Err(IpIteratorError::EmptyBatch)
631        } else {
632            Ok(Some((
633                futures::stream::iter(
634                    batch
635                        .into_iter()
636                        .map(|s| s.try_into().map_err(|e| IpIteratorError::Conversion(e))),
637                ),
638                (proxy, has_more),
639            )))
640        }
641    })
642    .try_flatten())
643}
644
645/// Errors returned by [`disconnect_ip`]
646#[derive(Debug, Error)]
647pub enum DisconnectIpError {
648    /// The specified matcher was the first invalid one.
649    #[error("invalid matcher at position {0}")]
650    InvalidMatcher(usize),
651    /// Specified matchers would a priori match all sockets.
652    #[error("matchers were unconstrained")]
653    UnconstrainedMatchers,
654    /// An unknown response was received on the call to `Control.DisconnectIp`
655    #[error("unknown ordinal on Control.DisconnectIp call: {0}")]
656    UnknownOrdinal(u64),
657    /// A low-level FIDL error was encountered on the call to
658    /// `Control.DisconnectIp`.
659    #[error("fidl error during Control.DisconnectIp call: {0}")]
660    Fidl(fidl::Error),
661}
662
663/// Send a request to `Control.DisconnectIp` with the provided matchers.
664pub async fn disconnect_ip<M, I>(
665    control: &fnet_sockets::ControlProxy,
666    matchers: M,
667) -> Result<usize, DisconnectIpError>
668where
669    M: IntoIterator<Item = I>,
670    I: Into<fnet_sockets::IpSocketMatcher>,
671{
672    match control
673        .disconnect_ip(&fnet_sockets::ControlDisconnectIpRequest {
674            matchers: Some(matchers.into_iter().map(Into::into).collect()),
675            __source_breaking: fidl::marker::SourceBreaking,
676        })
677        .await
678    {
679        Ok(r) => match r {
680            fnet_sockets::DisconnectIpResult::Ok(fnet_sockets::DisconnectIpResponse {
681                disconnected,
682            }) => {
683                // Unwrap is safe because usize is always at least u32.
684                Ok(disconnected.try_into().unwrap())
685            }
686            fnet_sockets::DisconnectIpResult::InvalidMatcher(fnet_sockets::InvalidMatcher {
687                index,
688            }) => {
689                // Unwrap is safe because usize is always at least u32.
690                Err(DisconnectIpError::InvalidMatcher(index.try_into().unwrap()))
691            }
692            fnet_sockets::DisconnectIpResult::UnconstrainedMatchers(fnet_sockets::Empty) => {
693                Err(DisconnectIpError::UnconstrainedMatchers)
694            }
695            fnet_sockets::DisconnectIpResult::__SourceBreaking { unknown_ordinal } => {
696                Err(DisconnectIpError::UnknownOrdinal(unknown_ordinal))
697            }
698        },
699        Err(e) => Err(DisconnectIpError::Fidl(e)),
700    }
701}
702
703/// Errors returned by the stream returned from [`watch_destruction`].
704#[derive(Debug, Error)]
705pub enum DestructionWatcherError {
706    /// A low-level FIDL error was encountered on the call to
707    /// `DestructionWatcher.Watch`.
708    #[error("fidl error during DestructionWatcher.Watch call: {0}")]
709    Fidl(fidl::Error),
710    /// An error was encountered while converting a socket state.
711    #[error("error converting socket state: {0}")]
712    Conversion(IpSocketStateError),
713    /// The netstack returned an empty batch of sockets.
714    #[error("received empty batch of sockets")]
715    EmptyBatch,
716}
717
718impl From<fidl::Error> for DestructionWatcherError {
719    fn from(e: fidl::Error) -> Self {
720        DestructionWatcherError::Fidl(e)
721    }
722}
723
724/// Get a destruction watcher and drive it to yield individual sockets.
725pub async fn watch_destruction(
726    diagnostics: &fnet_sockets::DiagnosticsProxy,
727) -> Result<impl Stream<Item = Result<IpSocketState, DestructionWatcherError>> + use<>, fidl::Error>
728{
729    let (proxy, server_end) =
730        fidl::endpoints::create_proxy::<fnet_sockets::DestructionWatcherMarker>();
731    diagnostics.get_destruction_watcher(server_end).await?;
732
733    Ok(futures::stream::try_unfold(proxy, |proxy| async {
734        let batch = proxy.watch().await?;
735        if batch.is_empty() {
736            Err(DestructionWatcherError::EmptyBatch)
737        } else {
738            let batch = batch
739                .into_iter()
740                .map(|s| s.try_into().map_err(DestructionWatcherError::Conversion))
741                .collect::<Result<Vec<_>, _>>()?;
742            Ok::<_, DestructionWatcherError>(Some((
743                futures::stream::iter(batch.into_iter().map(Ok)),
744                proxy,
745            )))
746        }
747    })
748    .try_flatten())
749}
750
751#[cfg(test)]
752mod tests {
753    use super::*;
754
755    use std::num::NonZeroU64;
756
757    use assert_matches::assert_matches;
758    use fidl_fuchsia_net as fnet;
759    use fidl_fuchsia_net_tcp as fnet_tcp;
760    use futures::{FutureExt as _, StreamExt as _, future, pin_mut};
761    use net_declare::{fidl_ip, fidl_subnet, net_ip_v4, net_ip_v6};
762    use test_case::test_case;
763
764    #[test_case(
765        fnet_sockets::IpSocketMatcher::Family(fnet::IpVersion::V4),
766        IpSocketMatcher::Family(ip::IpVersion::V4);
767        "FamilyIpv4"
768    )]
769    #[test_case(
770        fnet_sockets::IpSocketMatcher::Family(fnet::IpVersion::V6),
771        IpSocketMatcher::Family(ip::IpVersion::V6);
772        "FamilyIpv6"
773    )]
774    #[test_case(
775        fnet_sockets::IpSocketMatcher::SrcAddr(fnet_matchers::BoundAddress::Bound(
776            fnet_matchers::Address {
777                matcher: fnet_matchers::AddressMatcherType::Subnet(fidl_subnet!("192.0.2.0/24")),
778                invert: true,
779            }
780        )),
781        IpSocketMatcher::SrcAddr(fnet_matchers_ext::BoundAddress::Bound(
782            fnet_matchers_ext::Address {
783                matcher: fnet_matchers_ext::AddressMatcherType::Subnet(
784                    fnet_matchers_ext::Subnet::try_from(fidl_subnet!("192.0.2.0/24")).unwrap()
785                ),
786                invert: true,
787            }
788        ));
789        "SrcAddr"
790    )]
791    #[test_case(
792        fnet_sockets::IpSocketMatcher::DstAddr(fnet_matchers::BoundAddress::Bound(
793            fnet_matchers::Address {
794                matcher: fnet_matchers::AddressMatcherType::Subnet(fidl_subnet!("2001:db8::/32")),
795                invert: false,
796            }
797        )),
798        IpSocketMatcher::DstAddr(fnet_matchers_ext::BoundAddress::Bound(
799            fnet_matchers_ext::Address {
800                matcher: fnet_matchers_ext::AddressMatcherType::Subnet(
801                    fnet_matchers_ext::Subnet::try_from(fidl_subnet!("2001:db8::/32")).unwrap()
802                ),
803                invert: false,
804            }
805        ));
806        "DstAddr"
807    )]
808    #[test_case(
809        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Tcp(
810            fnet_matchers::TcpSocket::Empty(fnet_matchers::Empty)
811        )),
812        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Tcp(
813            fnet_matchers_ext::TcpSocket::Empty
814        ));
815        "ProtoTcp"
816    )]
817    #[test_case(
818        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Udp(
819            fnet_matchers::UdpSocket::Empty(fnet_matchers::Empty)
820        )),
821        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Udp(
822            fnet_matchers_ext::UdpSocket::Empty
823        ));
824        "ProtoUdp"
825    )]
826    #[test_case(
827        fnet_sockets::IpSocketMatcher::BoundInterface(fnet_matchers::BoundInterface::Unbound(
828            fnet_matchers::Empty
829        )),
830        IpSocketMatcher::BoundInterface(fnet_matchers_ext::BoundInterface::Unbound);
831        "BoundInterfaceUnbound"
832    )]
833    #[test_case(
834        fnet_sockets::IpSocketMatcher::BoundInterface(fnet_matchers::BoundInterface::Bound(
835            fnet_matchers::Interface::Id(1)
836        )),
837        IpSocketMatcher::BoundInterface(fnet_matchers_ext::BoundInterface::Bound(
838            fnet_matchers_ext::Interface::Id(NonZeroU64::new(1).unwrap())
839        ));
840        "BoundInterfaceBound"
841    )]
842    #[test_case(
843        fnet_sockets::IpSocketMatcher::Cookie(fnet_matchers::SocketCookie {
844            cookie: 12345,
845            invert: false,
846        }),
847        IpSocketMatcher::Cookie(fnet_matchers::SocketCookie {
848            cookie: 12345,
849            invert: false,
850        });
851        "Cookie"
852    )]
853    #[test_case(
854        fnet_sockets::IpSocketMatcher::Mark(fnet_matchers::MarkInDomain {
855            domain: fnet::MarkDomain::Mark1,
856            mark: fnet_matchers::Mark::Unmarked(fnet_matchers::Unmarked),
857        }),
858        IpSocketMatcher::Mark(fnet_matchers_ext::MarkInDomain {
859            domain: fnet::MarkDomain::Mark1,
860            mark: fnet_matchers_ext::Mark::Unmarked,
861        });
862        "Mark"
863    )]
864    #[test_case(
865        fnet_sockets::IpSocketMatcher::SrcAddr(fnet_matchers::BoundAddress::Unbound(fnet_matchers::Empty)),
866        IpSocketMatcher::SrcAddr(fnet_matchers_ext::BoundAddress::Unbound);
867        "SrcAddrUnbound"
868    )]
869    #[test_case(
870        fnet_sockets::IpSocketMatcher::DstAddr(fnet_matchers::BoundAddress::Unbound(fnet_matchers::Empty)),
871        IpSocketMatcher::DstAddr(fnet_matchers_ext::BoundAddress::Unbound);
872        "DstAddrUnbound"
873    )]
874    #[test_case(
875        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Tcp(
876            fnet_matchers::TcpSocket::SrcPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
877        )),
878        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Tcp(
879            fnet_matchers_ext::TcpSocket::SrcPort(fnet_matchers_ext::BoundPort::Unbound)
880        ));
881        "ProtoTcpSrcPortUnbound"
882    )]
883    #[test_case(
884        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Tcp(
885            fnet_matchers::TcpSocket::DstPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
886        )),
887        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Tcp(
888            fnet_matchers_ext::TcpSocket::DstPort(fnet_matchers_ext::BoundPort::Unbound)
889        ));
890        "ProtoTcpDstPortUnbound"
891    )]
892    #[test_case(
893        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Udp(
894            fnet_matchers::UdpSocket::SrcPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
895        )),
896        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Udp(
897            fnet_matchers_ext::UdpSocket::SrcPort(fnet_matchers_ext::BoundPort::Unbound)
898        ));
899        "ProtoUdpSrcPortUnbound"
900    )]
901    #[test_case(
902        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Udp(
903            fnet_matchers::UdpSocket::DstPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
904        )),
905        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Udp(
906            fnet_matchers_ext::UdpSocket::DstPort(fnet_matchers_ext::BoundPort::Unbound)
907        ));
908        "ProtoUdpDstPortUnbound"
909    )]
910    #[test_case(
911        fnet_tcp::Info {
912            state: Some(fnet_tcp::State::Established),
913            ca_state: Some(fnet_tcp::CongestionControlState::Open),
914            rto_usec: Some(1),
915            tcpi_last_data_sent_msec: Some(2),
916            tcpi_last_ack_recv_msec: Some(3),
917            rtt_usec: Some(4),
918            rtt_var_usec: Some(5),
919            snd_ssthresh: Some(6),
920            snd_cwnd: Some(7),
921            tcpi_total_retrans: Some(8),
922            tcpi_segs_out: Some(9),
923            tcpi_segs_in: Some(10),
924            reorder_seen: Some(true),
925            tcpi_snd_mss: Some(11),
926            tcpi_rcv_mss: Some(12),
927            __source_breaking: fidl::marker::SourceBreaking,
928        },
929        TcpInfo {
930            state: fnet_tcp::State::Established,
931            ca_state: fnet_tcp::CongestionControlState::Open,
932            rto_usec: Some(1),
933            tcpi_last_data_sent_msec: Some(2),
934            tcpi_last_ack_recv_msec: Some(3),
935            rtt_usec: Some(4),
936            rtt_var_usec: Some(5),
937            snd_ssthresh: 6,
938            snd_cwnd: 7,
939            tcpi_total_retrans: 8,
940            tcpi_segs_out: 9,
941            tcpi_segs_in: 10,
942            reorder_seen: true,
943            tcpi_snd_mss: Some(11),
944            tcpi_rcv_mss: Some(12),
945        };
946        "TcpInfo"
947    )]
948    #[test_case(
949        fnet_sockets::IpSocketState {
950            family: Some(fnet::IpVersion::V4),
951            src_addr: Some(fidl_ip!("192.168.1.1")),
952            dst_addr: Some(fidl_ip!("192.168.1.2")),
953            cookie: Some(1234),
954            marks: Some(fnet::Marks {
955                mark_1: Some(1111),
956                mark_2: None,
957                __source_breaking: fidl::marker::SourceBreaking,
958            }),
959            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
960                fnet_sockets::IpSocketTcpState {
961                    src_port: Some(1111),
962                    dst_port: Some(2222),
963                    state: Some(fnet_tcp::State::Established),
964                    tcp_info: None,
965                    __source_breaking: fidl::marker::SourceBreaking,
966                },
967            )),
968            __source_breaking: fidl::marker::SourceBreaking,
969        },
970        IpSocketState::V4(IpSocketStateSpecific {
971            src_addr: Some(net_ip_v4!("192.168.1.1")),
972            dst_addr: Some(net_ip_v4!("192.168.1.2")),
973            cookie: 1234,
974            marks: fnet::Marks {
975                mark_1: Some(1111),
976                mark_2: None,
977                __source_breaking: fidl::marker::SourceBreaking,
978            }.into(),
979            transport: IpSocketTransportState::Tcp(IpSocketTcpState {
980                src_port: Some(1111),
981                dst_port: Some(2222),
982                state: fnet_tcp::State::Established,
983                tcp_info: None,
984            }),
985        });
986        "IpSocketStateV4"
987    )]
988    #[test_case(
989        fnet_sockets::IpSocketState {
990            family: Some(fnet::IpVersion::V6),
991            src_addr: Some(fidl_ip!("2001:db8::1")),
992            dst_addr: Some(fidl_ip!("2001:db8::2")),
993            cookie: Some(1234),
994            marks: Some(fnet::Marks {
995                mark_1: Some(1111),
996                mark_2: None,
997                __source_breaking: fidl::marker::SourceBreaking,
998            }),
999            transport: Some(fnet_sockets::IpSocketTransportState::Udp(
1000                fnet_sockets::IpSocketUdpState {
1001                    src_port: Some(3333),
1002                    dst_port: Some(4444),
1003                    state: Some(fnet_udp::State::Connected),
1004                    __source_breaking: fidl::marker::SourceBreaking,
1005                },
1006            )),
1007            __source_breaking: fidl::marker::SourceBreaking,
1008        },
1009        IpSocketState::V6(IpSocketStateSpecific {
1010            src_addr: Some(net_ip_v6!("2001:db8::1")),
1011            dst_addr: Some(net_ip_v6!("2001:db8::2")),
1012            cookie: 1234,
1013            marks: fnet::Marks {
1014                mark_1: Some(1111),
1015                mark_2: None,
1016                __source_breaking: fidl::marker::SourceBreaking,
1017            }.into(),
1018            transport: IpSocketTransportState::Udp(IpSocketUdpState {
1019                src_port: Some(3333),
1020                dst_port: Some(4444),
1021                state: fnet_udp::State::Connected,
1022            }),
1023        });
1024        "IpSocketStateV6"
1025    )]
1026    fn convert_from_fidl_and_back<F, E>(fidl_type: F, local_type: E)
1027    where
1028        E: TryFrom<F> + Clone + std::fmt::Debug + PartialEq,
1029        <E as TryFrom<F>>::Error: std::fmt::Debug + PartialEq,
1030        F: From<E> + Clone + std::fmt::Debug + PartialEq,
1031    {
1032        assert_eq!(fidl_type.clone().try_into(), Ok(local_type.clone()));
1033        assert_eq!(<_ as Into<F>>::into(local_type), fidl_type);
1034    }
1035
1036    #[test_case(
1037        fnet_sockets::IpSocketMatcher::__SourceBreaking { unknown_ordinal: 100 } =>
1038            Err(IpSocketMatcherError::UnknownUnionVariant(100));
1039        "UnknownUnionVariant"
1040    )]
1041    #[test_case(
1042        fnet_sockets::IpSocketMatcher::SrcAddr(fnet_matchers::BoundAddress::Bound(
1043            fnet_matchers::Address {
1044                matcher: fnet_matchers::AddressMatcherType::__SourceBreaking { unknown_ordinal: 100 },
1045                invert: false,
1046            }
1047        )) => Err(IpSocketMatcherError::Address(fnet_matchers_ext::BoundAddressError::Address(
1048            fnet_matchers_ext::AddressError::AddressMatcherType(
1049                fnet_matchers_ext::AddressMatcherTypeError::UnknownUnionVariant
1050            )
1051        )));
1052        "AddressError"
1053    )]
1054    #[test_case(
1055        fnet_sockets::IpSocketMatcher::Proto(
1056            fnet_matchers::SocketTransportProtocol::__SourceBreaking { unknown_ordinal: 100 }
1057        ) => Err(IpSocketMatcherError::TransportProtocol(
1058            fnet_matchers_ext::SocketTransportProtocolError::UnknownUnionVariant(100)
1059        ));
1060        "TransportProtocolError"
1061    )]
1062    #[test_case(
1063        fnet_sockets::IpSocketMatcher::BoundInterface(
1064            fnet_matchers::BoundInterface::__SourceBreaking { unknown_ordinal: 100 }
1065        ) => Err(IpSocketMatcherError::BoundInterface(
1066            fnet_matchers_ext::BoundInterfaceError::UnknownUnionVariant(100)
1067        ));
1068        "BoundInterfaceError"
1069    )]
1070    #[test_case(
1071        fnet_sockets::IpSocketMatcher::Mark(fnet_matchers::MarkInDomain {
1072            domain: fnet::MarkDomain::Mark1,
1073            mark: fnet_matchers::Mark::__SourceBreaking { unknown_ordinal: 100 },
1074        }) => Err(IpSocketMatcherError::Mark(
1075            fnet_matchers_ext::MarkInDomainError::Mark(
1076                fnet_matchers_ext::MarkError::UnknownUnionVariant(100)
1077            )
1078        ));
1079        "MarkError"
1080    )]
1081    fn ip_socket_matcher_try_from_error(
1082        fidl: fnet_sockets::IpSocketMatcher,
1083    ) -> Result<IpSocketMatcher, IpSocketMatcherError> {
1084        IpSocketMatcher::try_from(fidl)
1085    }
1086
1087    #[test_case(
1088        fnet_sockets::IpSocketState {
1089            family: None,
1090            src_addr: Some(fidl_ip!("192.168.1.1")),
1091            dst_addr: Some(fidl_ip!("192.168.1.2")),
1092            cookie: Some(1234),
1093            marks: Some(fnet::Marks {
1094                mark_1: Some(1111),
1095                mark_2: None,
1096                __source_breaking: fidl::marker::SourceBreaking,
1097            }),
1098            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1099                fnet_sockets::IpSocketTcpState {
1100                    src_port: Some(1111),
1101                    dst_port: Some(2222),
1102                    state: Some(fnet_tcp::State::Established),
1103                    tcp_info: None,
1104                    __source_breaking: fidl::marker::SourceBreaking,
1105                },
1106            )),
1107            __source_breaking: fidl::marker::SourceBreaking,
1108        } => Err(IpSocketStateError::MissingField("family"));
1109        "MissingFamily"
1110    )]
1111    #[test_case(
1112        fnet_sockets::IpSocketState {
1113            family: Some(fnet::IpVersion::V4),
1114            src_addr: Some(fidl_ip!("192.168.1.1")),
1115            dst_addr: Some(fidl_ip!("192.168.1.2")),
1116            cookie: None,
1117            marks: Some(fnet::Marks {
1118                mark_1: Some(1111),
1119                mark_2: None,
1120                __source_breaking: fidl::marker::SourceBreaking,
1121            }),
1122            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1123                fnet_sockets::IpSocketTcpState {
1124                    src_port: Some(1111),
1125                    dst_port: Some(2222),
1126                    state: Some(fnet_tcp::State::Established),
1127                    tcp_info: None,
1128                    __source_breaking: fidl::marker::SourceBreaking,
1129                },
1130            )),
1131            __source_breaking: fidl::marker::SourceBreaking,
1132        } => Err(IpSocketStateError::MissingField("cookie"));
1133        "MissingCookie"
1134    )]
1135    #[test_case(
1136        fnet_sockets::IpSocketState {
1137            family: Some(fnet::IpVersion::V4),
1138            src_addr: Some(fidl_ip!("192.168.1.1")),
1139            dst_addr: Some(fidl_ip!("192.168.1.2")),
1140            cookie: Some(1234),
1141            marks: None,
1142            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1143                fnet_sockets::IpSocketTcpState {
1144                    src_port: Some(1111),
1145                    dst_port: Some(2222),
1146                    state: Some(fnet_tcp::State::Established),
1147                    tcp_info: None,
1148                    __source_breaking: fidl::marker::SourceBreaking,
1149                },
1150            )),
1151            __source_breaking: fidl::marker::SourceBreaking,
1152        } => Err(IpSocketStateError::MissingField("marks"));
1153        "MissingMarks"
1154    )]
1155    #[test_case(
1156        fnet_sockets::IpSocketState {
1157            family: Some(fnet::IpVersion::V4),
1158            src_addr: Some(fidl_ip!("192.168.1.1")),
1159            dst_addr: Some(fidl_ip!("192.168.1.2")),
1160            cookie: Some(1234),
1161            marks: Some(fnet::Marks {
1162                mark_1: Some(1111),
1163                mark_2: None,
1164                __source_breaking: fidl::marker::SourceBreaking,
1165            }),
1166            transport: None,
1167            __source_breaking: fidl::marker::SourceBreaking,
1168        } => Err(IpSocketStateError::MissingField("transport"));
1169        "MissingTransport"
1170    )]
1171    #[test_case(
1172        fnet_sockets::IpSocketState {
1173            family: Some(fnet::IpVersion::V4),
1174            src_addr: Some(fidl_ip!("192.168.1.1")),
1175            dst_addr: Some(fidl_ip!("2001:db8::2")),
1176            cookie: Some(1234),
1177            marks: Some(fnet::Marks {
1178                mark_1: Some(1111),
1179                mark_2: None,
1180                __source_breaking: fidl::marker::SourceBreaking,
1181            }),
1182            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1183                fnet_sockets::IpSocketTcpState {
1184                    src_port: Some(1111),
1185                    dst_port: Some(2222),
1186                    state: Some(fnet_tcp::State::Established),
1187                    tcp_info: None,
1188                    __source_breaking: fidl::marker::SourceBreaking,
1189                },
1190            )),
1191            __source_breaking: fidl::marker::SourceBreaking,
1192        } => Err(IpSocketStateError::VersionMismatch);
1193        "VersionMismatchV4"
1194    )]
1195    #[test_case(
1196        fnet_sockets::IpSocketState {
1197            family: Some(fnet::IpVersion::V6),
1198            src_addr: Some(fidl_ip!("192.168.1.1")),
1199            dst_addr: Some(fidl_ip!("2001:db8::2")),
1200            cookie: Some(1234),
1201            marks: Some(fnet::Marks {
1202                mark_1: Some(1111),
1203                mark_2: None,
1204                __source_breaking: fidl::marker::SourceBreaking,
1205            }),
1206            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1207                fnet_sockets::IpSocketTcpState {
1208                    src_port: Some(1111),
1209                    dst_port: Some(2222),
1210                    state: Some(fnet_tcp::State::Established),
1211                    tcp_info: None,
1212                    __source_breaking: fidl::marker::SourceBreaking,
1213                },
1214            )),
1215            __source_breaking: fidl::marker::SourceBreaking,
1216        } => Err(IpSocketStateError::VersionMismatch);
1217        "VersionMismatchV6"
1218    )]
1219    #[test_case(
1220        fnet_sockets::IpSocketState {
1221            family: Some(fnet::IpVersion::V4),
1222            src_addr: Some(fidl_ip!("192.168.1.1")),
1223            dst_addr: Some(fidl_ip!("192.168.1.2")),
1224            cookie: Some(1234),
1225            marks: Some(fnet::Marks {
1226                mark_1: Some(1111),
1227                mark_2: None,
1228                __source_breaking: fidl::marker::SourceBreaking,
1229            }),
1230            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1231                fnet_sockets::IpSocketTcpState {
1232                    src_port: Some(1111),
1233                    dst_port: Some(2222),
1234                    state: None,
1235                    tcp_info: None,
1236                    __source_breaking: fidl::marker::SourceBreaking,
1237                },
1238            )),
1239            __source_breaking: fidl::marker::SourceBreaking,
1240        } => Err(IpSocketStateError::Transport(IpSocketTransportStateError::Tcp(
1241                IpSocketTcpStateError::MissingField("state"),
1242        )));
1243        "MissingTcpState"
1244    )]
1245    #[test_case(
1246        fnet_sockets::IpSocketState {
1247            family: Some(fnet::IpVersion::V6),
1248            src_addr: Some(fidl_ip!("2001:db8::1")),
1249            dst_addr: Some(fidl_ip!("2001:db8::2")),
1250            cookie: Some(1234),
1251            marks: Some(fnet::Marks {
1252                mark_1: Some(1111),
1253                mark_2: None,
1254                __source_breaking: fidl::marker::SourceBreaking,
1255            }),
1256            transport: Some(fnet_sockets::IpSocketTransportState::Udp(
1257                fnet_sockets::IpSocketUdpState {
1258                    src_port: Some(3333),
1259                    dst_port: Some(4444),
1260                    state: None,
1261                    __source_breaking: fidl::marker::SourceBreaking,
1262                },
1263            )),
1264            __source_breaking: fidl::marker::SourceBreaking,
1265        } => Err(IpSocketStateError::Transport(IpSocketTransportStateError::Udp(
1266                IpSocketUdpStateError::MissingField("state"),
1267        )));
1268        "MissingUdpState"
1269    )]
1270    fn ip_socket_state_try_from_error(
1271        fidl: fnet_sockets::IpSocketState,
1272    ) -> Result<IpSocketState, IpSocketStateError> {
1273        IpSocketState::try_from(fidl)
1274    }
1275
1276    #[fuchsia_async::run_singlethreaded(test)]
1277    async fn iterate_ip_diagnostics_iterate_ip_error() {
1278        async fn serve_matcher_error(req: fnet_sockets::DiagnosticsRequest) {
1279            match req {
1280                fnet_sockets::DiagnosticsRequest::IterateIp {
1281                    s: _,
1282                    extensions: _,
1283                    matchers: _,
1284                    responder,
1285                } => responder
1286                    .send(&fnet_sockets::IterateIpResult::InvalidMatcher(
1287                        fnet_sockets::InvalidMatcher { index: 0 },
1288                    ))
1289                    .unwrap(),
1290                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1291            };
1292        }
1293
1294        let (diagnostics, diagnostics_server_end) =
1295            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1296
1297        let (mut diagnostics_request_stream, _control_handle) =
1298            diagnostics_server_end.into_stream_and_control_handle();
1299        let server_fut = diagnostics_request_stream
1300            .next()
1301            .then(|req| {
1302                serve_matcher_error(req.expect("Request stream ended unexpectedly").unwrap())
1303            })
1304            .fuse();
1305        let client_fut = iterate_ip::<[IpSocketMatcher; 0], _>(
1306            &diagnostics,
1307            fnet_sockets::Extensions::empty(),
1308            [],
1309        );
1310
1311        pin_mut!(server_fut);
1312        pin_mut!(client_fut);
1313
1314        let ((), resp) = future::join(server_fut, client_fut).await;
1315
1316        assert_matches!(
1317            // Discard the stream because it can't be formatted.
1318            resp.map(|_| ()),
1319            Err(IterateIpError::InvalidMatcher(0))
1320        );
1321    }
1322
1323    #[fuchsia_async::run_singlethreaded(test)]
1324    async fn iterate_ip_next_error() {
1325        async fn serve_matcher(req: fnet_sockets::DiagnosticsRequest) {
1326            match req {
1327                fnet_sockets::DiagnosticsRequest::IterateIp {
1328                    s,
1329                    extensions: _,
1330                    matchers: _,
1331                    responder,
1332                } => {
1333                    s.close_with_epitaph(zx_status::Status::PEER_CLOSED).unwrap();
1334                    responder.send(&fnet_sockets::IterateIpResult::Ok(fnet_sockets::Empty)).unwrap()
1335                }
1336                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1337            }
1338        }
1339
1340        let (diagnostics, diagnostics_server_end) =
1341            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1342
1343        let (mut diagnostics_request_stream, _control_handle) =
1344            diagnostics_server_end.into_stream_and_control_handle();
1345        let server_fut = diagnostics_request_stream
1346            .next()
1347            .then(|req| serve_matcher(req.expect("Request stream ended unexpectedly").unwrap()))
1348            .fuse();
1349
1350        let client_fut = iterate_ip::<[IpSocketMatcher; 0], _>(
1351            &diagnostics,
1352            fnet_sockets::Extensions::empty(),
1353            [],
1354        );
1355
1356        let ((), resp) = future::join(server_fut, client_fut).await;
1357        let stream = resp.unwrap();
1358        pin_mut!(stream);
1359
1360        assert_matches!(
1361            stream.try_next().await,
1362            Err(IpIteratorError::Fidl(fidl::Error::ClientChannelClosed { .. }))
1363        );
1364    }
1365
1366    #[fuchsia_async::run_singlethreaded(test)]
1367    async fn iterate_ip_empty_batch() {
1368        async fn serve_matcher(req: fnet_sockets::DiagnosticsRequest) {
1369            match req {
1370                fnet_sockets::DiagnosticsRequest::IterateIp {
1371                    s,
1372                    extensions: _,
1373                    matchers: _,
1374                    responder,
1375                } => {
1376                    responder
1377                        .send(&fnet_sockets::IterateIpResult::Ok(fnet_sockets::Empty))
1378                        .unwrap();
1379
1380                    let (mut stream, _control) = s.into_stream_and_control_handle();
1381                    match stream.next().await.unwrap().unwrap() {
1382                        fidl_fuchsia_net_sockets::IpIteratorRequest::Next { responder } => {
1383                            // Send an empty batch but indicate there's more to come.
1384                            responder.send(&[], true).unwrap();
1385                        }
1386                        fidl_fuchsia_net_sockets::IpIteratorRequest::_UnknownMethod { .. } => {
1387                            unreachable!()
1388                        }
1389                    }
1390                }
1391                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1392            }
1393        }
1394
1395        let (diagnostics, diagnostics_server_end) =
1396            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1397
1398        let (mut diagnostics_request_stream, _control_handle) =
1399            diagnostics_server_end.into_stream_and_control_handle();
1400        let server_fut = diagnostics_request_stream
1401            .next()
1402            .then(|req| serve_matcher(req.expect("Request stream ended unexpectedly").unwrap()))
1403            .fuse();
1404
1405        let client_fut = async {
1406            let stream = iterate_ip::<[IpSocketMatcher; 0], _>(
1407                &diagnostics,
1408                fnet_sockets::Extensions::empty(),
1409                [],
1410            )
1411            .await
1412            .unwrap();
1413            pin_mut!(stream);
1414            stream.try_next().await
1415        };
1416
1417        let ((), resp) = future::join(server_fut, client_fut).await;
1418        assert_matches!(resp, Err(IpIteratorError::EmptyBatch));
1419    }
1420
1421    #[fuchsia_async::run_singlethreaded(test)]
1422    async fn iterate_ip_success() {
1423        let socket_1 = fnet_sockets::IpSocketState {
1424            family: Some(fnet::IpVersion::V4),
1425            src_addr: Some(fidl_ip!("192.168.1.1")),
1426            dst_addr: Some(fidl_ip!("192.168.1.2")),
1427            cookie: Some(1234),
1428            marks: Some(fnet::Marks {
1429                mark_1: Some(1111),
1430                mark_2: None,
1431                __source_breaking: fidl::marker::SourceBreaking,
1432            }),
1433            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1434                fnet_sockets::IpSocketTcpState {
1435                    src_port: Some(1111),
1436                    dst_port: Some(2222),
1437                    state: Some(fnet_tcp::State::Established),
1438                    tcp_info: None,
1439                    __source_breaking: fidl::marker::SourceBreaking,
1440                },
1441            )),
1442            __source_breaking: fidl::marker::SourceBreaking,
1443        };
1444
1445        let socket_2 = fnet_sockets::IpSocketState {
1446            family: Some(fnet::IpVersion::V4),
1447            src_addr: Some(fidl_ip!("192.168.8.1")),
1448            dst_addr: Some(fidl_ip!("192.168.8.2")),
1449            cookie: Some(9876),
1450            marks: Some(fnet::Marks {
1451                mark_1: None,
1452                mark_2: Some(2222),
1453                __source_breaking: fidl::marker::SourceBreaking,
1454            }),
1455            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1456                fnet_sockets::IpSocketTcpState {
1457                    src_port: Some(3333),
1458                    dst_port: Some(4444),
1459                    state: Some(fnet_tcp::State::TimeWait),
1460                    tcp_info: None,
1461                    __source_breaking: fidl::marker::SourceBreaking,
1462                },
1463            )),
1464            __source_breaking: fidl::marker::SourceBreaking,
1465        };
1466
1467        let socket_3 = fnet_sockets::IpSocketState {
1468            family: Some(fnet::IpVersion::V6),
1469            src_addr: Some(fidl_ip!("2001:db8::1")),
1470            dst_addr: Some(fidl_ip!("2001:db8::2")),
1471            cookie: Some(5678),
1472            marks: Some(fnet::Marks {
1473                mark_1: None,
1474                mark_2: None,
1475                __source_breaking: fidl::marker::SourceBreaking,
1476            }),
1477            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1478                fnet_sockets::IpSocketTcpState {
1479                    src_port: Some(5555),
1480                    dst_port: Some(6666),
1481                    state: Some(fnet_tcp::State::TimeWait),
1482                    tcp_info: None,
1483                    __source_breaking: fidl::marker::SourceBreaking,
1484                },
1485            )),
1486            __source_breaking: fidl::marker::SourceBreaking,
1487        };
1488
1489        let (diagnostics, diagnostics_server_end) =
1490            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1491
1492        let serve_matcher = async |req: fnet_sockets::DiagnosticsRequest| {
1493            let responses = &[vec![socket_1.clone()], vec![socket_2.clone(), socket_3.clone()]];
1494
1495            match req {
1496                fnet_sockets::DiagnosticsRequest::IterateIp {
1497                    s,
1498                    extensions: _,
1499                    matchers: _,
1500                    responder,
1501                } => {
1502                    responder
1503                        .send(&fnet_sockets::IterateIpResult::Ok(fnet_sockets::Empty))
1504                        .unwrap();
1505
1506                    let (mut stream, _control) = s.into_stream_and_control_handle();
1507                    for (i, resp) in responses.iter().enumerate() {
1508                        match stream.next().await.unwrap().unwrap() {
1509                            fidl_fuchsia_net_sockets::IpIteratorRequest::Next { responder } => {
1510                                let has_more = i < responses.len() - 1;
1511                                responder.send(&resp, has_more).unwrap();
1512                            }
1513                            fidl_fuchsia_net_sockets::IpIteratorRequest::_UnknownMethod {
1514                                ..
1515                            } => {
1516                                unreachable!()
1517                            }
1518                        }
1519                    }
1520                }
1521                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1522            };
1523        };
1524
1525        let (mut diagnostics_request_stream, _control_handle) =
1526            diagnostics_server_end.into_stream_and_control_handle();
1527
1528        let server_fut = diagnostics_request_stream
1529            .next()
1530            .then(|req| serve_matcher(req.expect("Request stream ended unexpectedly").unwrap()))
1531            .fuse();
1532
1533        let client_fut = async {
1534            iterate_ip::<[IpSocketMatcher; 0], _>(
1535                &diagnostics,
1536                fnet_sockets::Extensions::empty(),
1537                [],
1538            )
1539            .await
1540            .unwrap()
1541            .try_collect::<Vec<_>>()
1542            .await
1543            .unwrap()
1544        };
1545
1546        let ((), sockets) = future::join(server_fut, client_fut).await;
1547        assert_eq!(
1548            sockets,
1549            vec![
1550                socket_1.clone().try_into().unwrap(),
1551                socket_2.clone().try_into().unwrap(),
1552                socket_3.clone().try_into().unwrap()
1553            ]
1554        );
1555    }
1556
1557    #[fuchsia_async::run_singlethreaded(test)]
1558    async fn watch_destruction_success() {
1559        let socket_1 = fnet_sockets::IpSocketState {
1560            family: Some(fnet::IpVersion::V4),
1561            src_addr: Some(fidl_ip!("192.168.1.1")),
1562            dst_addr: Some(fidl_ip!("192.168.1.2")),
1563            cookie: Some(1234),
1564            marks: Some(fnet::Marks {
1565                mark_1: Some(1111),
1566                mark_2: None,
1567                __source_breaking: fidl::marker::SourceBreaking,
1568            }),
1569            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1570                fnet_sockets::IpSocketTcpState {
1571                    src_port: Some(1111),
1572                    dst_port: Some(2222),
1573                    state: Some(fnet_tcp::State::Established),
1574                    tcp_info: None,
1575                    __source_breaking: fidl::marker::SourceBreaking,
1576                },
1577            )),
1578            __source_breaking: fidl::marker::SourceBreaking,
1579        };
1580
1581        let socket_2 = fnet_sockets::IpSocketState {
1582            family: Some(fnet::IpVersion::V4),
1583            src_addr: Some(fidl_ip!("192.168.8.1")),
1584            dst_addr: Some(fidl_ip!("192.168.8.2")),
1585            cookie: Some(9876),
1586            marks: Some(fnet::Marks {
1587                mark_1: None,
1588                mark_2: Some(2222),
1589                __source_breaking: fidl::marker::SourceBreaking,
1590            }),
1591            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1592                fnet_sockets::IpSocketTcpState {
1593                    src_port: Some(3333),
1594                    dst_port: Some(4444),
1595                    state: Some(fnet_tcp::State::TimeWait),
1596                    tcp_info: None,
1597                    __source_breaking: fidl::marker::SourceBreaking,
1598                },
1599            )),
1600            __source_breaking: fidl::marker::SourceBreaking,
1601        };
1602
1603        let (diagnostics, diagnostics_server_end) =
1604            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1605        let serve_watcher = async |req: fnet_sockets::DiagnosticsRequest| match req {
1606            fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { watcher, responder } => {
1607                responder.send().unwrap();
1608                let mut stream = watcher.into_stream();
1609                let batches = [
1610                    Some(vec![socket_1.clone()]),
1611                    Some(vec![socket_2.clone(), socket_1.clone()]),
1612                    None,
1613                ];
1614                for batch in batches {
1615                    let req = stream.next().await.unwrap().unwrap();
1616                    let responder = match req {
1617                        fnet_sockets::DestructionWatcherRequest::Watch { responder } => responder,
1618                        fnet_sockets::DestructionWatcherRequest::_UnknownMethod { .. } => {
1619                            unreachable!()
1620                        }
1621                    };
1622                    if let Some(batch) = batch {
1623                        responder.send(&batch).unwrap();
1624                    } else {
1625                        drop(responder);
1626                    }
1627                }
1628            }
1629            fnet_sockets::DiagnosticsRequest::IterateIp { .. } => unreachable!(),
1630        };
1631
1632        let (mut diagnostics_request_stream, _control_handle) =
1633            diagnostics_server_end.into_stream_and_control_handle();
1634        let server_fut = diagnostics_request_stream
1635            .next()
1636            .then(|req| serve_watcher(req.expect("Request stream ended unexpectedly").unwrap()))
1637            .fuse();
1638
1639        let expected_socket_1: IpSocketState = socket_1.clone().try_into().unwrap();
1640        let expected_socket_2: IpSocketState = socket_2.clone().try_into().unwrap();
1641
1642        let client_fut = async {
1643            let stream = watch_destruction(&diagnostics).await.unwrap();
1644            pin_mut!(stream);
1645
1646            assert_matches!(
1647                stream.next().await,
1648                Some(Ok(sock)) => assert_eq!(sock, expected_socket_1)
1649            );
1650            assert_matches!(
1651                stream.next().await,
1652                Some(Ok(sock)) => assert_eq!(sock, expected_socket_2)
1653            );
1654            assert_matches!(
1655                stream.next().await,
1656                Some(Ok(sock)) => assert_eq!(sock, expected_socket_1)
1657            );
1658            assert_matches!(stream.next().await, Some(Err(DestructionWatcherError::Fidl(_))));
1659        };
1660
1661        let _: ((), ()) = future::join(server_fut, client_fut).await;
1662    }
1663
1664    #[test_case(
1665        None,
1666        DestructionWatcherError::EmptyBatch;
1667        "empty_batch"
1668    )]
1669    #[test_case(
1670        Some(fnet_sockets::IpSocketState {
1671            family: None,
1672            src_addr: Some(fidl_ip!("192.168.1.1")),
1673            dst_addr: Some(fidl_ip!("192.168.1.2")),
1674            cookie: Some(1234),
1675            marks: Some(fnet::Marks {
1676                mark_1: Some(1111),
1677                mark_2: None,
1678                __source_breaking: fidl::marker::SourceBreaking,
1679            }),
1680            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1681                fnet_sockets::IpSocketTcpState {
1682                    src_port: Some(1111),
1683                    dst_port: Some(2222),
1684                    state: Some(fnet_tcp::State::Established),
1685                    tcp_info: None,
1686                    __source_breaking: fidl::marker::SourceBreaking,
1687                },
1688            )),
1689            __source_breaking: fidl::marker::SourceBreaking,
1690        }),
1691        DestructionWatcherError::Conversion(IpSocketStateError::MissingField("family"));
1692        "conversion_error"
1693    )]
1694    #[fuchsia_async::run_singlethreaded(test)]
1695    async fn watch_destruction_error(
1696        socket: Option<fnet_sockets::IpSocketState>,
1697        expected_error: DestructionWatcherError,
1698    ) {
1699        let (diagnostics, diagnostics_server_end) =
1700            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1701        let serve_watcher = async |req: fnet_sockets::DiagnosticsRequest| match req {
1702            fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { watcher, responder } => {
1703                responder.send().unwrap();
1704                let mut stream = watcher.into_stream();
1705                let req = stream.next().await.unwrap().unwrap();
1706                let responder = match req {
1707                    fnet_sockets::DestructionWatcherRequest::Watch { responder } => responder,
1708                    fnet_sockets::DestructionWatcherRequest::_UnknownMethod { .. } => {
1709                        unreachable!()
1710                    }
1711                };
1712                let batch = match socket {
1713                    None => vec![],
1714                    Some(s) => vec![s],
1715                };
1716                responder.send(&batch).unwrap();
1717            }
1718            fnet_sockets::DiagnosticsRequest::IterateIp { .. } => unreachable!(),
1719        };
1720
1721        let (mut diagnostics_request_stream, _control_handle) =
1722            diagnostics_server_end.into_stream_and_control_handle();
1723        let server_fut = diagnostics_request_stream
1724            .next()
1725            .then(|req| serve_watcher(req.expect("Request stream ended unexpectedly").unwrap()))
1726            .fuse();
1727
1728        let client_fut = async {
1729            let stream = watch_destruction(&diagnostics).await.unwrap();
1730            pin_mut!(stream);
1731
1732            let result = stream.next().await.expect("got a result");
1733            assert_matches!(stream.next().await, None);
1734            result
1735        };
1736
1737        let ((), result) = future::join(server_fut, client_fut).await;
1738        match expected_error {
1739            DestructionWatcherError::EmptyBatch => {
1740                assert_matches!(result, Err(DestructionWatcherError::EmptyBatch));
1741            }
1742            DestructionWatcherError::Conversion(b) => {
1743                assert_matches!(result, Err(DestructionWatcherError::Conversion(a)) if a == b);
1744            }
1745            DestructionWatcherError::Fidl(_) => unreachable!(),
1746        }
1747    }
1748}
fidl_fuchsia_net_sockets_ext/lib.rs

fidl_fuchsia_net_sockets_ext/
lib.rs
