netlink/

lib.rs

// Copyright 2023 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.

//! An implementation of Linux's Netlink API for Fuchsia.
//!
//! Netlink is a socket-based API provided by Linux that user space applications
//! can use to interact with the kernel. The API is split up into several
//! protocol families each offering different functionality. This crate targets
//! the implementation of families related to networking.

#![warn(missing_docs, unused)]

mod client;
pub mod interfaces;
pub(crate) mod logging;
pub mod messaging;
pub mod multicast_groups;
mod nduseropt;
pub mod neighbors;
mod netlink_packet;
pub mod protocol_family;
pub(crate) mod route_eventloop;
pub(crate) mod route_tables;
pub mod routes;
mod rules;
pub(crate) mod util;

use std::num::NonZeroU64;

use fidl_fuchsia_net_interfaces as fnet_interfaces;
use fidl_fuchsia_net_ndp as fnet_ndp;
use fidl_fuchsia_net_neighbor as fnet_neighbor;
use fidl_fuchsia_net_root as fnet_root;
use fidl_fuchsia_net_routes as fnet_routes;
use fidl_fuchsia_net_routes_admin as fnet_routes_admin;
use fidl_fuchsia_net_routes_ext as fnet_routes_ext;
use fidl_fuchsia_net_sockets as fnet_sockets;
use fuchsia_component::client::connect_to_protocol;
use futures::StreamExt as _;
use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
use futures::channel::oneshot;
use net_types::ip::{Ipv4, Ipv6};
use netlink_packet_route::RouteNetlinkMessage;
use netlink_packet_sock_diag::{SockDiagRequest, SockDiagResponse};
use protocol_family::route::NetlinkRouteNotifiedGroup;

use crate::client::{AsyncWorkItem, ClientIdGenerator, ClientTable, InternalClient};
use crate::logging::{log_debug, log_warn};
use crate::messaging::{NetlinkContext, UnvalidatedNetlinkMessage as _, ValidationError};
pub use crate::netlink_packet::errno::Errno;
use crate::protocol_family::route::{
    NetlinkRoute, NetlinkRouteClient, NetlinkRouteRequestHandler, RouteAsyncWork,
};
use crate::protocol_family::sock_diag::{
    NetlinkSockDiag, NetlinkSockDiagClient, NetlinkSockDiagRequestHandler, SockDiagEventLoop,
};
use crate::protocol_family::{NetlinkFamilyRequestHandler as _, ProtocolFamily};
use crate::route_eventloop::RouteEventLoop;

/// The tag added to all logs generated by this crate.
pub const NETLINK_LOG_TAG: &'static str = "netlink";

/// Selects the interface for the sysctl.
#[derive(Debug, Clone, Copy)]
pub enum SysctlInterfaceSelector {
    /// "all" interfaces.
    ///
    /// This is supposed to change all interfaces' settings, but this is a
    /// lie for most of the sysctls, they have no effect at all when written.
    All,
    /// "default" interface, all interface created after this write will inherit the value.
    Default,
    /// The id of the interface to change.
    Id(NonZeroU64),
}

/// The implementation of the Netlink protocol suite.
pub struct Netlink<C: NetlinkContext> {
    /// Generator of new Client IDs.
    id_generator: ClientIdGenerator,
    /// Sender to attach new `NETLINK_ROUTE` clients to the Netlink worker.
    route_client_sender: UnboundedSender<ClientWithReceiver<C, NetlinkRoute>>,
    /// Sender to send other async work items to the Netlink worker.
    route_async_work_sink: mpsc::UnboundedSender<AsyncWorkItem<NetlinkRoute>>,
    /// Sender to attach new `NETLINK_SOCK_DIAG` clients to the Netlink worker.
    sock_diag_client_sender: UnboundedSender<ClientWithReceiver<C, NetlinkSockDiag>>,
    /// Sender to send other async work items to the Netlink worker.
    sock_diag_async_work_sink: mpsc::UnboundedSender<AsyncWorkItem<NetlinkSockDiag>>,
}

