r_lanlib/scanners/
arp_scanner.rs

1//! Provides Scanner implementation for ARP scanning
2
3use derive_builder::Builder;
4use pnet::packet::{Packet, arp, ethernet};
5use std::{
6    net::Ipv4Addr,
7    sync::{self, Arc},
8    thread::{self, JoinHandle},
9    time::Duration,
10};
11use threadpool::ThreadPool;
12
13use crate::{
14    error::{RLanLibError, Result},
15    network::NetworkInterface,
16    packet::{self, arp_packet::ArpPacketBuilder, wire::Wire},
17    scanners::{Device, PortSet, Scanning},
18    targets::ips::IPTargets,
19};
20
21use super::{ScanMessage, Scanner, heartbeat::HeartBeat};
22
23/// Data structure representing an ARP scanner
24#[derive(Clone, Builder)]
25#[builder(setter(into))]
26pub struct ARPScanner {
27    /// Network interface to use for scanning
28    interface: Arc<NetworkInterface>,
29    /// Wire for reading and sending packets on the wire
30    wire: Wire,
31    /// IP targets to scan
32    targets: Arc<IPTargets>,
33    /// Source port for packet listener and incoming packet identification
34    source_port: u16,
35    /// Whether to include vendor lookups for discovered devices
36    include_vendor: bool,
37    /// Whether to include hostname lookups for discovered devices
38    include_host_names: bool,
39    /// Duration to wait for responses after scanning completes
40    idle_timeout: Duration,
41    /// Channel for sending scan results and status messages
42    notifier: sync::mpsc::Sender<ScanMessage>,
43}
44
45impl ARPScanner {
46    /// Returns builder for ARPScanner
47    pub fn builder() -> ARPScannerBuilder {
48        ARPScannerBuilder::default()
49    }
50
51    fn process_target(&self, target: Ipv4Addr) -> Result<()> {
52        // throttle packet sending to prevent packet loss
53        thread::sleep(packet::DEFAULT_PACKET_SEND_TIMING);
54
55        log::debug!("scanning ARP target: {}", target);
56
57        let arp_packet = ArpPacketBuilder::default()
58            .source_ip(self.interface.ipv4)
59            .source_mac(self.interface.mac)
60            .dest_ip(target)
61            .build()?;
62
63        let pkt_buf = arp_packet.to_raw();
64
65        // inform consumer we are scanning this target (ignore error on failure to notify)
66        self.notifier
67            .send(ScanMessage::Info(Scanning {
68                ip: target,
69                port: None,
70            }))
71            .map_err(RLanLibError::from_channel_send_error)?;
72
73        let mut pkt_sender = self.wire.0.lock()?;
74
75        // Send to the broadcast address
76        pkt_sender.send(&pkt_buf)?;
77
78        Ok(())
79    }
80
81    fn process_incoming_packet(
82        &self,
83        pkt: &[u8],
84        pool: &ThreadPool,
85    ) -> Result<()> {
86        let Some(eth) = ethernet::EthernetPacket::new(pkt) else {
87            return Ok(());
88        };
89
90        let Some(header) = arp::ArpPacket::new(eth.payload()) else {
91            return Ok(());
92        };
93
94        // Capture ANY ARP reply as it's an indication that there's a
95        // device on the network
96        if header.get_operation() != arp::ArpOperations::Reply {
97            return Ok(());
98        }
99
100        let ip4 = header.get_sender_proto_addr();
101        let mac = eth.get_source();
102
103        let notification_sender = self.notifier.clone();
104        let interface = Arc::clone(&self.interface);
105        let include_host_names = self.include_host_names;
106        let include_vendor = self.include_vendor;
107
108        // use a thread pool here so we don't slow down packet
109        // processing while limiting concurrent threads
110        pool.execute(move || {
111            let hostname = if include_host_names {
112                log::debug!("looking up hostname for {}", ip4);
113                dns_lookup::lookup_addr(&ip4.into()).unwrap_or_default()
114            } else {
115                String::new()
116            };
117
118            let vendor = if include_vendor {
119                oui_data::lookup(&mac.to_string())
120                    .map(|v| v.organization().to_owned())
121                    .unwrap_or_default()
122            } else {
123                String::new()
124            };
125
126            let _ =
127                notification_sender.send(ScanMessage::ARPScanDevice(Device {
128                    hostname,
129                    ip: ip4,
130                    mac,
131                    vendor,
132                    is_current_host: ip4 == interface.ipv4,
133                    open_ports: PortSet::new(),
134                }));
135        });
136
137        Ok(())
138    }
139
140    // Implements packet reading in a separate thread so we can send and
141    // receive packets simultaneously
142    fn read_packets(
143        &self,
144        done: sync::mpsc::Receiver<()>,
145    ) -> Result<JoinHandle<Result<()>>> {
146        let (heartbeat_tx, heartbeat_rx) = sync::mpsc::channel::<()>();
147
148        let heartbeat = HeartBeat::builder()
149            .source_mac(self.interface.mac)
150            .source_ipv4(self.interface.ipv4)
151            .source_port(self.source_port)
152            .packet_sender(Arc::clone(&self.wire.0))
153            .build()?;
154
155        heartbeat.start_in_thread(heartbeat_rx)?;
156
157        let self_clone = self.clone();
158
159        Ok(thread::spawn(move || -> Result<()> {
160            let mut reader = self_clone.wire.1.lock()?;
161            // Use a bounded thread pool for DNS/vendor lookups to prevent
162            // spawning thousands of threads on large networks
163            let lookup_pool = ThreadPool::new(8);
164
165            loop {
166                if done.try_recv().is_ok() {
167                    log::debug!("exiting arp packet reader");
168                    if let Err(e) = heartbeat_tx.send(()) {
169                        log::error!("failed to stop heartbeat: {}", e);
170                    }
171                    break;
172                }
173
174                let pkt = reader.next_packet()?;
175
176                self_clone.process_incoming_packet(pkt, &lookup_pool)?;
177            }
178
179            Ok(())
180        }))
181    }
182}
183
184// Implements the Scanner trait for ARPScanner
185impl Scanner for ARPScanner {
186    fn scan(&self) -> Result<JoinHandle<Result<()>>> {
187        log::debug!("performing ARP scan on targets: {:?}", self.targets);
188        log::debug!("include_vendor: {}", self.include_vendor);
189        log::debug!("include_host_names: {}", self.include_host_names);
190        log::debug!("starting arp packet reader");
191
192        let self_clone = self.clone();
193        let (done_tx, done_rx) = sync::mpsc::channel::<()>();
194
195        let read_handle = self.read_packets(done_rx)?;
196
197        // prevent blocking thread so messages can be freely sent to consumer
198        let scan_handle = thread::spawn(move || -> Result<()> {
199            let mut scan_error: Option<RLanLibError> = None;
200
201            if let Err(err) = self_clone
202                .targets
203                .lazy_loop(|t| self_clone.process_target(t))
204            {
205                scan_error = Some(err);
206            }
207
208            thread::sleep(self_clone.idle_timeout);
209
210            self_clone
211                .notifier
212                .send(ScanMessage::Done)
213                .map_err(RLanLibError::from_channel_send_error)?;
214
215            // ignore errors here as the thread may already be dead due to error
216            // we'll catch any errors from that thread below and report
217            let _ = done_tx.send(());
218
219            let read_result = read_handle.join()?;
220
221            if let Some(err) = scan_error {
222                return Err(err);
223            }
224
225            read_result
226        });
227
228        Ok(scan_handle)
229    }
230}
231
232#[cfg(test)]
233#[path = "./arp_scanner_tests.rs"]
234mod tests;
r_lanlib/scanners/arp_scanner.rs

r_lanlib/scanners/
arp_scanner.rs
