dhcp_client_core/
deps.rs

1// Copyright 2023 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//! Defines trait abstractions for platform dependencies of the DHCP client
6//! core, and provides fake implementations of these dependencies for testing
7//! purposes.
8
9use diagnostics_traits::InspectableInstant;
10use fuchsia_async as fasync;
11use rand::Rng;
12use std::future::Future;
13
14/// Provides access to random number generation.
15pub trait RngProvider {
16    /// The random number generator being provided.
17    type RNG: Rng + ?Sized;
18
19    /// Get access to a random number generator.
20    fn get_rng(&mut self) -> &mut Self::RNG;
21}
22
23impl RngProvider for rand::rngs::StdRng {
24    type RNG = Self;
25    fn get_rng(&mut self) -> &mut Self::RNG {
26        self
27    }
28}
29
30#[derive(Clone, Copy, PartialEq, Debug)]
31/// Contains information about a datagram received on a socket.
32pub struct DatagramInfo<T> {
33    /// The length in bytes of the datagram received on the socket.
34    pub length: usize,
35    /// The address associated with the datagram received on the socket
36    /// (usually, the address from which the datagram was received).
37    pub address: T,
38}
39
40#[derive(thiserror::Error, Debug)]
41/// Errors encountered while performing a socket operation.
42pub enum SocketError {
43    /// Failure while attempting to open a socket.
44    #[error("failed to open socket: {0}")]
45    FailedToOpen(anyhow::Error),
46    /// Tried to bind a socket on a nonexistent interface.
47    #[error("tried to bind socket on nonexistent interface")]
48    NoInterface,
49    /// The hardware type of the interface is unsupported.
50    #[error("unsupported hardware type")]
51    UnsupportedHardwareType,
52    /// The host we are attempting to send to is unreachable.
53    #[error("host unreachable")]
54    HostUnreachable,
55    /// The network is unreachable.
56    #[error("network unreachable")]
57    NetworkUnreachable,
58    /// The address is not available.
59    #[error("address not available")]
60    AddrNotAvailable,
61    /// Broken pipe.
62    #[error("broken pipe")]
63    BrokenPipe,
64    /// The connection was aborted.
65    #[error("connection aborted")]
66    ConnectionAborted,
67    /// Other IO errors observed on socket operations.
68    #[error("socket error: {0}")]
69    Other(std::io::Error),
70}
71
72/// Abstracts sending and receiving datagrams on a socket.
73pub trait Socket<T> {
74    /// Sends a datagram containing the contents of `buf` to `addr`.
75    fn send_to(&self, buf: &[u8], addr: T) -> impl Future<Output = Result<(), SocketError>>;
76
77    /// Receives a datagram into `buf`, returning the number of bytes received
78    /// and the address the datagram was received from.
79    fn recv_from(
80        &self,
81        buf: &mut [u8],
82    ) -> impl Future<Output = Result<DatagramInfo<T>, SocketError>>;
83}
84
85/// Provides access to AF_PACKET sockets.
86pub trait PacketSocketProvider {
87    /// The type of sockets provided by this `PacketSocketProvider`.
88    type Sock: Socket<net_types::ethernet::Mac>;
89
90    /// Gets a packet socket bound to the device on which the DHCP client
91    /// protocol is being performed. The packet socket should already be bound
92    /// to the appropriate device and protocol number.
93    fn get_packet_socket(&self) -> impl Future<Output = Result<Self::Sock, SocketError>>;
94}
95
96/// Provides access to UDP sockets.
97pub trait UdpSocketProvider {
98    /// The type of sockets provided by this `UdpSocketProvider`.
99    type Sock: Socket<std::net::SocketAddr>;
100
101    /// Gets a UDP socket bound to the given address. The UDP socket should be
102    /// allowed to send broadcast packets.
103    fn bind_new_udp_socket(
104        &self,
105        bound_addr: std::net::SocketAddr,
106    ) -> impl Future<Output = Result<Self::Sock, SocketError>>;
107}
108
109/// A type representing an instant in time.
110pub trait Instant:
111    Sized + Ord + Copy + Clone + std::fmt::Debug + Send + Sync + InspectableInstant
112{
113    /// Returns the time `self + duration`. Panics if `self + duration` would
114    /// overflow the underlying instant storage type.
115    fn add(&self, duration: std::time::Duration) -> Self;
116
117    /// Returns the instant halfway between `self` and `other`.
118    fn average(&self, other: Self) -> Self;
119}
120
121impl Instant for fasync::MonotonicInstant {
122    fn add(&self, duration: std::time::Duration) -> Self {
123        *self + duration.into()
124    }
125
126    fn average(&self, other: Self) -> Self {
127        let lower = *self.min(&other);
128        let higher = *self.max(&other);
129        lower + (higher - lower) / 2
130    }
131}
132
133/// Provides access to system-time-related operations.
134pub trait Clock {
135    /// The type representing monotonic system time.
136    type Instant: Instant;
137
138    /// Completes once the monotonic system time is at or after the given time.
139    fn wait_until(&self, time: Self::Instant) -> impl Future<Output = ()>;
140
141    /// Gets the monotonic system time.
142    fn now(&self) -> Self::Instant;
143}
144
145#[cfg(test)]
146pub(crate) mod testutil {
147    use super::*;
148    use diagnostics_traits::InstantPropertyName;
149    use futures::StreamExt as _;
150    use futures::channel::{mpsc, oneshot};
151    use futures::lock::Mutex;
152    use rand::SeedableRng as _;
153    use std::cell::RefCell;
154    use std::cmp::Reverse;
155    use std::collections::BTreeMap;
156    use std::future::Future;
157    use std::ops::{Deref as _, DerefMut as _};
158    use std::rc::Rc;
159
160    /// Provides a seedable implementation of `RngProvider` using `StdRng`.
161    pub(crate) struct FakeRngProvider {
162        std_rng: rand::rngs::StdRng,
163    }
164
165    impl FakeRngProvider {
166        pub(crate) fn new(seed: u64) -> Self {
167            Self { std_rng: rand::rngs::StdRng::seed_from_u64(seed) }
168        }
169    }
170
171    impl RngProvider for FakeRngProvider {
172        type RNG = rand::rngs::StdRng;
173        fn get_rng(&mut self) -> &mut Self::RNG {
174            &mut self.std_rng
175        }
176    }
177
178    /// Provides a fake implementation of `Socket` using `mpsc` channels.
179    ///
180    /// Simply forwards pairs of (payload, address) over the channel. This means
181    /// that the "sent to" address from the sender side is actually observed as
182    /// the "received from" address on the receiver side.
183    pub(crate) struct FakeSocket<T> {
184        sender: mpsc::UnboundedSender<(Vec<u8>, T)>,
185        receiver: Mutex<mpsc::UnboundedReceiver<(Vec<u8>, T)>>,
186    }
187
188    impl<T> FakeSocket<T> {
189        pub(crate) fn new_pair() -> (FakeSocket<T>, FakeSocket<T>) {
190            let (send_a, recv_a) = mpsc::unbounded();
191            let (send_b, recv_b) = mpsc::unbounded();
192            (
193                FakeSocket { sender: send_a, receiver: Mutex::new(recv_b) },
194                FakeSocket { sender: send_b, receiver: Mutex::new(recv_a) },
195            )
196        }
197    }
198
199    impl<T: Send> Socket<T> for FakeSocket<T> {
200        async fn send_to(&self, buf: &[u8], addr: T) -> Result<(), SocketError> {
201            let FakeSocket { sender, receiver: _ } = self;
202            sender.clone().unbounded_send((buf.to_vec(), addr)).expect("unbounded_send error");
203            Ok(())
204        }
205
206        async fn recv_from(&self, buf: &mut [u8]) -> Result<DatagramInfo<T>, SocketError> {
207            let FakeSocket { receiver, sender: _ } = self;
208            let mut receiver = receiver.lock().await;
209            let (bytes, addr) = receiver.next().await.expect("TestSocket receiver closed");
210            if buf.len() < bytes.len() {
211                panic!("TestSocket receiver would produce short read")
212            }
213            (buf[..bytes.len()]).copy_from_slice(&bytes);
214            Ok(DatagramInfo { length: bytes.len(), address: addr })
215        }
216    }
217
218    impl<T, U> Socket<U> for T
219    where
220        T: AsRef<FakeSocket<U>>,
221        U: Send + 'static,
222    {
223        async fn send_to(&self, buf: &[u8], addr: U) -> Result<(), SocketError> {
224            self.as_ref().send_to(buf, addr).await
225        }
226
227        async fn recv_from(&self, buf: &mut [u8]) -> Result<DatagramInfo<U>, SocketError> {
228            self.as_ref().recv_from(buf).await
229        }
230    }
231
232    /// Fake socket provider implementation that vends out copies of
233    /// the same `FakeSocket`.
234    ///
235    /// These copies will compete to receive and send on the same underlying
236    /// `mpsc` channels.
237    pub(crate) struct FakeSocketProvider<T, E> {
238        /// The socket being vended out.
239        pub(crate) socket: Rc<FakeSocket<T>>,
240
241        /// If present, used to notify tests when the client binds new sockets.
242        pub(crate) bound_events: Option<mpsc::UnboundedSender<E>>,
243    }
244
245    impl<T, E> FakeSocketProvider<T, E> {
246        pub(crate) fn new(socket: FakeSocket<T>) -> Self {
247            Self { socket: Rc::new(socket), bound_events: None }
248        }
249
250        pub(crate) fn new_with_events(
251            socket: FakeSocket<T>,
252            bound_events: mpsc::UnboundedSender<E>,
253        ) -> Self {
254            Self { socket: Rc::new(socket), bound_events: Some(bound_events) }
255        }
256    }
257
258    impl PacketSocketProvider for FakeSocketProvider<net_types::ethernet::Mac, ()> {
259        type Sock = Rc<FakeSocket<net_types::ethernet::Mac>>;
260        async fn get_packet_socket(&self) -> Result<Self::Sock, SocketError> {
261            let Self { socket, bound_events } = self;
262            if let Some(bound_events) = bound_events {
263                bound_events.unbounded_send(()).expect("events receiver should not be dropped");
264            }
265            Ok(socket.clone())
266        }
267    }
268
269    impl UdpSocketProvider for FakeSocketProvider<std::net::SocketAddr, std::net::SocketAddr> {
270        type Sock = Rc<FakeSocket<std::net::SocketAddr>>;
271        async fn bind_new_udp_socket(
272            &self,
273            bound_addr: std::net::SocketAddr,
274        ) -> Result<Self::Sock, SocketError> {
275            let Self { socket, bound_events } = self;
276            if let Some(bound_events) = bound_events {
277                bound_events
278                    .unbounded_send(bound_addr)
279                    .expect("events receiver should not be dropped");
280            }
281            Ok(socket.clone())
282        }
283    }
284
285    #[derive(Copy, Clone, PartialEq, Eq, Debug, PartialOrd, Ord)]
286    pub(crate) struct TestInstant(pub(crate) std::time::Duration);
287
288    impl InspectableInstant for TestInstant {
289        fn record<I: diagnostics_traits::Inspector>(
290            &self,
291            name: InstantPropertyName,
292            inspector: &mut I,
293        ) {
294            inspector.record_debug(name.into(), self);
295        }
296    }
297
298    impl Instant for TestInstant {
299        fn add(&self, duration: std::time::Duration) -> Self {
300            Self(self.0.checked_add(duration).unwrap())
301        }
302
303        fn average(&self, other: Self) -> Self {
304            let lower = self.0.min(other.0);
305            let higher = self.0.max(other.0);
306            Self(lower + (higher - lower) / 2)
307        }
308    }
309
310    /// Fake implementation of `Time` that uses `std::time::Duration` as its
311    /// `Instant` type.
312    pub(crate) struct FakeTimeController {
313        pub(super) timer_heap:
314            BTreeMap<std::cmp::Reverse<std::time::Duration>, Vec<oneshot::Sender<()>>>,
315        pub(super) current_time: std::time::Duration,
316    }
317
318    impl FakeTimeController {
319        pub(crate) fn new() -> Rc<RefCell<FakeTimeController>> {
320            Rc::new(RefCell::new(FakeTimeController {
321                timer_heap: BTreeMap::default(),
322                current_time: std::time::Duration::default(),
323            }))
324        }
325    }
326
327    /// Advances the "current time" encoded by `ctl` by `duration`. Any timers
328    /// that were set at or before the resulting "current time" will fire.
329    pub(crate) fn advance(ctl: &Rc<RefCell<FakeTimeController>>, duration: std::time::Duration) {
330        let timers_to_fire = {
331            let mut ctl = ctl.borrow_mut();
332            let FakeTimeController { timer_heap, current_time } = ctl.deref_mut();
333            let next_time = *current_time + duration;
334            *current_time = next_time;
335            timer_heap.split_off(&std::cmp::Reverse(next_time))
336        };
337        for (_, senders) in timers_to_fire {
338            for sender in senders {
339                match sender.send(()) {
340                    Ok(()) => (),
341                    Err(()) => {
342                        // ignore, it's fine for the client core to drop a timer
343                        // to cancel it
344                    }
345                }
346            }
347        }
348    }
349
350    pub(crate) fn run_until_next_timers_fire<F>(
351        executor: &mut fasync::TestExecutor,
352        time: &Rc<RefCell<FakeTimeController>>,
353        main_future: &mut F,
354    ) -> std::task::Poll<F::Output>
355    where
356        F: Future + Unpin,
357    {
358        let poll: std::task::Poll<_> = executor.run_until_stalled(main_future);
359        if poll.is_ready() {
360            return poll;
361        }
362
363        {
364            let mut time = time.borrow_mut();
365            let FakeTimeController { timer_heap, current_time } = time.deref_mut();
366
367            // NOTE: the timer heap is ordered by Reverse<Duration> in order to
368            // facilitate the implementation of `advance()` by making
369            // `BTreeMap::split_off` have the right edge-case behavior. This
370            // makes it easy to get it first_entry vs last_entry mixed up here,
371            // though.
372            let earliest_entry = timer_heap.last_entry().expect("no timers installed");
373
374            let (Reverse(instant), senders) = earliest_entry.remove_entry();
375            *current_time = instant;
376            for sender in senders {
377                match sender.send(()) {
378                    Ok(()) => (),
379                    Err(()) => {
380                        // ignore, it's fine for the client core to drop a timer
381                        // to cancel it
382                    }
383                }
384            }
385        }
386
387        executor.run_until_stalled(main_future)
388    }
389
390    impl Clock for Rc<RefCell<FakeTimeController>> {
391        type Instant = TestInstant;
392
393        fn now(&self) -> Self::Instant {
394            let ctl = self.borrow_mut();
395            let FakeTimeController { timer_heap: _, current_time } = ctl.deref();
396            TestInstant(*current_time)
397        }
398
399        async fn wait_until(&self, TestInstant(time): Self::Instant) {
400            log::info!("registering timer at {:?}", time);
401            let receiver = {
402                let mut ctl = self.borrow_mut();
403                let FakeTimeController { timer_heap, current_time } = ctl.deref_mut();
404                if *current_time >= time {
405                    return;
406                }
407                let (sender, receiver) = oneshot::channel();
408                timer_heap.entry(std::cmp::Reverse(time)).or_default().push(sender);
409                receiver
410            };
411            receiver.await.expect("shouldn't be cancelled")
412        }
413    }
414}
415
416#[cfg(test)]
417mod test {
418    use super::testutil::*;
419    use super::*;
420    use fuchsia_async as fasync;
421    use futures::channel::mpsc;
422    use futures::{FutureExt, StreamExt};
423    use net_declare::std_socket_addr;
424    use std::pin::pin;
425
426    #[test]
427    fn test_rng() {
428        let make_sequence = |seed| {
429            let mut rng = FakeRngProvider::new(seed);
430            std::iter::from_fn(|| Some(rng.get_rng().random::<u32>())).take(5).collect::<Vec<_>>()
431        };
432        assert_eq!(
433            make_sequence(42),
434            make_sequence(42),
435            "should provide identical sequences with same seed"
436        );
437        assert_ne!(
438            make_sequence(42),
439            make_sequence(999999),
440            "should provide different sequences with different seeds"
441        );
442    }
443
444    #[fasync::run_singlethreaded(test)]
445    async fn test_socket() {
446        let (a, b) = FakeSocket::new_pair();
447        let to_send = [
448            (b"hello".to_vec(), "1.2.3.4:5".to_string()),
449            (b"test".to_vec(), "1.2.3.5:5".to_string()),
450            (b"socket".to_vec(), "1.2.3.6:5".to_string()),
451        ];
452
453        let mut buf = [0u8; 10];
454        for (msg, addr) in &to_send {
455            a.send_to(msg, addr.clone()).await.unwrap();
456
457            let DatagramInfo { length: n, address: received_addr } =
458                b.recv_from(&mut buf).await.unwrap();
459            assert_eq!(&received_addr, addr);
460            assert_eq!(&buf[..n], msg);
461        }
462
463        let (a, b) = (b, a);
464        for (msg, addr) in &to_send {
465            a.send_to(msg, addr.clone()).await.unwrap();
466
467            let DatagramInfo { length: n, address: received_addr } =
468                b.recv_from(&mut buf).await.unwrap();
469            assert_eq!(&received_addr, addr);
470            assert_eq!(&buf[..n], msg);
471        }
472    }
473
474    #[fasync::run_singlethreaded(test)]
475    #[should_panic]
476    async fn test_socket_panics_on_short_read() {
477        let (a, b) = FakeSocket::new_pair();
478
479        let mut buf = [0u8; 10];
480        let message = b"this message is way longer than 10 bytes";
481        a.send_to(message, "1.2.3.4:5".to_string()).await.unwrap();
482
483        // Should panic here.
484        let _: Result<_, _> = b.recv_from(&mut buf).await;
485    }
486
487    #[fasync::run_singlethreaded(test)]
488    async fn test_fake_udp_socket_provider() {
489        let (a, b) = FakeSocket::new_pair();
490        let (events_sender, mut events_receiver) = mpsc::unbounded();
491        let provider = FakeSocketProvider::new_with_events(b, events_sender);
492        const ADDR_1: std::net::SocketAddr = std_socket_addr!("1.1.1.1:11");
493        const ADDR_2: std::net::SocketAddr = std_socket_addr!("2.2.2.2:22");
494        const ADDR_3: std::net::SocketAddr = std_socket_addr!("3.3.3.3:33");
495        let b_1 = provider.bind_new_udp_socket(ADDR_1).await.expect("get packet socket");
496        assert_eq!(
497            events_receiver
498                .next()
499                .now_or_never()
500                .expect("should have received bound event")
501                .expect("stream should not have ended"),
502            ADDR_1
503        );
504
505        let b_2 = provider.bind_new_udp_socket(ADDR_2).await.expect("get packet socket");
506        assert_eq!(
507            events_receiver
508                .next()
509                .now_or_never()
510                .expect("should have received bound event")
511                .expect("stream should not have ended"),
512            ADDR_2
513        );
514
515        a.send_to(b"hello", ADDR_3).await.unwrap();
516        a.send_to(b"world", ADDR_3).await.unwrap();
517
518        let mut buf = [0u8; 5];
519        let DatagramInfo { length, address } = b_1.recv_from(&mut buf).await.unwrap();
520        assert_eq!(&buf[..length], b"hello");
521        assert_eq!(address, ADDR_3);
522
523        let DatagramInfo { length, address } = b_2.recv_from(&mut buf).await.unwrap();
524        assert_eq!(&buf[..length], b"world");
525        assert_eq!(address, ADDR_3);
526    }
527
528    #[fasync::run_singlethreaded(test)]
529    async fn test_fake_packet_socket_provider() {
530        let (a, b) = FakeSocket::new_pair();
531        let (events_sender, mut events_receiver) = mpsc::unbounded();
532        let provider = FakeSocketProvider::new_with_events(b, events_sender);
533        let b_1 = provider.get_packet_socket().await.expect("get packet socket");
534        events_receiver
535            .next()
536            .now_or_never()
537            .expect("should have received bound event")
538            .expect("stream should not have ended");
539
540        let b_2 = provider.get_packet_socket().await.expect("get packet socket");
541        events_receiver
542            .next()
543            .now_or_never()
544            .expect("should have received bound event")
545            .expect("stream should not have ended");
546
547        const ADDRESS: net_types::ethernet::Mac = net_declare::net_mac!("01:02:03:04:05:06");
548
549        a.send_to(b"hello", ADDRESS).await.unwrap();
550
551        a.send_to(b"world", ADDRESS).await.unwrap();
552
553        let mut buf = [0u8; 5];
554        let DatagramInfo { length, address } = b_1.recv_from(&mut buf).await.unwrap();
555        assert_eq!(&buf[..length], b"hello");
556        assert_eq!(address, ADDRESS);
557
558        let DatagramInfo { length, address } = b_2.recv_from(&mut buf).await.unwrap();
559        assert_eq!(&buf[..length], b"world");
560        assert_eq!(address, ADDRESS);
561    }
562
563    #[test]
564    fn test_time_controller() {
565        let time_ctl = FakeTimeController::new();
566        assert!(time_ctl.borrow().timer_heap.is_empty());
567        assert_eq!(time_ctl.borrow().current_time, std::time::Duration::from_secs(0));
568        assert_eq!(time_ctl.now(), TestInstant(std::time::Duration::from_secs(0)));
569
570        let mut timer_registered_before_should_fire_1 =
571            pin!(time_ctl.wait_until(TestInstant(std::time::Duration::from_secs(1))));
572        let mut timer_registered_before_should_fire_2 =
573            pin!(time_ctl.wait_until(TestInstant(std::time::Duration::from_secs(1))));
574
575        let mut timer_should_not_fire =
576            pin!(time_ctl.wait_until(TestInstant(std::time::Duration::from_secs(10))));
577
578        // Poll the timer futures once so that they have the chance to
579        // register themselves in our timer heap.
580        {
581            let waker = std::task::Waker::noop();
582            let mut context = futures::task::Context::from_waker(&waker);
583            assert_eq!(
584                timer_registered_before_should_fire_1.poll_unpin(&mut context),
585                futures::task::Poll::Pending
586            );
587            assert_eq!(
588                timer_registered_before_should_fire_2.poll_unpin(&mut context),
589                futures::task::Poll::Pending
590            );
591            assert_eq!(
592                timer_should_not_fire.poll_unpin(&mut context),
593                futures::task::Poll::Pending
594            );
595        }
596
597        {
598            let time_ctl = time_ctl.borrow_mut();
599            let entries = time_ctl.timer_heap.iter().collect::<Vec<_>>();
600            assert_eq!(entries.len(), 2);
601
602            let (time, senders) = entries[0];
603            assert_eq!(time, &std::cmp::Reverse(std::time::Duration::from_secs(10)));
604            assert_eq!(senders.len(), 1);
605
606            let (time, senders) = entries[1];
607            assert_eq!(time, &std::cmp::Reverse(std::time::Duration::from_secs(1)));
608            assert_eq!(senders.len(), 2);
609        }
610
611        advance(&time_ctl, std::time::Duration::from_secs(1));
612
613        assert_eq!(time_ctl.now(), TestInstant(std::time::Duration::from_secs(1)));
614        {
615            let time_ctl = time_ctl.borrow_mut();
616            let entries = time_ctl.timer_heap.iter().collect::<Vec<_>>();
617            assert_eq!(entries.len(), 1);
618            let (time, senders) = entries[0];
619            assert_eq!(time, &std::cmp::Reverse(std::time::Duration::from_secs(10)));
620            assert_eq!(senders.len(), 1);
621        }
622
623        assert_eq!(timer_registered_before_should_fire_1.now_or_never(), Some(()));
624        assert_eq!(timer_registered_before_should_fire_2.now_or_never(), Some(()));
625        assert_eq!(timer_should_not_fire.now_or_never(), None);
626
627        let timer_set_in_past = time_ctl.wait_until(TestInstant(std::time::Duration::from_secs(0)));
628        assert_eq!(timer_set_in_past.now_or_never(), Some(()));
629
630        let timer_set_for_present = time_ctl.wait_until(time_ctl.now());
631        assert_eq!(timer_set_for_present.now_or_never(), Some(()));
632    }
633}
dhcp_client_core/deps.rs

dhcp_client_core/
deps.rs