impl<C: NetlinkContext> Netlink<C> {
    /// Returns a newly instantiated [`Netlink`] and parameters used to start the
    /// asynchronous worker.
    ///
    /// Caller is expected to run the worker by calling `run_netlink_worker()`.
    pub fn new<H: interfaces::InterfacesHandler>(
        interfaces_handler: H,
    ) -> (Self, NetlinkWorkerParams<H, C>) {
        let (route_client_sender, route_client_receiver) = mpsc::unbounded();
        let (route_async_work_sink, async_work_receiver) = mpsc::unbounded();
        let (sock_diag_client_sender, sock_diag_client_receiver) = mpsc::unbounded();
        let (sock_diag_async_work_sink, sock_diag_async_work_receiver) = mpsc::unbounded();
        (
            Netlink {
                id_generator: ClientIdGenerator::default(),
                route_client_sender,
                sock_diag_client_sender,
                route_async_work_sink,
                sock_diag_async_work_sink,
            },
            NetlinkWorkerParams {
                interfaces_handler,
                route_client_receiver,
                route_async_work_receiver: async_work_receiver,
                sock_diag_client_receiver,
                sock_diag_async_work_receiver,
            },
        )
    }

    /// Writes the accept_ra_rt_table sysctl for the selected interface.
    pub fn write_accept_ra_rt_table(
        &self,
        interface: SysctlInterfaceSelector,
        value: i32,
    ) -> Result<(), SysctlError> {
        let (responder, receiver) = oneshot_sync::channel();
        self.route_async_work_sink
            .unbounded_send(AsyncWorkItem::Inner(RouteAsyncWork::SetAcceptRaRtTable {
                interface,
                value: value.into(),
                responder,
            }))
            .map_err(|_| SysctlError::Disconnected)?;
        receiver.receive().map_err(|_| SysctlError::Disconnected)?
    }

    /// Reads the accept_ra_rt_table sysctl for the selected interface.
    pub fn read_accept_ra_rt_table(
        &self,
        interface: SysctlInterfaceSelector,
    ) -> Result<i32, SysctlError> {
        let (responder, receiver) = oneshot_sync::channel();
        self.route_async_work_sink
            .unbounded_send(AsyncWorkItem::Inner(RouteAsyncWork::GetAcceptRaRtTable {
                interface,
                responder,
            }))
            .map_err(|_| SysctlError::Disconnected)?;
        Ok(receiver.receive().map_err(|_| SysctlError::Disconnected)??.into())
    }

    /// Creates a new client of the `NETLINK_ROUTE` protocol family.
    ///
    /// `sender` is used by Netlink to send messages to the client.
    /// `receiver` is used by Netlink to receive messages from the client.
    ///
    /// Closing the `receiver` will close this client, disconnecting `sender`.
    pub fn new_route_client(
        &self,
        sender: C::Sender<RouteNetlinkMessage>,
        receiver: C::Receiver<RouteNetlinkMessage>,
    ) -> Result<NetlinkRouteClient, NewClientError> {
        let Netlink {
            id_generator,
            route_client_sender,
            route_async_work_sink,
            sock_diag_client_sender: _,
            sock_diag_async_work_sink: _,
        } = self;
        let (external_client, internal_client) = client::new_client_pair::<NetlinkRoute, _>(
            id_generator.new_id(),
            sender,
            route_async_work_sink.clone(),
        );
        route_client_sender
            .unbounded_send(ClientWithReceiver { client: internal_client, receiver })
            .map_err(|e| {
                // Sending on an `UnboundedSender` can never fail with `is_full()`.
                debug_assert!(e.is_disconnected());
                NewClientError::Disconnected
            })?;
        Ok(NetlinkRouteClient(external_client))
    }

    /// Creates a new client of the `NETLINK_SOCK_DIAG` protocol family.
    ///
    /// `sender` is used by Netlink to send messages to the client.
    /// `receiver` is used by Netlink to receive messages from the client.
    ///
    /// Closing the `receiver` will close this client, disconnecting `sender`.
    pub fn new_sock_diag_client(
        &self,
        sender: C::Sender<SockDiagResponse>,
        receiver: C::Receiver<SockDiagRequest>,
    ) -> Result<NetlinkSockDiagClient, NewClientError> {
        let Netlink {
            id_generator,
            route_client_sender: _,
            route_async_work_sink: _,
            sock_diag_client_sender,
            sock_diag_async_work_sink,
        } = self;
        let (external_client, internal_client) = client::new_client_pair::<NetlinkSockDiag, _>(
            id_generator.new_id(),
            sender,
            sock_diag_async_work_sink.clone(),
        );
        sock_diag_client_sender
            .unbounded_send(ClientWithReceiver { client: internal_client, receiver })
            .map_err(|e| {
                // Sending on an `UnboundedSender` can never fail with `is_full()`.
                debug_assert!(e.is_disconnected());
                NewClientError::Disconnected
            })?;
        Ok(NetlinkSockDiagClient(external_client))
    }
}

