| // Copyright 2018 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| //! Netstack3 bindings. |
| //! |
| //! This module provides Fuchsia bindings for the [`netstack3_core`] crate. |
| |
| #[macro_use] |
| mod macros; |
| |
| #[cfg(test)] |
| mod integration_tests; |
| |
| mod context; |
| mod debug_fidl_worker; |
| mod devices; |
| mod ethernet_worker; |
| mod interfaces_admin; |
| mod interfaces_watcher; |
| mod netdevice_worker; |
| mod socket; |
| mod stack_fidl_worker; |
| mod timers; |
| mod util; |
| |
| use std::convert::TryFrom as _; |
| use std::future::Future; |
| use std::num::NonZeroU16; |
| use std::ops::DerefMut as _; |
| use std::sync::Arc; |
| use std::time::Duration; |
| |
| use fidl::endpoints::{DiscoverableProtocolMarker, RequestStream}; |
| use fidl_fuchsia_net_stack as fidl_net_stack; |
| use fuchsia_async as fasync; |
| use fuchsia_component::server::{ServiceFs, ServiceFsDir}; |
| use fuchsia_zircon as zx; |
| use futures::{ |
| channel::mpsc, lock::Mutex, FutureExt as _, SinkExt as _, StreamExt as _, TryStreamExt as _, |
| }; |
| use log::{debug, error}; |
| use packet::{BufferMut, Serializer}; |
| use packet_formats::icmp::{IcmpEchoReply, IcmpMessage, IcmpUnusedCode}; |
| use rand::rngs::OsRng; |
| use util::ConversionContext; |
| |
| use context::Lockable; |
| use devices::{ |
| BindingId, CommonInfo, DeviceInfo, DeviceSpecificInfo, Devices, EthernetInfo, LoopbackInfo, |
| NetdeviceInfo, |
| }; |
| use interfaces_watcher::{InterfaceEventProducer, InterfaceProperties, InterfaceUpdate}; |
| use timers::TimerDispatcher; |
| |
| use net_types::ip::{AddrSubnet, AddrSubnetEither, Ip, Ipv4, Ipv6}; |
| use netstack3_core::{ |
| add_ip_addr_subnet, add_route, |
| context::{CounterContext, EventContext, InstantContext, RngContext, TimerContext}, |
| handle_timer, icmp, update_ipv4_configuration, update_ipv6_configuration, AddableEntryEither, |
| BufferUdpContext, Ctx, DeviceId, DeviceLayerEventDispatcher, IpDeviceConfiguration, IpExt, |
| Ipv4DeviceConfiguration, Ipv6DeviceConfiguration, NonSyncContext, SlaacConfiguration, TimerId, |
| UdpBoundId, UdpConnId, UdpContext, UdpListenerId, |
| }; |
| |
| /// Default MTU for loopback. |
| /// |
| /// This value is also the default value used on Linux. As of writing: |
| /// |
| /// ```shell |
| /// $ ip link show dev lo |
| /// 1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN mode DEFAULT group default qlen 1000 |
| /// link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 |
| /// ``` |
| const DEFAULT_LOOPBACK_MTU: u32 = 65536; |
| |
| pub(crate) trait LockableContext: for<'a> Lockable<'a, Ctx<Self::NonSyncCtx>> { |
| type NonSyncCtx: NonSyncContext + Send; |
| } |
| |
| pub(crate) trait DeviceStatusNotifier { |
| /// A notification that the state of the device with binding Id `id` |
| /// changed. |
| /// |
| /// This method is called by workers that observe devices, such as |
| /// [`EthernetWorker`]. This method is called after all the internal |
| /// structures that cache or store device state are already up to date. The |
| /// only side effect should be notifying other workers or external |
| /// applications that are listening for status changes. |
| fn device_status_changed(&mut self, id: u64); |
| } |
| |
| pub(crate) trait InterfaceEventProducerFactory { |
| fn create_interface_event_producer( |
| &self, |
| id: BindingId, |
| properties: InterfaceProperties, |
| ) -> InterfaceEventProducer; |
| } |
| |
| type IcmpEchoSockets = socket::datagram::SocketCollectionPair<socket::datagram::IcmpEcho>; |
| type UdpSockets = socket::datagram::SocketCollectionPair<socket::datagram::Udp>; |
| |
| impl DeviceStatusNotifier for BindingsNonSyncCtxImpl { |
| fn device_status_changed(&mut self, _id: u64) { |
| // NOTE(brunodalbo) we may want to do more things here in the future, |
| // for now this is only intercepted for testing |
| } |
| } |
| |
| /// Provides an implementation of [`NonSyncContext`]. |
| #[derive(Default)] |
| pub(crate) struct BindingsNonSyncCtxImpl { |
| rng: OsRng, |
| timers: timers::TimerDispatcher<TimerId>, |
| devices: Devices, |
| icmp_echo_sockets: IcmpEchoSockets, |
| udp_sockets: UdpSockets, |
| } |
| |
| impl AsRef<timers::TimerDispatcher<TimerId>> for BindingsNonSyncCtxImpl { |
| fn as_ref(&self) -> &TimerDispatcher<TimerId> { |
| &self.timers |
| } |
| } |
| |
| impl AsMut<timers::TimerDispatcher<TimerId>> for BindingsNonSyncCtxImpl { |
| fn as_mut(&mut self) -> &mut TimerDispatcher<TimerId> { |
| &mut self.timers |
| } |
| } |
| |
| impl AsRef<Devices> for BindingsNonSyncCtxImpl { |
| fn as_ref(&self) -> &Devices { |
| &self.devices |
| } |
| } |
| |
| impl AsMut<Devices> for BindingsNonSyncCtxImpl { |
| fn as_mut(&mut self) -> &mut Devices { |
| &mut self.devices |
| } |
| } |
| |
| impl<'a> Lockable<'a, Ctx<BindingsNonSyncCtxImpl>> for Netstack { |
| type Guard = futures::lock::MutexGuard<'a, Ctx<BindingsNonSyncCtxImpl>>; |
| type Fut = futures::lock::MutexLockFuture<'a, Ctx<BindingsNonSyncCtxImpl>>; |
| fn lock(&'a self) -> Self::Fut { |
| self.ctx.lock() |
| } |
| } |
| |
| impl AsRef<IcmpEchoSockets> for BindingsNonSyncCtxImpl { |
| fn as_ref(&self) -> &IcmpEchoSockets { |
| &self.icmp_echo_sockets |
| } |
| } |
| |
| impl AsMut<IcmpEchoSockets> for BindingsNonSyncCtxImpl { |
| fn as_mut(&mut self) -> &mut IcmpEchoSockets { |
| &mut self.icmp_echo_sockets |
| } |
| } |
| |
| impl AsRef<UdpSockets> for BindingsNonSyncCtxImpl { |
| fn as_ref(&self) -> &UdpSockets { |
| &self.udp_sockets |
| } |
| } |
| |
| impl AsMut<UdpSockets> for BindingsNonSyncCtxImpl { |
| fn as_mut(&mut self) -> &mut UdpSockets { |
| &mut self.udp_sockets |
| } |
| } |
| |
| impl<NonSyncCtx> timers::TimerHandler<TimerId> for Ctx<NonSyncCtx> |
| where |
| NonSyncCtx: NonSyncContext + AsMut<timers::TimerDispatcher<TimerId>> + Send + Sync + 'static, |
| { |
| fn handle_expired_timer(&mut self, timer: TimerId) { |
| let Ctx { sync_ctx, non_sync_ctx } = self; |
| handle_timer(sync_ctx, non_sync_ctx, timer) |
| } |
| |
| fn get_timer_dispatcher(&mut self) -> &mut timers::TimerDispatcher<TimerId> { |
| self.non_sync_ctx.as_mut() |
| } |
| } |
| |
| impl<C> timers::TimerContext<TimerId> for C |
| where |
| C: LockableContext + Clone + Send + Sync + 'static, |
| C::NonSyncCtx: AsMut<timers::TimerDispatcher<TimerId>> + Send + Sync + 'static, |
| { |
| type Handler = Ctx<C::NonSyncCtx>; |
| } |
| |
| impl<D> ConversionContext for D |
| where |
| D: AsRef<Devices>, |
| { |
| fn get_core_id(&self, binding_id: u64) -> Option<DeviceId> { |
| self.as_ref().get_core_id(binding_id) |
| } |
| |
| fn get_binding_id(&self, core_id: DeviceId) -> Option<u64> { |
| self.as_ref().get_binding_id(core_id) |
| } |
| } |
| |
| /// A thin wrapper around `fuchsia_async::Time` that implements `core::Instant`. |
| #[derive(PartialEq, Eq, PartialOrd, Ord, Copy, Clone, Debug)] |
| pub(crate) struct StackTime(fasync::Time); |
| |
| impl netstack3_core::Instant for StackTime { |
| fn duration_since(&self, earlier: StackTime) -> Duration { |
| assert!(self.0 >= earlier.0); |
| // guaranteed not to panic because the assertion ensures that the |
| // difference is non-negative, and all non-negative i64 values are also |
| // valid u64 values |
| Duration::from_nanos(u64::try_from(self.0.into_nanos() - earlier.0.into_nanos()).unwrap()) |
| } |
| |
| fn checked_add(&self, duration: Duration) -> Option<StackTime> { |
| Some(StackTime(fasync::Time::from_nanos( |
| self.0.into_nanos().checked_add(i64::try_from(duration.as_nanos()).ok()?)?, |
| ))) |
| } |
| |
| fn checked_sub(&self, duration: Duration) -> Option<StackTime> { |
| Some(StackTime(fasync::Time::from_nanos( |
| self.0.into_nanos().checked_sub(i64::try_from(duration.as_nanos()).ok()?)?, |
| ))) |
| } |
| } |
| |
| impl InstantContext for BindingsNonSyncCtxImpl { |
| type Instant = StackTime; |
| |
| fn now(&self) -> StackTime { |
| StackTime(fasync::Time::now()) |
| } |
| } |
| |
| impl CounterContext for BindingsNonSyncCtxImpl { |
| fn increment_counter(&mut self, _key: &'static str) {} |
| } |
| |
| impl RngContext for BindingsNonSyncCtxImpl { |
| type Rng = OsRng; |
| |
| fn rng(&self) -> &OsRng { |
| &self.rng |
| } |
| |
| fn rng_mut(&mut self) -> &mut OsRng { |
| &mut self.rng |
| } |
| } |
| |
| impl TimerContext<TimerId> for BindingsNonSyncCtxImpl { |
| fn schedule_timer_instant(&mut self, time: StackTime, id: TimerId) -> Option<StackTime> { |
| self.timers.schedule_timer(id, time) |
| } |
| |
| fn cancel_timer(&mut self, id: TimerId) -> Option<StackTime> { |
| self.timers.cancel_timer(&id) |
| } |
| |
| fn cancel_timers_with<F: FnMut(&TimerId) -> bool>(&mut self, f: F) { |
| self.timers.cancel_timers_with(f); |
| } |
| |
| fn scheduled_instant(&self, id: TimerId) -> Option<StackTime> { |
| self.timers.scheduled_time(&id) |
| } |
| } |
| |
| impl<B> DeviceLayerEventDispatcher<B> for BindingsNonSyncCtxImpl |
| where |
| B: BufferMut, |
| { |
| fn send_frame<S: Serializer<Buffer = B>>( |
| &mut self, |
| device: DeviceId, |
| frame: S, |
| ) -> Result<(), S> { |
| // TODO(wesleyac): Error handling |
| let frame = frame.serialize_vec_outer().map_err(|(_, ser)| ser)?; |
| let dev = match self.devices.get_core_device_mut(device) { |
| Some(dev) => dev, |
| None => { |
| error!("Tried to send frame on device that is not listed: {:?}", device); |
| return Ok(()); |
| } |
| }; |
| |
| match dev.info_mut() { |
| DeviceSpecificInfo::Ethernet(EthernetInfo { |
| common_info: CommonInfo { admin_enabled, mtu: _, events: _, name: _ }, |
| client, |
| mac: _, |
| features: _, |
| phy_up, |
| }) => { |
| if *admin_enabled && *phy_up { |
| client.send(frame.as_ref()) |
| } |
| } |
| DeviceSpecificInfo::Netdevice(NetdeviceInfo { |
| common_info: CommonInfo { admin_enabled, mtu: _, events: _, name: _ }, |
| handler, |
| mac: _, |
| phy_up, |
| }) => { |
| if *admin_enabled && *phy_up { |
| handler.send(frame.as_ref()).unwrap_or_else(|e| { |
| log::warn!("failed to send frame to {:?}: {:?}", handler, e) |
| }) |
| } |
| } |
| DeviceSpecificInfo::Loopback(LoopbackInfo { .. }) => { |
| unreachable!("loopback must not send packets out of the node") |
| } |
| } |
| |
| Ok(()) |
| } |
| } |
| |
| impl<I> icmp::IcmpContext<I> for BindingsNonSyncCtxImpl |
| where |
| I: socket::datagram::SocketCollectionIpExt<socket::datagram::IcmpEcho> + icmp::IcmpIpExt, |
| { |
| fn receive_icmp_error(&mut self, conn: icmp::IcmpConnId<I>, seq_num: u16, err: I::ErrorCode) { |
| I::get_collection_mut(self).receive_icmp_error(conn, seq_num, err) |
| } |
| } |
| |
| impl<I, B: BufferMut> icmp::BufferIcmpContext<I, B> for BindingsNonSyncCtxImpl |
| where |
| I: socket::datagram::SocketCollectionIpExt<socket::datagram::IcmpEcho> + icmp::IcmpIpExt, |
| IcmpEchoReply: for<'a> IcmpMessage<I, &'a [u8], Code = IcmpUnusedCode>, |
| { |
| fn receive_icmp_echo_reply( |
| &mut self, |
| conn: icmp::IcmpConnId<I>, |
| src_ip: I::Addr, |
| dst_ip: I::Addr, |
| id: u16, |
| seq_num: u16, |
| data: B, |
| ) { |
| I::get_collection_mut(self).receive_icmp_echo_reply(conn, src_ip, dst_ip, id, seq_num, data) |
| } |
| } |
| |
| impl<I> UdpContext<I> for BindingsNonSyncCtxImpl |
| where |
| I: socket::datagram::SocketCollectionIpExt<socket::datagram::Udp> + icmp::IcmpIpExt, |
| { |
| fn receive_icmp_error(&mut self, id: UdpBoundId<I>, err: I::ErrorCode) { |
| I::get_collection_mut(self).receive_icmp_error(id, err) |
| } |
| } |
| |
| impl<I, B: BufferMut> BufferUdpContext<I, B> for BindingsNonSyncCtxImpl |
| where |
| I: socket::datagram::SocketCollectionIpExt<socket::datagram::Udp> + IpExt, |
| { |
| fn receive_udp_from_conn( |
| &mut self, |
| conn: UdpConnId<I>, |
| src_ip: I::Addr, |
| src_port: NonZeroU16, |
| body: &B, |
| ) { |
| I::get_collection_mut(self).receive_udp_from_conn(conn, src_ip, src_port, body) |
| } |
| |
| fn receive_udp_from_listen( |
| &mut self, |
| listener: UdpListenerId<I>, |
| src_ip: I::Addr, |
| dst_ip: I::Addr, |
| src_port: Option<NonZeroU16>, |
| body: &B, |
| ) { |
| I::get_collection_mut(self) |
| .receive_udp_from_listen(listener, src_ip, dst_ip, src_port, body) |
| } |
| } |
| |
| impl<I: Ip> EventContext<netstack3_core::IpDeviceEvent<DeviceId, I>> for BindingsNonSyncCtxImpl { |
| fn on_event(&mut self, event: netstack3_core::IpDeviceEvent<DeviceId, I>) { |
| let (device, event) = match event { |
| netstack3_core::IpDeviceEvent::AddressAdded { device, addr, state } => ( |
| device, |
| InterfaceUpdate::AddressAdded { |
| addr: addr.into(), |
| initial_state: interfaces_watcher::AddressState { |
| valid_until: zx::Time::INFINITE, |
| assignment_state: state, |
| }, |
| }, |
| ), |
| netstack3_core::IpDeviceEvent::AddressRemoved { device, addr } => { |
| (device, InterfaceUpdate::AddressRemoved(addr.into())) |
| } |
| netstack3_core::IpDeviceEvent::AddressStateChanged { device, addr, state } => ( |
| device, |
| InterfaceUpdate::AddressAssignmentStateChanged { |
| addr: addr.into(), |
| new_state: state, |
| }, |
| ), |
| }; |
| self.notify_interface_update(device, event); |
| } |
| } |
| |
| impl<I: Ip> EventContext<netstack3_core::IpLayerEvent<DeviceId, I>> for BindingsNonSyncCtxImpl { |
| fn on_event(&mut self, event: netstack3_core::IpLayerEvent<DeviceId, I>) { |
| let (device, subnet, has_default_route) = match event { |
| netstack3_core::IpLayerEvent::DeviceRouteAdded { device, subnet } => { |
| (device, subnet, true) |
| } |
| netstack3_core::IpLayerEvent::DeviceRouteRemoved { device, subnet } => { |
| (device, subnet, false) |
| } |
| }; |
| // We only care about the default route. |
| if subnet.prefix() != 0 || subnet.network() != I::UNSPECIFIED_ADDRESS { |
| return; |
| } |
| self.notify_interface_update( |
| device, |
| InterfaceUpdate::DefaultRouteChanged { version: I::VERSION, has_default_route }, |
| ); |
| } |
| } |
| |
| impl EventContext<netstack3_core::DadEvent<DeviceId>> for BindingsNonSyncCtxImpl { |
| fn on_event(&mut self, event: netstack3_core::DadEvent<DeviceId>) { |
| match event { |
| netstack3_core::DadEvent::AddressAssigned { device, addr } => { |
| self.on_event(netstack3_core::IpDeviceEvent::<_, Ipv6>::AddressStateChanged { |
| device, |
| addr: *addr, |
| state: netstack3_core::IpAddressState::Assigned, |
| }) |
| } |
| } |
| } |
| } |
| |
| impl EventContext<netstack3_core::Ipv6RouteDiscoveryEvent<DeviceId>> for BindingsNonSyncCtxImpl { |
| fn on_event(&mut self, _event: netstack3_core::Ipv6RouteDiscoveryEvent<DeviceId>) { |
| // TODO(https://fxbug.dev/97203): Update forwarding table in response to |
| // the event. |
| } |
| } |
| |
| impl BindingsNonSyncCtxImpl { |
| fn notify_interface_update(&self, device: DeviceId, event: InterfaceUpdate) { |
| self.devices |
| .get_core_device(device) |
| .unwrap_or_else(|| panic!("issued event {:?} for deleted device {:?}", event, device)) |
| .info() |
| .common_info() |
| .events |
| .notify(event) |
| .expect("interfaces worker closed"); |
| } |
| } |
| |
| trait MutableDeviceState { |
| /// Invoke a function on the state associated with the device `id`. |
| fn update_device_state<F: FnOnce(&mut DeviceInfo)>(&mut self, id: u64, f: F); |
| } |
| |
| impl<NonSyncCtx> MutableDeviceState for Ctx<NonSyncCtx> |
| where |
| NonSyncCtx: NonSyncContext + DeviceStatusNotifier + AsRef<Devices> + AsMut<Devices>, |
| { |
| fn update_device_state<F: FnOnce(&mut DeviceInfo)>(&mut self, id: u64, f: F) { |
| if let Some(device_info) = self.non_sync_ctx.as_mut().get_device_mut(id) { |
| f(device_info); |
| self.non_sync_ctx.device_status_changed(id) |
| } |
| } |
| } |
| |
| trait InterfaceControl { |
| /// Enables an interface. |
| /// |
| /// Both `admin_enabled` and `phy_up` must be true for the interface to be |
| /// enabled. |
| fn enable_interface(&mut self, id: u64) -> Result<(), fidl_net_stack::Error>; |
| |
| /// Disables an interface. |
| /// |
| /// Either an Admin (fidl) or Phy change can disable an interface. |
| fn disable_interface(&mut self, id: u64) -> Result<(), fidl_net_stack::Error>; |
| } |
| |
| fn set_interface_enabled<NonSyncCtx: NonSyncContext + AsRef<Devices> + AsMut<Devices>>( |
| Ctx { sync_ctx, non_sync_ctx }: &mut Ctx<NonSyncCtx>, |
| id: u64, |
| should_enable: bool, |
| ) -> Result<(), fidl_net_stack::Error> { |
| let device = non_sync_ctx.as_mut().get_device_mut(id).ok_or(fidl_net_stack::Error::NotFound)?; |
| let core_id = device.core_id(); |
| |
| let (dev_enabled, events) = match device.info_mut() { |
| DeviceSpecificInfo::Ethernet(EthernetInfo { |
| common_info: CommonInfo { admin_enabled, mtu: _, events, name: _ }, |
| client: _, |
| mac: _, |
| features: _, |
| phy_up, |
| }) |
| | DeviceSpecificInfo::Netdevice(NetdeviceInfo { |
| common_info: CommonInfo { admin_enabled, mtu: _, events, name: _ }, |
| handler: _, |
| mac: _, |
| phy_up, |
| }) => (*admin_enabled && *phy_up, events), |
| DeviceSpecificInfo::Loopback(LoopbackInfo { |
| common_info: CommonInfo { admin_enabled, mtu: _, events, name: _ }, |
| }) => (*admin_enabled, events), |
| }; |
| |
| if should_enable { |
| // We want to enable the interface, but its device is considered |
| // disabled so we do nothing further. |
| // |
| // This can happen when the interface was set to be administratively up |
| // but the phy is down. |
| if !dev_enabled { |
| return Ok(()); |
| } |
| } else { |
| assert!(!dev_enabled, "caller attemped to disable an interface that is considered enabled"); |
| } |
| |
| events |
| .notify(InterfaceUpdate::OnlineChanged(should_enable)) |
| .expect("interfaces worker not running"); |
| |
| update_ipv4_configuration(sync_ctx, non_sync_ctx, core_id, |config| { |
| config.ip_config.ip_enabled = should_enable; |
| }); |
| update_ipv6_configuration(sync_ctx, non_sync_ctx, core_id, |config| { |
| config.ip_config.ip_enabled = should_enable; |
| }); |
| |
| Ok(()) |
| } |
| |
| impl<NonSyncCtx> InterfaceControl for Ctx<NonSyncCtx> |
| where |
| NonSyncCtx: NonSyncContext + AsRef<Devices> + AsMut<Devices>, |
| { |
| fn enable_interface(&mut self, id: u64) -> Result<(), fidl_net_stack::Error> { |
| set_interface_enabled(self, id, true /* should_enable */) |
| } |
| |
| fn disable_interface(&mut self, id: u64) -> Result<(), fidl_net_stack::Error> { |
| set_interface_enabled(self, id, false /* should_enable */) |
| } |
| } |
| |
| type NetstackContext = Arc<Mutex<Ctx<BindingsNonSyncCtxImpl>>>; |
| |
| /// The netstack. |
| /// |
| /// Provides the entry point for creating a netstack to be served as a |
| /// component. |
| #[derive(Clone)] |
| pub struct Netstack { |
| ctx: NetstackContext, |
| interfaces_event_sink: interfaces_watcher::WorkerInterfaceSink, |
| } |
| |
| /// Contains the information needed to start serving a network stack over FIDL. |
| pub struct NetstackSeed { |
| netstack: Netstack, |
| interfaces_worker: interfaces_watcher::Worker, |
| interfaces_watcher_sink: interfaces_watcher::WorkerWatcherSink, |
| } |
| |
| impl Default for NetstackSeed { |
| fn default() -> Self { |
| let (interfaces_worker, interfaces_watcher_sink, interfaces_event_sink) = |
| interfaces_watcher::Worker::new(); |
| Self { |
| netstack: Netstack { ctx: Default::default(), interfaces_event_sink }, |
| interfaces_worker, |
| interfaces_watcher_sink, |
| } |
| } |
| } |
| |
| impl LockableContext for Netstack { |
| type NonSyncCtx = BindingsNonSyncCtxImpl; |
| } |
| |
| impl InterfaceEventProducerFactory for Netstack { |
| fn create_interface_event_producer( |
| &self, |
| id: BindingId, |
| properties: InterfaceProperties, |
| ) -> InterfaceEventProducer { |
| self.interfaces_event_sink |
| .add_interface(id, properties) |
| .expect("interface worker not running") |
| } |
| } |
| |
| enum Service { |
| Stack(fidl_fuchsia_net_stack::StackRequestStream), |
| Socket(fidl_fuchsia_posix_socket::ProviderRequestStream), |
| Interfaces(fidl_fuchsia_net_interfaces::StateRequestStream), |
| InterfacesAdmin(fidl_fuchsia_net_interfaces_admin::InstallerRequestStream), |
| Debug(fidl_fuchsia_net_debug::InterfacesRequestStream), |
| } |
| |
| enum WorkItem { |
| Incoming(Service), |
| Task(fasync::Task<()>), |
| } |
| |
| trait RequestStreamExt: RequestStream { |
| fn serve_with<F, Fut, E>(self, f: F) -> futures::future::Map<Fut, fn(Result<(), E>) -> ()> |
| where |
| E: std::error::Error, |
| F: FnOnce(Self) -> Fut, |
| Fut: Future<Output = Result<(), E>>; |
| } |
| |
| impl<D: DiscoverableProtocolMarker, S: RequestStream<Protocol = D>> RequestStreamExt for S { |
| fn serve_with<F, Fut, E>(self, f: F) -> futures::future::Map<Fut, fn(Result<(), E>) -> ()> |
| where |
| E: std::error::Error, |
| F: FnOnce(Self) -> Fut, |
| Fut: Future<Output = Result<(), E>>, |
| { |
| f(self).map(|res| res.unwrap_or_else(|err| error!("{} error: {}", D::PROTOCOL_NAME, err))) |
| } |
| } |
| |
| impl NetstackSeed { |
| /// Consumes the netstack and starts serving all the FIDL services it |
| /// implements to the outgoing service directory. |
| pub async fn serve(self) -> Result<(), anyhow::Error> { |
| use anyhow::Context as _; |
| |
| debug!("Serving netstack"); |
| |
| let Self { netstack, interfaces_worker, interfaces_watcher_sink } = self; |
| |
| { |
| let mut ctx = netstack.lock().await; |
| let Ctx { sync_ctx, non_sync_ctx } = ctx.deref_mut(); |
| |
| // Add and initialize the loopback interface with the IPv4 and IPv6 |
| // loopback addresses and on-link routes to the loopback subnets. |
| let loopback = netstack3_core::add_loopback_device(sync_ctx, DEFAULT_LOOPBACK_MTU) |
| .expect("error adding loopback device"); |
| let devices: &mut Devices = non_sync_ctx.as_mut(); |
| let _binding_id: u64 = devices |
| .add_device(loopback, |id| { |
| const LOOPBACK_NAME: &'static str = "lo"; |
| let events = netstack.create_interface_event_producer( |
| id, |
| InterfaceProperties { |
| name: LOOPBACK_NAME.to_string(), |
| device_class: fidl_fuchsia_net_interfaces::DeviceClass::Loopback( |
| fidl_fuchsia_net_interfaces::Empty {}, |
| ), |
| }, |
| ); |
| events |
| .notify(InterfaceUpdate::OnlineChanged(true)) |
| .expect("interfaces worker not running"); |
| DeviceSpecificInfo::Loopback(LoopbackInfo { |
| common_info: CommonInfo { |
| mtu: DEFAULT_LOOPBACK_MTU, |
| admin_enabled: true, |
| events, |
| name: LOOPBACK_NAME.to_string(), |
| }, |
| }) |
| }) |
| .expect("error adding loopback device"); |
| // Don't need DAD and IGMP/MLD on loopback. |
| update_ipv4_configuration(sync_ctx, non_sync_ctx, loopback, |config| { |
| *config = Ipv4DeviceConfiguration { |
| ip_config: IpDeviceConfiguration { ip_enabled: true, gmp_enabled: false }, |
| }; |
| }); |
| update_ipv6_configuration(sync_ctx, non_sync_ctx, loopback, |config| { |
| *config = Ipv6DeviceConfiguration { |
| dad_transmits: None, |
| max_router_solicitations: None, |
| slaac_config: SlaacConfiguration { |
| enable_stable_addresses: true, |
| temporary_address_configuration: None, |
| }, |
| ip_config: IpDeviceConfiguration { ip_enabled: true, gmp_enabled: false }, |
| }; |
| }); |
| add_ip_addr_subnet( |
| sync_ctx, |
| non_sync_ctx, |
| loopback, |
| AddrSubnetEither::V4( |
| AddrSubnet::from_witness( |
| Ipv4::LOOPBACK_ADDRESS, |
| Ipv4::LOOPBACK_SUBNET.prefix(), |
| ) |
| .expect("error creating IPv4 loopback AddrSub"), |
| ), |
| ) |
| .expect("error adding IPv4 loopback address"); |
| add_route( |
| sync_ctx, |
| non_sync_ctx, |
| AddableEntryEither::new(Ipv4::LOOPBACK_SUBNET.into(), Some(loopback), None) |
| .expect("error creating IPv4 route entry"), |
| ) |
| .expect("error adding IPv4 loopback on-link subnet route"); |
| add_ip_addr_subnet( |
| sync_ctx, |
| non_sync_ctx, |
| loopback, |
| AddrSubnetEither::V6( |
| AddrSubnet::from_witness( |
| Ipv6::LOOPBACK_ADDRESS, |
| Ipv6::LOOPBACK_SUBNET.prefix(), |
| ) |
| .expect("error creating IPv6 loopback AddrSub"), |
| ), |
| ) |
| .expect("error adding IPv6 loopback address"); |
| add_route( |
| sync_ctx, |
| non_sync_ctx, |
| AddableEntryEither::new(Ipv6::LOOPBACK_SUBNET.into(), Some(loopback), None) |
| .expect("error creating IPv6 route entry"), |
| ) |
| .expect("error adding IPv6 loopback on-link subnet route"); |
| |
| // Start servicing timers. |
| let BindingsNonSyncCtxImpl { |
| rng: _, |
| timers, |
| devices: _, |
| icmp_echo_sockets: _, |
| udp_sockets: _, |
| } = non_sync_ctx; |
| timers.spawn(netstack.clone()); |
| } |
| |
| let interfaces_worker_task = fuchsia_async::Task::spawn(async move { |
| let result = interfaces_worker.run().await; |
| // The worker is not expected to end for the lifetime of the stack. |
| panic!("interfaces worker finished unexpectedly {:?}", result); |
| }); |
| |
| let mut fs = ServiceFs::new_local(); |
| let _: &mut ServiceFsDir<'_, _> = fs |
| .dir("svc") |
| .add_fidl_service(Service::Debug) |
| .add_fidl_service(Service::Stack) |
| .add_fidl_service(Service::Socket) |
| .add_fidl_service(Service::Interfaces) |
| .add_fidl_service(Service::InterfacesAdmin); |
| |
| let services = fs.take_and_serve_directory_handle().context("directory handle")?; |
| |
| // Buffer size doesn't matter much, we're just trying to reduce |
| // allocations. |
| const TASK_CHANNEL_BUFFER_SIZE: usize = 16; |
| let (task_sink, task_stream) = mpsc::channel(TASK_CHANNEL_BUFFER_SIZE); |
| let work_items = futures::stream::select( |
| services.map(WorkItem::Incoming), |
| task_stream.map(WorkItem::Task), |
| ); |
| let work_items_fut = work_items.for_each_concurrent(None, |wi| async { |
| match wi { |
| WorkItem::Incoming(Service::Stack(stack)) => { |
| stack |
| .serve_with(|rs| { |
| stack_fidl_worker::StackFidlWorker::serve(netstack.clone(), rs) |
| }) |
| .await |
| } |
| WorkItem::Incoming(Service::Socket(socket)) => { |
| socket.serve_with(|rs| socket::serve(netstack.clone(), rs)).await |
| } |
| WorkItem::Incoming(Service::Interfaces(interfaces)) => { |
| interfaces |
| .serve_with(|rs| { |
| interfaces_watcher::serve(rs, interfaces_watcher_sink.clone()) |
| }) |
| .await |
| } |
| WorkItem::Incoming(Service::InterfacesAdmin(installer)) => { |
| log::debug!( |
| "serving {}", |
| fidl_fuchsia_net_interfaces_admin::InstallerMarker::PROTOCOL_NAME |
| ); |
| interfaces_admin::serve(netstack.clone(), installer) |
| .map_err(anyhow::Error::from) |
| .forward(task_sink.clone().sink_map_err(anyhow::Error::from)) |
| .await |
| .unwrap_or_else(|e| { |
| log::warn!( |
| "error serving {}: {:?}", |
| fidl_fuchsia_net_interfaces_admin::InstallerMarker::PROTOCOL_NAME, |
| e |
| ) |
| }) |
| } |
| WorkItem::Incoming(Service::Debug(debug)) => { |
| debug.serve_with(|rs| debug_fidl_worker::serve(netstack.clone(), rs)).await |
| } |
| WorkItem::Task(task) => task.await, |
| } |
| }); |
| |
| let ((), ()) = futures::future::join(work_items_fut, interfaces_worker_task).await; |
| debug!("Services stream finished"); |
| Ok(()) |
| } |
| } |
