netscli_core/
scan.rs

1use futures::stream::{self, StreamExt};
2use serde::Serialize;
3use std::net::{IpAddr, SocketAddr};
4use std::sync::{
5    atomic::{AtomicUsize, Ordering},
6    Arc,
7};
8use std::time::Duration;
9use tokio::net::TcpStream;
10use tokio::sync::Semaphore;
11use tokio::time::timeout;
12
13#[derive(Debug, Clone, Serialize)]
14pub struct PortResult {
15    pub port: u16,
16    pub open: bool,
17    pub service: Option<String>,
18    /// Populated when the probe failed for reasons other than a closed port
19    /// (e.g. the scanner's concurrency semaphore was closed). Omitted on
20    /// normal open/closed results.
21    #[serde(skip_serializing_if = "Option::is_none")]
22    pub error: Option<String>,
23}
24
25/// Concurrent TCP port scanner.
26///
27/// The `Arc<Semaphore>` caps TCP connects across *all* concurrent `scan_host`
28/// calls on the same scanner instance. `buffer_unordered` alone would only
29/// bound a single call — `SweepEngine` reuses one scanner across hundreds of
30/// hosts in parallel, so without the shared permit pool a /24 sweep could
31/// attempt hosts × ports simultaneous connects (easily 64k sockets).
32pub struct PortScanner {
33    semaphore: Arc<Semaphore>,
34    concurrency: usize,
35}
36
37#[derive(Debug, Clone)]
38pub struct PortScanProgress {
39    pub completed: usize,
40    pub total: usize,
41    pub port: u16,
42    pub open: bool,
43    pub open_found: usize,
44}
45
46impl PortScanner {
47    pub fn new(concurrency: usize) -> Self {
48        let concurrency = concurrency.max(1);
49        Self {
50            semaphore: Arc::new(Semaphore::new(concurrency)),
51            concurrency,
52        }
53    }
54
55    pub async fn scan_host(
56        &self,
57        target: IpAddr,
58        ports: Vec<u16>,
59        timeout_ms: u64,
60    ) -> Vec<PortResult> {
61        self.scan_host_with_progress(target, ports, timeout_ms, None)
62            .await
63    }
64
65    pub async fn scan_host_with_progress(
66        &self,
67        target: IpAddr,
68        ports: Vec<u16>,
69        timeout_ms: u64,
70        progress: Option<Arc<dyn Fn(PortScanProgress) + Send + Sync>>,
71    ) -> Vec<PortResult> {
72        if ports.is_empty() {
73            return Vec::new();
74        }
75
76        let total = ports.len();
77        let completed = Arc::new(AtomicUsize::new(0));
78        let open_found = Arc::new(AtomicUsize::new(0));
79
80        stream::iter(ports)
81            .map(|port| {
82                let scanner = self.clone();
83                let completed = completed.clone();
84                let open_found = open_found.clone();
85                let progress = progress.clone();
86                async move {
87                    let res = scanner.check_port(target, port, timeout_ms).await;
88                    let done = completed.fetch_add(1, Ordering::SeqCst) + 1;
89                    let open_count = if res.open {
90                        open_found.fetch_add(1, Ordering::SeqCst) + 1
91                    } else {
92                        open_found.load(Ordering::SeqCst)
93                    };
94
95                    if let Some(cb) = &progress {
96                        cb(PortScanProgress {
97                            completed: done,
98                            total,
99                            port,
100                            open: res.open,
101                            open_found: open_count,
102                        });
103                    }
104
105                    res
106                }
107            })
108            .buffer_unordered(self.concurrency)
109            .collect::<Vec<PortResult>>()
110            .await
111    }
112
113    async fn check_port(&self, target: IpAddr, port: u16, timeout_ms: u64) -> PortResult {
114        let _permit = match self.semaphore.acquire().await {
115            Ok(p) => p,
116            Err(_) => {
117                // Semaphore closed — record the reason so downstream callers
118                // can distinguish this from a genuinely closed port.
119                return PortResult {
120                    port,
121                    open: false,
122                    service: None,
123                    error: Some("scanner shut down (semaphore closed)".to_string()),
124                };
125            }
126        };
127
128        let addr = SocketAddr::new(target, port);
129        let result = timeout(Duration::from_millis(timeout_ms), TcpStream::connect(addr)).await;
130
131        match result {
132            Ok(Ok(_)) => PortResult {
133                port,
134                open: true,
135                service: Self::guess_service(port),
136                error: None,
137            },
138            Ok(Err(_)) | Err(_) => PortResult {
139                port,
140                open: false,
141                service: None,
142                error: None,
143            },
144        }
145    }
146
147    /// Map well-known port numbers to their canonical service name.
148    ///
149    /// Extended from the original 15-entry list to cover the services users
150    /// most commonly encounter when scanning home/office networks.
151    fn guess_service(port: u16) -> Option<String> {
152        let name = match port {
153            20 => "ftp-data",
154            21 => "ftp",
155            22 => "ssh",
156            23 => "telnet",
157            25 => "smtp",
158            53 => "dns",
159            67 => "dhcp-server",
160            68 => "dhcp-client",
161            69 => "tftp",
162            80 => "http",
163            110 => "pop3",
164            111 => "rpcbind",
165            123 => "ntp",
166            135 => "msrpc",
167            137 => "netbios-ns",
168            138 => "netbios-dgm",
169            139 => "netbios-ssn",
170            143 => "imap",
171            161 => "snmp",
172            162 => "snmp-trap",
173            389 => "ldap",
174            443 => "https",
175            445 => "smb",
176            465 => "smtps",
177            514 => "syslog",
178            587 => "smtp-submission",
179            636 => "ldaps",
180            873 => "rsync",
181            993 => "imaps",
182            995 => "pop3s",
183            1080 => "socks",
184            1433 => "mssql",
185            1521 => "oracle",
186            1723 => "pptp",
187            1883 => "mqtt",
188            2049 => "nfs",
189            2375 => "docker",
190            2376 => "docker-tls",
191            3000 => "dev-http",
192            3306 => "mysql",
193            3389 => "rdp",
194            4369 => "epmd",
195            5000 => "upnp",
196            5060 => "sip",
197            5061 => "sips",
198            5222 => "xmpp-client",
199            5432 => "postgresql",
200            5601 => "kibana",
201            5672 => "amqp",
202            5900 => "vnc",
203            5984 => "couchdb",
204            6379 => "redis",
205            6443 => "kubernetes-api",
206            6667 => "irc",
207            7000 => "cassandra",
208            8000 => "dev-http",
209            8008 => "http-alt",
210            8080 => "http-alt",
211            8081 => "http-alt",
212            8086 => "influxdb",
213            8443 => "https-alt",
214            8888 => "http-alt",
215            9000 => "http-alt",
216            9042 => "cassandra",
217            9090 => "prometheus",
218            9092 => "kafka",
219            9200 => "elasticsearch",
220            9418 => "git",
221            11211 => "memcached",
222            15672 => "rabbitmq-mgmt",
223            27017 => "mongodb",
224            27018 => "mongodb",
225            50000 => "sap",
226            _ => return None,
227        };
228        Some(name.to_string())
229    }
230}
231
232impl Clone for PortScanner {
233    fn clone(&self) -> Self {
234        Self {
235            semaphore: self.semaphore.clone(),
236            concurrency: self.concurrency,
237        }
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use super::*;
244
245    #[test]
246    fn test_guess_service() {
247        assert_eq!(PortScanner::guess_service(22), Some("ssh".to_string()));
248        assert_eq!(PortScanner::guess_service(80), Some("http".to_string()));
249        assert_eq!(PortScanner::guess_service(443), Some("https".to_string()));
250        assert_eq!(PortScanner::guess_service(9999), None);
251    }
252
253    #[test]
254    fn test_port_scanner_creation() {
255        let scanner = PortScanner::new(256);
256        assert_eq!(scanner.concurrency, 256);
257    }
258
259    #[tokio::test]
260    async fn test_scan_localhost_common_ports() {
261        let scanner = PortScanner::new(10);
262        let localhost: IpAddr = "127.0.0.1".parse().unwrap();
263        let ports = vec![22, 80, 443, 8080];
264
265        let results = scanner.scan_host(localhost, ports.clone(), 1000).await;
266
267        assert_eq!(results.len(), ports.len());
268        for result in results {
269            assert!(ports.contains(&result.port));
270            // Verify structure: if open, service may be set; if closed, service should be None
271            if result.open {
272                // Open ports may have service names for known ports
273                if result.service.is_some() {
274                    // Service name should match expected for known ports
275                    match result.port {
276                        22 => assert_eq!(result.service, Some("ssh".to_string())),
277                        80 => assert_eq!(result.service, Some("http".to_string())),
278                        443 => assert_eq!(result.service, Some("https".to_string())),
279                        8080 => assert_eq!(result.service, Some("http-alt".to_string())),
280                        _ => {}
281                    }
282                }
283            } else {
284                // Closed ports should not have service names
285                assert_eq!(result.service, None);
286            }
287        }
288    }
289}
netscli_core/scan.rs

netscli_core/
scan.rs