/// A wrapper to hold an [`InternalClient`], and its [`Receiver`] of requests.
struct ClientWithReceiver<C: NetlinkContext, F: ProtocolFamily> {
    client: InternalClient<F, C::Sender<F::Response>>,
    receiver: C::Receiver<F::Request>,
}

/// The possible error types when instantiating a new client.
#[derive(Debug)]
pub enum NewClientError {
    /// The [`Netlink`] is disconnected from its associated worker, perhaps as a
    /// result of dropping the worker.
    Disconnected,
}

/// The possible error types when trying to access a sysctl.
#[derive(Debug)]
pub enum SysctlError {
    /// The [`Netlink`] is disconnected from its associated worker.
    Disconnected,
    /// The interface went away.
    NoInterface,
    /// The written value requests for an unsupported operation.
    Unsupported,
}

/// Parameters used to start the Netlink asynchronous worker.
pub struct NetlinkWorkerParams<H, C: NetlinkContext> {
    interfaces_handler: H,
    /// Receiver of newly created `NETLINK_ROUTE` clients.
    route_client_receiver: UnboundedReceiver<ClientWithReceiver<C, NetlinkRoute>>,
    route_async_work_receiver:
        futures::channel::mpsc::UnboundedReceiver<AsyncWorkItem<NetlinkRoute>>,
    /// Receiver of newly created `NETLINK_SOCK_DIAG` clients.
    sock_diag_client_receiver: UnboundedReceiver<ClientWithReceiver<C, NetlinkSockDiag>>,
    sock_diag_async_work_receiver:
        futures::channel::mpsc::UnboundedReceiver<AsyncWorkItem<NetlinkSockDiag>>,
}

/// All of the protocols that the netlink worker connects to.
#[allow(missing_docs)]
pub struct NetlinkWorkerDiscoverableProtocols {
    pub root_interfaces: fnet_root::InterfacesProxy,
    pub interfaces_state: fnet_interfaces::StateProxy,
    pub v4_routes_state: fnet_routes::StateV4Proxy,
    pub v6_routes_state: fnet_routes::StateV6Proxy,
    pub v4_main_route_table: fnet_routes_admin::RouteTableV4Proxy,
    pub v6_main_route_table: fnet_routes_admin::RouteTableV6Proxy,
    pub v4_route_table_provider: fnet_routes_admin::RouteTableProviderV4Proxy,
    pub v6_route_table_provider: fnet_routes_admin::RouteTableProviderV6Proxy,
    pub v4_rule_table: fnet_routes_admin::RuleTableV4Proxy,
    pub v6_rule_table: fnet_routes_admin::RuleTableV6Proxy,
    pub ndp_option_watcher_provider: fnet_ndp::RouterAdvertisementOptionWatcherProviderProxy,
    pub socket_diagnostics: fnet_sockets::DiagnosticsProxy,
    pub socket_control: fnet_sockets::ControlProxy,
    pub neighbors_view: fnet_neighbor::ViewProxy,
    pub neighbors_controller: fnet_neighbor::ControllerProxy,
}

