roowho2_lib/server/rwhod/
packet_sender.rs

1use nix::{ifaddrs::getifaddrs, net::if_::InterfaceFlags};
2use std::{
3    collections::HashSet,
4    net::{IpAddr, Ipv4Addr, SocketAddr},
5    sync::Arc,
6};
7use tokio::{
8    net::UdpSocket,
9    sync::mpsc,
10    time::{Duration as TokioDuration, interval},
11};
12
13use crate::{
14    proto::Whod,
15    server::{ignore_list::IgnoreList, rwhod::rwhod_status::generate_rwhod_status_update},
16};
17
18/// Default port for rwhod communication.
19pub const RWHOD_BROADCAST_PORT: u16 = 513;
20
21/// How long to wait after a realtime update before actually reading utmp and sending a status update.
22/// This is necessary because audit log events can arrive before the utmp record gets updated.
23const REALTIME_UPDATE_SETTLE_DELAY: TokioDuration = TokioDuration::from_millis(500);
24
25#[derive(Debug, Clone)]
26pub struct RwhodSendTarget {
27    /// Name of the network interface.
28    pub name: String,
29
30    /// Address to send rwhod packets to.
31    /// This is either the broadcast address (for broadcast interfaces)
32    /// or the point-to-point destination address (for point-to-point interfaces).
33    pub addr: IpAddr,
34}
35
36/// Computes the broadcast address for an IPv4 address/netmask pair
37fn ipv4_broadcast_address(address: Ipv4Addr, netmask: Ipv4Addr) -> Ipv4Addr {
38    Ipv4Addr::from(u32::from(address) | !u32::from(netmask))
39}
40
41/// Find all networks network interfaces suitable for rwhod communication.
42///
43/// If `allowed_interfaces` is `Some`, only interfaces whose name is contained
44/// in it are considered; otherwise all suitable interfaces are returned.
45pub fn determine_relevant_interfaces(
46    allowed_interfaces: Option<&HashSet<String>>,
47) -> anyhow::Result<Vec<RwhodSendTarget>> {
48    getifaddrs().map_err(|e| e.into()).map(|ifaces| {
49        ifaces
50            // interface must be up
51            .filter(|iface| iface.flags.contains(InterfaceFlags::IFF_UP))
52            // interface must be broadcast or point-to-point
53            .filter(|iface| {
54                iface
55                    .flags
56                    .intersects(InterfaceFlags::IFF_BROADCAST | InterfaceFlags::IFF_POINTOPOINT)
57            })
58            // interface must be in the configured allowlist, if any
59            .filter(|iface| {
60                allowed_interfaces
61                    .is_none_or(|allowed| allowed.contains(iface.interface_name.as_str()))
62            })
63            .filter_map(|iface| {
64                let neighbor_addr = if iface.flags.contains(InterfaceFlags::IFF_BROADCAST) {
65                    match (
66                        iface.address.as_ref().and_then(|a| a.as_sockaddr_in()),
67                        iface.netmask.as_ref().and_then(|a| a.as_sockaddr_in()),
68                    ) {
69                        (Some(addr), Some(mask)) => {
70                            Some(ipv4_broadcast_address(addr.ip(), mask.ip()).into())
71                        }
72                        _ => None,
73                    }
74                } else if iface.flags.contains(InterfaceFlags::IFF_POINTOPOINT) {
75                    iface.destination.and_then(|addr| {
76                        addr.as_sockaddr_in()
77                            .map(|sa| IpAddr::V4(sa.ip()))
78                            .or_else(|| addr.as_sockaddr_in6().map(|sa| IpAddr::V6(sa.ip())))
79                    })
80                } else {
81                    None
82                };
83
84                neighbor_addr.map(|ip_addr| RwhodSendTarget {
85                    name: iface.interface_name,
86                    addr: ip_addr,
87                })
88            })
89            // keep first occurrence per interface name
90            .scan(HashSet::new(), |seen, n| {
91                if seen.insert(n.name.clone()) {
92                    Some(n)
93                } else {
94                    None
95                }
96            })
97            .collect::<Vec<RwhodSendTarget>>()
98    })
99}
100
101pub async fn send_rwhod_packet_to_interface(
102    socket: Arc<UdpSocket>,
103    interface: &RwhodSendTarget,
104    packet: &Whod,
105) -> anyhow::Result<()> {
106    let serialized_packet = packet.to_bytes();
107
108    // TODO: the old rwhod daemon doesn't actually ever listen to ipv6, maybe remove it
109    let target_addr = match interface.addr {
110        IpAddr::V4(addr) => SocketAddr::new(IpAddr::V4(addr), RWHOD_BROADCAST_PORT),
111        IpAddr::V6(addr) => SocketAddr::new(IpAddr::V6(addr), RWHOD_BROADCAST_PORT),
112    };
113
114    tracing::debug!(
115        "Sending rwhod packet to interface {} at address {}",
116        interface.name,
117        target_addr
118    );
119
120    socket
121        .send_to(&serialized_packet, &target_addr)
122        .await
123        .map_err(|e| anyhow::anyhow!("Failed to send rwhod packet: {}", e))?;
124
125    Ok(())
126}
127
128pub async fn rwhod_packet_sender_task(
129    socket: Arc<UdpSocket>,
130    interfaces: Vec<RwhodSendTarget>,
131    ignore_list: Option<IgnoreList>,
132    send_interval: TokioDuration,
133    mut realtime_update_trigger: mpsc::Receiver<()>,
134) -> anyhow::Result<()> {
135    let mut interval = interval(send_interval);
136    let mut trigger_closed = false;
137
138    loop {
139        if trigger_closed {
140            interval.tick().await;
141        } else {
142            tokio::select! {
143                _ = interval.tick() => {}
144                triggered = realtime_update_trigger.recv() => {
145                    if triggered.is_some() {
146                        tracing::debug!("Sending an early rwhod update due to realtime trigger");
147                        interval.reset();
148                        tokio::time::sleep(REALTIME_UPDATE_SETTLE_DELAY).await;
149                    } else {
150                        tracing::warn!(
151                            "Realtime update channel closed unexpectedly; falling back to interval-only updates"
152                        );
153                        trigger_closed = true;
154                        continue;
155                    }
156                }
157            }
158        }
159
160        let status_update = generate_rwhod_status_update(ignore_list.as_ref())?;
161
162        tracing::debug!("Generated rwhod packet: {:?}", status_update);
163
164        let packet = status_update
165            .try_into()
166            .map_err(|e| anyhow::anyhow!("{}", e))?;
167
168        for interface in &interfaces {
169            if let Err(e) = send_rwhod_packet_to_interface(socket.clone(), interface, &packet).await
170            {
171                tracing::error!(
172                    "Failed to send rwhod packet on interface {}: {}",
173                    interface.name,
174                    e
175                );
176            }
177        }
178    }
179}
180
181#[cfg(test)]
182mod tests {
183    use super::*;
184
185    #[test]
186    fn test_ipv4_broadcast_address() {
187        assert_eq!(
188            ipv4_broadcast_address(
189                Ipv4Addr::new(192, 168, 1, 2),
190                Ipv4Addr::new(255, 255, 255, 0)
191            ),
192            Ipv4Addr::new(192, 168, 1, 255)
193        );
194
195        assert_eq!(
196            ipv4_broadcast_address(Ipv4Addr::new(10, 0, 5, 200), Ipv4Addr::new(255, 0, 0, 0)),
197            Ipv4Addr::new(10, 255, 255, 255)
198        );
199
200        assert_eq!(
201            ipv4_broadcast_address(
202                Ipv4Addr::new(203, 0, 113, 42),
203                Ipv4Addr::new(255, 255, 255, 255)
204            ),
205            Ipv4Addr::new(203, 0, 113, 42)
206        );
207    }
208
209    #[test]
210    fn test_determine_relevant_interfaces() {
211        let interfaces = determine_relevant_interfaces(None).unwrap();
212        for interface in interfaces {
213            println!("Interface: {} Address: {}", interface.name, interface.addr);
214        }
215    }
216
217    #[test]
218    fn test_determine_relevant_interfaces_with_allowlist_excludes_unlisted() {
219        let allowed = HashSet::from(["definitely-not-a-real-interface".to_string()]);
220        let interfaces = determine_relevant_interfaces(Some(&allowed)).unwrap();
221        assert!(interfaces.is_empty());
222    }
223}
roowho2_lib/server/rwhod/packet_sender.rs

roowho2_lib/server/rwhod/
packet_sender.rs