impl NetlinkWorkerDiscoverableProtocols {
    fn from_environment() -> Self {
        let root_interfaces = connect_to_protocol::<fnet_root::InterfacesMarker>()
            .expect("connect to fuchsia.net.root.Interfaces");
        let interfaces_state = connect_to_protocol::<fnet_interfaces::StateMarker>()
            .expect("connect to fuchsia.net.interfaces.State");
        let v4_routes_state =
            connect_to_protocol::<<Ipv4 as fnet_routes_ext::FidlRouteIpExt>::StateMarker>()
                .expect("connect to fuchsia.net.routes.StateV4");
        let v6_routes_state =
            connect_to_protocol::<<Ipv6 as fnet_routes_ext::FidlRouteIpExt>::StateMarker>()
                .expect("connect to fuchsia.net.routes.StateV6");
        let v4_main_route_table = connect_to_protocol::<
            <Ipv4 as fnet_routes_ext::admin::FidlRouteAdminIpExt>::RouteTableMarker,
        >()
        .expect("connect to fuchsia.net.routes.admin.RouteTableV4");
        let v6_main_route_table = connect_to_protocol::<
            <Ipv6 as fnet_routes_ext::admin::FidlRouteAdminIpExt>::RouteTableMarker,
        >()
        .expect("connect to fuchsia.net.routes.admin.RouteTableV6");
        let v4_route_table_provider = connect_to_protocol::<
            <Ipv4 as fnet_routes_ext::admin::FidlRouteAdminIpExt>::RouteTableProviderMarker,
        >()
        .expect("connect to fuchsia.net.routes.admin.RouteTableProviderV4");
        let v6_route_table_provider = connect_to_protocol::<
            <Ipv6 as fnet_routes_ext::admin::FidlRouteAdminIpExt>::RouteTableProviderMarker,
        >()
        .expect("connect to fuchsia.net.routes.admin.RouteTableProviderV6");
        let v4_rule_table = connect_to_protocol::<
            <Ipv4 as fnet_routes_ext::rules::FidlRuleAdminIpExt>::RuleTableMarker,
        >()
        .expect("connect to fuchsia.net.routes.admin.RuleTableV4");
        let v6_rule_table = connect_to_protocol::<
            <Ipv6 as fnet_routes_ext::rules::FidlRuleAdminIpExt>::RuleTableMarker,
        >()
        .expect("connect to fuchsia.net.routes.admin.RuleTableV6");
        let ndp_option_watcher_provider =
            connect_to_protocol::<fnet_ndp::RouterAdvertisementOptionWatcherProviderMarker>()
                .expect("connect to fuchsia.net.ndp.RouterAdvertisementOptionWatcherProvider");
        let socket_diagnostics = connect_to_protocol::<fnet_sockets::DiagnosticsMarker>()
            .expect("connect to fuchsia.net.sockets.Diagnostics");
        let socket_control = connect_to_protocol::<fnet_sockets::ControlMarker>()
            .expect("connect to fuchsia.net.sockets.Control");
        let neighbors_view = connect_to_protocol::<fnet_neighbor::ViewMarker>()
            .expect("connect to fuchsia.net.neighbor.View");
        let neighbors_controller = connect_to_protocol::<fnet_neighbor::ControllerMarker>()
            .expect("connect to fuchsia.net.neighbor.Controller");

        Self {
            root_interfaces,
            interfaces_state,
            v4_routes_state,
            v6_routes_state,
            v4_main_route_table,
            v6_main_route_table,
            v4_route_table_provider,
            v6_route_table_provider,
            v4_rule_table,
            v6_rule_table,
            ndp_option_watcher_provider,
            socket_diagnostics,
            socket_control,
            neighbors_view,
            neighbors_controller,
        }
    }
}

/// The worker encompassing all asynchronous Netlink work.
///
/// The worker is never expected to complete.
///
/// `protocols` is taken as a closure because we need to avoid creating asynchronous FIDL proxies
/// until an executor is running, so it's helpful to defer creation until the event loop starts
/// running.
///
/// # Panics
///
/// Panics if a non-recoverable error is encountered by the worker. For example,
/// a FIDL error on one of the FIDL connections with the netstack.
pub async fn run_netlink_worker<H: interfaces::InterfacesHandler, C: NetlinkContext>(
    params: NetlinkWorkerParams<H, C>,
    access_control: C::AccessControl<'_>,
) {
    run_netlink_worker_with_protocols(
        params,
        NetlinkWorkerDiscoverableProtocols::from_environment(),
        None,
        access_control,
    )
    .await;
}

/// Same as `run_netlink_worker()`, but allows to pass custom
/// `NetlinkWorkerDiscoverableProtocols`.
pub async fn run_netlink_worker_with_protocols<
    H: interfaces::InterfacesHandler,
    C: NetlinkContext,
>(
    params: NetlinkWorkerParams<H, C>,
    protocols: NetlinkWorkerDiscoverableProtocols,
    on_route_initialized: Option<oneshot::Sender<()>>,
    access_control: C::AccessControl<'_>,
) {
    let NetlinkWorkerParams {
        interfaces_handler,
        route_client_receiver,
        route_async_work_receiver,
        sock_diag_client_receiver,
        sock_diag_async_work_receiver,
    } = params;

    let NetlinkWorkerDiscoverableProtocols {
        root_interfaces,
        interfaces_state,
        v4_routes_state,
        v6_routes_state,
        v4_main_route_table,
        v6_main_route_table,
        v4_route_table_provider,
        v6_route_table_provider,
        v4_rule_table,
        v6_rule_table,
        ndp_option_watcher_provider,
        socket_diagnostics,
        socket_control,
        neighbors_view,
        neighbors_controller,
    } = protocols;

    let route_clients = ClientTable::default();
    let (route_request_sink, route_request_stream) = mpsc::channel(1);

    let route_event_loop = {
        let route_clients = route_clients.clone();
        async move {
            let event_loop: RouteEventLoop<H, C::Sender<_>> = RouteEventLoop {
                interfaces_proxy: root_interfaces,
                interfaces_state_proxy: interfaces_state,
                v4_routes_state,
                v6_routes_state,
                v4_main_route_table,
                v6_main_route_table,
                v4_route_table_provider,
                v6_route_table_provider,
                v4_rule_table,
                v6_rule_table,
                ndp_option_watcher_provider,
                neighbors_view,
                neighbors_controller,
                route_clients,
                request_stream: route_request_stream,
                interfaces_handler,
                async_work_receiver: route_async_work_receiver,
            };

            event_loop.run(on_route_initialized).await;
        }
    };

    let route_client_receiver_loop = {
        let access_control = access_control.clone();
        async move {
            // Accept new NETLINK_ROUTE clients.
            connect_new_clients::<C, NetlinkRoute>(
                route_clients,
                route_client_receiver,
                NetlinkRouteRequestHandler { unified_request_sink: route_request_sink },
                access_control,
            )
            .await;
            panic!("route_client_receiver stream unexpectedly finished");
        }
    };

    let sock_diag_clients = ClientTable::default();
    let (sock_diag_request_sink, sock_diag_request_stream) = mpsc::channel(1);

    let sock_diag_event_loop = {
        let sock_diag_clients = sock_diag_clients.clone();
        async move {
            SockDiagEventLoop::new(
                socket_diagnostics,
                socket_control,
                sock_diag_request_stream,
                sock_diag_async_work_receiver,
                sock_diag_clients,
            )
            .run()
            .await;
        }
    };

    let sock_diag_client_receiver_loop = async move {
        // Accept new NETLINK_SOCK_DIAG clients.
        connect_new_clients::<C, NetlinkSockDiag>(
            sock_diag_clients,
            sock_diag_client_receiver,
            NetlinkSockDiagRequestHandler { sock_diag_request_sink },
            access_control,
        )
        .await;
        panic!("sock_diag_client_receiver stream unexpectedly finished");
    };

    futures::future::join4(
        route_event_loop,
        route_client_receiver_loop,
        sock_diag_event_loop,
        sock_diag_client_receiver_loop,
    )
    .await;
}

/// Receives clients from the given receiver, adding them to the given table.
///
/// A "Request Handler" Task will be spawned for each received client. The given
/// `request_handler_impl` defines how the requests will be handled.
async fn connect_new_clients<C: NetlinkContext, F: ProtocolFamily>(
    client_table: ClientTable<F, C::Sender<F::Response>>,
    client_receiver: UnboundedReceiver<ClientWithReceiver<C, F>>,
    request_handler_impl: F::RequestHandler<C::Sender<F::Response>>,
    access_control: C::AccessControl<'_>,
) {
    client_receiver
        // Drive each client concurrently with `for_each_concurrent`.
        .for_each_concurrent(None, async |ClientWithReceiver { client, receiver }| {
            client_table.add_client(client.clone());
            let client = run_client_request_handler::<C, F>(
                client,
                receiver,
                request_handler_impl.clone(),
                access_control.clone(),
            )
            .await;
            client_table.remove_client(client);
        })
        .await;
}

/// Reads messages from the `receiver` and handles them using the `handler`.
///
/// The task terminates when the underlying `Receiver` closes, yielding the
/// original client.
async fn run_client_request_handler<C: NetlinkContext, F: ProtocolFamily>(
    client: InternalClient<F, C::Sender<F::Response>>,
    receiver: C::Receiver<F::Request>,
    handler: F::RequestHandler<C::Sender<F::Response>>,
    access_control: C::AccessControl<'_>,
) -> InternalClient<F, C::Sender<F::Response>> {
    // State needed to handle an individual request, that is cycled through the
    // `fold` combinator below.
    struct FoldState<C, H, P> {
        client: C,
        handler: H,
        access_control: P,
    }

    // Use `fold` for two reasons. First, it processes requests serially,
    // ensuring requests are handled in order. Second, it allows us to
    // "hand-off" the client/handler from one request to the other, avoiding
    // copies for each request.
    let FoldState { client, handler: _, access_control: _ } = receiver
        .fold(
            FoldState { client, handler, access_control },
            |FoldState { mut client, mut handler, access_control }, req| async {
                match req.validate_creds_and_get_message(&access_control) {
                    Ok(req) => {
                        log_debug!("{} Received request: {:?}", client, req);
                        handler.handle_request(req, &mut client).await
                    }
                    Err(e) => {
                        match &e {
                            ValidationError::Parse(e) => {
                                log_warn!("{client} failed to parse netlink message: {e:?}");
                            }
                            p @ ValidationError::Permission { .. } => {
                                log_debug!("{client} permission check failed {p:?}")
                            }
                        }
                        if let Some(rsp) = e.into_error_message() {
                            client.send_unicast(rsp)
                        }
                    }
                }
                FoldState { client, handler, access_control }
            },
        )
        .await;

    client
}

#[cfg(test)]
mod tests {
    use super::*;
    use fuchsia_async as fasync;
    use futures::FutureExt as _;

    use assert_matches::assert_matches;
    use netlink_packet_core::{ErrorMessage, NetlinkPayload};
    use std::num::NonZeroI32;
    use std::pin::pin;

    use crate::messaging::NetlinkMessageWithCreds;
    use crate::messaging::testutil::{FakeCreds, SentMessage, TestNetlinkContext};
    use crate::protocol_family::testutil::{
        FakeNetlinkRequestHandler, FakeProtocolFamily, new_fake_netlink_message,
        new_fake_netlink_message_with_creds,
    };

    #[fasync::run_singlethreaded(test)]
    async fn test_run_client_request_handler() {
        let (mut req_sender, req_receiver) = mpsc::channel(0);
        let (mut client_sink, client, async_work_drain_task) =
            crate::client::testutil::new_fake_client::<FakeProtocolFamily>(
                crate::client::testutil::CLIENT_ID_1,
                std::iter::empty(),
            );
        let join_handle = fasync::Task::spawn(async_work_drain_task);

        {
            let mut client_task = pin!(
                run_client_request_handler::<TestNetlinkContext, FakeProtocolFamily>(
                    client,
                    req_receiver,
                    FakeNetlinkRequestHandler,
                    Default::default()
                )
                .fuse()
            );

            assert_matches!((&mut client_task).now_or_never(), None);
            assert_eq!(&client_sink.take_messages()[..], &[]);

            // Send a message and expect to see the response on the `client_sink`.
            // NB: Use the sender's channel size as a synchronization method; If a
            // second message could be sent, the first *must* have been handled.
            req_sender
                .try_send(new_fake_netlink_message_with_creds())
                .expect("should send without error");
            let mut could_send_fut =
                pin!(futures::future::poll_fn(|ctx| req_sender.poll_ready(ctx)).fuse());
            futures::select!(
                res = could_send_fut => res.expect("should be able to send without error"),
                _client = client_task => panic!("client task unexpectedly finished"),
            );
            assert_eq!(
                &client_sink.take_messages()[..],
                &[SentMessage::unicast(new_fake_netlink_message())]
            );

            // Close the sender, and expect the Task to exit.
            req_sender.close_channel();
            let _client = client_task.await;
            assert_eq!(&client_sink.take_messages()[..], &[]);
        }
        join_handle.await;
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_connect_new_clients() {
        let client_table = ClientTable::default();
        let scope = fasync::Scope::new();
        let (client_sender, client_receiver) = futures::channel::mpsc::unbounded();
        let mut client_acceptor_fut = Box::pin(
            connect_new_clients::<TestNetlinkContext, FakeProtocolFamily>(
                client_table.clone(),
                client_receiver,
                FakeNetlinkRequestHandler,
                Default::default(),
            )
            .fuse(),
        );

        assert_eq!((&mut client_acceptor_fut).now_or_never(), None);

        // Connect Client 1.
        let (mut _client_sink1, client1, async_work_drain_task) =
            crate::client::testutil::new_fake_client::<FakeProtocolFamily>(
                crate::client::testutil::CLIENT_ID_1,
                std::iter::empty(),
            );
        let _join_handle = scope.spawn(async_work_drain_task);
        let (mut req_sender1, req_receiver1) = mpsc::channel(0);
        client_sender
            .unbounded_send(ClientWithReceiver { client: client1, receiver: req_receiver1 })
            .expect("should send without error");

        // Connect Client 2.
        let (mut client_sink2, client2, async_work_drain_task) =
            crate::client::testutil::new_fake_client::<FakeProtocolFamily>(
                crate::client::testutil::CLIENT_ID_2,
                std::iter::empty(),
            );
        let _join_handle = scope.spawn(async_work_drain_task);
        let (mut req_sender2, req_receiver2) = mpsc::channel(0);
        client_sender
            .unbounded_send(ClientWithReceiver { client: client2, receiver: req_receiver2 })
            .expect("should send without error");

        // Send a request to Client 2, and verify it's handled despite Client 1
        // being open (e.g. concurrent handling of requests across clients).
        // NB: Use the sender's channel size as a synchronization method; If a
        // second message could be sent, the first *must* have been handled.
        req_sender2
            .try_send(new_fake_netlink_message_with_creds())
            .expect("should send without error");
        let mut could_send_fut =
            pin!(futures::future::poll_fn(|ctx| req_sender2.poll_ready(ctx)).fuse());
        futures::select!(
            res = could_send_fut => res.expect("should be able to send without error"),
            () = client_acceptor_fut => panic!("client acceptor unexpectedly finished"),
        );
        assert_eq!(
            &client_table.client_ids()[..],
            [client::testutil::CLIENT_ID_1, client::testutil::CLIENT_ID_2]
        );
        assert_eq!(
            &client_sink2.take_messages()[..],
            &[SentMessage::unicast(new_fake_netlink_message())]
        );

        // Close the two clients, and verify the acceptor fut is still pending.
        req_sender1.close_channel();
        req_sender2.close_channel();
        assert_eq!((&mut client_acceptor_fut).now_or_never(), None);

        // Close the client_sender, and verify the acceptor fut finishes.
        client_sender.close_channel();
        client_acceptor_fut.await;

        // Confirm the clients have been cleaned up from the client table.
        assert_eq!(&client_table.client_ids()[..], []);

        drop(client_table);
        scope.join().await;
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_permissions() {
        let client_table = ClientTable::default();
        let scope = fasync::Scope::new();
        let (client_sender, client_receiver) = futures::channel::mpsc::unbounded();
        let mut client_acceptor_fut = Box::pin(
            connect_new_clients::<TestNetlinkContext, FakeProtocolFamily>(
                client_table.clone(),
                client_receiver,
                FakeNetlinkRequestHandler,
                Default::default(),
            )
            .fuse(),
        );
        assert_eq!((&mut client_acceptor_fut).now_or_never(), None);

        let (mut client_sink, client, async_work_drain_task) =
            crate::client::testutil::new_fake_client::<FakeProtocolFamily>(
                crate::client::testutil::CLIENT_ID_1,
                std::iter::empty(),
            );
        let _join_handle = scope.spawn(async_work_drain_task);
        let (mut req_sender, req_receiver) = mpsc::channel(0);
        client_sender
            .unbounded_send(ClientWithReceiver { client, receiver: req_receiver })
            .expect("should send without error");

        let message = NetlinkMessageWithCreds::new(
            new_fake_netlink_message(),
            FakeCreds::with_error(Errno::new(libc::EPERM).unwrap()),
        );
        req_sender.try_send(message).expect("should send without error");

        let response = futures::select!(
            res = client_sink.next_message().fuse() => res,
            () = client_acceptor_fut => panic!("client acceptor unexpectedly finished"),
        );

        assert_matches!(
            response.message.payload,
            NetlinkPayload::Error(ErrorMessage { code: Some(error_code), .. }) => {
              assert_eq!(error_code , NonZeroI32::new(-libc::EPERM).unwrap());
            }
        );
    }
}