Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions crates/netscli-core/src/discover.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use crate::arp::NetworkManager;
use crate::error::Result;
use crate::oui::lookup_vendor;
use crate::ping::PingScanner;
use futures::stream::{self, StreamExt};
Expand Down Expand Up @@ -56,7 +57,13 @@ impl DiscoverEngine {
ping_timeout_ms: u64,
dns_timeout_ms: u64,
) -> Self {
let concurrency = concurrency.max(1);
// Clamp both ends. `.max(1)` alone left the upper bound to
// `Semaphore::new`, which asserts `permits <= usize::MAX >> 3` --
// so a caller passing `usize::MAX` got a panic rather than an
// error, from a constructor that returns `Self` and cannot report
// one. Absurd input, but this is public API and a process abort is
// the wrong failure mode for it.
let concurrency = concurrency.clamp(1, crate::MAX_CONCURRENCY);
Self {
ping_scanner: PingScanner::new(concurrency),
concurrency,
Expand All @@ -65,16 +72,24 @@ impl DiscoverEngine {
}
}

pub async fn scan_subnet(&self, subnet: Ipv4Net, resolve: bool) -> Vec<Host> {
pub async fn scan_subnet(&self, subnet: Ipv4Net, resolve: bool) -> Result<Vec<Host>> {
self.scan_subnet_with_progress(subnet, resolve, None).await
}

/// Ping-sweep a subnet, then resolve names for the hosts that answered.
///
/// Enforces the /16 cap itself rather than trusting the caller. This is
/// public API re-exported at the crate root, and it used to collect every
/// address of whatever `Ipv4Net` it was given straight into a `Vec` --
/// `0.0.0.0/0` is 4,294,967,294 entries, roughly 73 GB, allocated before
/// a single packet is sent. `Ops` checked, the engine did not.
pub async fn scan_subnet_with_progress(
&self,
subnet: Ipv4Net,
resolve: bool,
progress: Option<Arc<dyn Fn(DiscoverProgress) + Send + Sync>>,
) -> Vec<Host> {
) -> Result<Vec<Host>> {
crate::ops::validation::ensure_subnet_limit(&subnet, &subnet.to_string())?;
let ips: Vec<IpAddr> = subnet.hosts().map(IpAddr::V4).collect();

// 1) Ping first (fast), to avoid reverse-DNS work on dead hosts.
Expand Down Expand Up @@ -184,7 +199,7 @@ impl DiscoverEngine {
HashMap::new()
};

alive
Ok(alive
.into_iter()
.map(|r| {
let hostname = hostname_map.get(&r.ip).cloned().unwrap_or(None);
Expand All @@ -201,6 +216,6 @@ impl DiscoverEngine {
rtt_ms: r.rtt_ms,
}
})
.collect()
.collect())
}
}
12 changes: 10 additions & 2 deletions crates/netscli-core/src/inspect.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ impl InspectEngine {
scan_timeout_ms: u64,
dns_timeout_ms: u64,
) -> Self {
let concurrency = concurrency.max(1);
let concurrency = concurrency.clamp(1, crate::MAX_CONCURRENCY);
Self {
ping: PingScanner::new(concurrency),
scan: PortScanner::new(concurrency),
Expand All @@ -49,6 +49,14 @@ impl InspectEngine {
}

pub async fn inspect(&self, host: String, ports: Vec<u16>) -> Result<InspectResult> {
// Validate before doing any work. `Ops::inspect_host` checks too, but
// this engine is public API and was reachable with a hand-built
// `Vec<u16>` that skipped both the 4,096-port cap and the port-0
// rejection.
if !ports.is_empty() {
crate::validate_ports(&ports)?;
}

let ip_for_scan =
crate::ops::resolve_host_ip_with_timeout(&host, self.dns_timeout_ms).await?;

Expand All @@ -63,7 +71,7 @@ impl InspectEngine {

let (ping_res, ports_res, hostname) = tokio::join!(ping_fut, scan_fut, hostname_fut);

let ports = ports_res;
let ports = ports_res?;
let open_ports = ports.iter().filter(|p| p.open).cloned().collect();

Ok(InspectResult {
Expand Down
2 changes: 1 addition & 1 deletion crates/netscli-core/src/ops.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ mod host;
mod network;
mod pcap;
mod scan;
mod validation;
pub(crate) mod validation;

pub use config::{Ops, OpsConfig, MAX_CONCURRENCY};
pub use host::{resolve_host_ip, resolve_host_ip_with_timeout, PingSummary};
4 changes: 2 additions & 2 deletions crates/netscli-core/src/ops/scan.rs
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ impl Ops {
);
let hosts = engine
.scan_subnet_with_progress(net, resolve, progress)
.await;
.await?;
Ok((subnet_str, hosts))
}

Expand All @@ -76,7 +76,7 @@ impl Ops {
let scanner = PortScanner::new(self.cfg.concurrency);
let results = scanner
.scan_host_with_progress(ip, ports, self.cfg.scan_timeout_ms, progress)
.await;
.await?;
Ok((ip, results))
}

Expand Down
10 changes: 9 additions & 1 deletion crates/netscli-core/src/ops/validation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,15 @@ pub(super) fn parse_limited_ipv4_subnet(subnet_str: &str) -> Result<Ipv4Net> {
Ok(net)
}

fn ensure_subnet_limit(net: &Ipv4Net, subnet_str: &str) -> Result<()> {
/// Refuse a subnet larger than /16.
///
/// `pub(crate)` so the engines can enforce it too. It used to be private to
/// `ops`, which meant `DiscoverEngine::scan_subnet` and `SweepEngine::sweep`
/// -- both re-exported at the crate root -- materialised every address of
/// whatever `Ipv4Net` they were handed. `0.0.0.0/0` asks for a
/// 4,294,967,294-element `Vec<IpAddr>`, about 73 GB, before a single packet
/// is sent.
pub(crate) fn ensure_subnet_limit(net: &Ipv4Net, subnet_str: &str) -> Result<()> {
let prefix = net.prefix_len() as u32;
let host_bits = 32u32.saturating_sub(prefix);
let total = 1u64.checked_shl(host_bits).unwrap_or(u64::MAX);
Expand Down
27 changes: 25 additions & 2 deletions crates/netscli-core/src/pcap.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,18 +34,41 @@ impl PcapEngine {
}

pub fn capture(config: PcapConfig) -> Result<PcapResult> {
capture::capture(config)
Self::capture_with_cancel(config, None)
}

/// Capture packets to `config.output_file`.
///
/// Refuses a config with no stopping condition. Every field of
/// `PcapConfig` is public, and the helper that guarantees at least one
/// bound lives in the `ops` layer -- so a caller constructing the config
/// directly with `duration: None, max_packets: None` got a loop that
/// breaks only on cancel, and `capture` does not even accept a cancel
/// token. It wrote packets until the filesystem filled.
pub fn capture_with_cancel(
config: PcapConfig,
cancel: Option<PcapCancelToken>,
) -> Result<PcapResult> {
if config.duration.is_none() && config.max_packets.is_none() && cancel.is_none() {
return Err(crate::error::Error::invalid_input(
"packet capture needs a duration, a max packet count, or a cancel token",
));
}
capture::capture_with_cancel(config, cancel)
}

/// Summarise packets from a capture file.
///
/// `max_packets` is bounded here rather than trusted. `None` meant
/// "retain every packet", and `Ops::parse_pcap_file` forwarded the
/// caller's value untouched while the capture paths normalised theirs --
/// so a ~500 MB file of 10M small frames built 10M summaries, about 5 GB,
/// each carrying an unconditional 191-char hex preview.
pub fn parse_file(path: PathBuf, max_packets: Option<usize>) -> Result<PcapParseResult> {
parse::parse_file(path, max_packets)
let bounded = max_packets
.unwrap_or(crate::MAX_PCAP_CAPTURE_PACKETS)
.min(crate::MAX_PCAP_CAPTURE_PACKETS);
parse::parse_file(path, Some(bounded))
}
}

Expand Down
4 changes: 0 additions & 4 deletions crates/netscli-core/src/pcap/capture.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,6 @@ pub(super) fn check_support() -> Result<Vec<String>> {
Ok(devices.into_iter().map(|d| d.name).collect())
}

pub(super) fn capture(config: PcapConfig) -> Result<PcapResult> {
capture_with_cancel(config, None)
}

pub(super) fn capture_with_cancel(
config: PcapConfig,
cancel: Option<PcapCancelToken>,
Expand Down
2 changes: 1 addition & 1 deletion crates/netscli-core/src/ping.rs
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ pub struct PingScanner {

impl PingScanner {
pub fn new(concurrency: usize) -> Self {
let concurrency = concurrency.max(1);
let concurrency = concurrency.clamp(1, crate::MAX_CONCURRENCY);
Self {
semaphore: Arc::new(Semaphore::new(concurrency)),
backend: if *RAW_ICMP_OK.get_or_init(can_use_raw_icmpv4) {
Expand Down
25 changes: 19 additions & 6 deletions crates/netscli-core/src/scan/tcp.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ use tokio::time::timeout;
use super::probes::{first_banner_line, probe_http, probe_tls, read_banner};
use super::services::{classify_connect_error, guess_service, is_http_port, is_tls_port};
use super::types::{PortResult, PortStatus};
use crate::error::Result;

/// Concurrent TCP port scanner.
///
Expand All @@ -36,7 +37,7 @@ pub struct PortScanProgress {

impl PortScanner {
pub fn new(concurrency: usize) -> Self {
let concurrency = concurrency.max(1);
let concurrency = concurrency.clamp(1, crate::MAX_CONCURRENCY);
Self {
semaphore: Arc::new(Semaphore::new(concurrency)),
concurrency,
Expand All @@ -48,27 +49,37 @@ impl PortScanner {
target: IpAddr,
ports: Vec<u16>,
timeout_ms: u64,
) -> Vec<PortResult> {
) -> Result<Vec<PortResult>> {
self.scan_host_with_progress(target, ports, timeout_ms, None)
.await
}

/// Scan `ports` on `target`.
///
/// Validates the port list rather than trusting the caller. `Ops` does
/// the same check before calling in, but this is public API on a
/// published crate: a consumer building a `Vec<u16>` by hand reached the
/// scanner directly and neither the 4,096-port cap nor the port-0
/// rejection applied. The doc comment on `Ops::resolve_ports` claimed
/// "every scanning entry point funnels through here", and this was one
/// of two that did not.
pub async fn scan_host_with_progress(
&self,
target: IpAddr,
ports: Vec<u16>,
timeout_ms: u64,
progress: Option<Arc<dyn Fn(PortScanProgress) + Send + Sync>>,
) -> Vec<PortResult> {
) -> Result<Vec<PortResult>> {
if ports.is_empty() {
return Vec::new();
return Ok(Vec::new());
}
crate::validate_ports(&ports)?;

let total = ports.len();
let completed = Arc::new(AtomicUsize::new(0));
let open_found = Arc::new(AtomicUsize::new(0));

stream::iter(ports)
let results = stream::iter(ports)
.map(|port| {
let scanner = self.clone();
let completed = completed.clone();
Expand Down Expand Up @@ -98,7 +109,9 @@ impl PortScanner {
})
.buffer_unordered(self.concurrency)
.collect::<Vec<PortResult>>()
.await
.await;

Ok(results)
}

async fn check_port(&self, target: IpAddr, port: u16, timeout_ms: u64) -> PortResult {
Expand Down
11 changes: 8 additions & 3 deletions crates/netscli-core/src/scan/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,10 @@ async fn test_scan_localhost_common_ports() {
let localhost: IpAddr = "127.0.0.1".parse().unwrap();
let ports = vec![22, 80, 443, 8080];

let results = scanner.scan_host(localhost, ports.clone(), 1000).await;
let results = scanner
.scan_host(localhost, ports.clone(), 1000)
.await
.expect("valid port list");

assert_eq!(results.len(), ports.len());
for result in results {
Expand Down Expand Up @@ -76,7 +79,8 @@ async fn test_scan_open_localhost_port_returns_open_with_banner() {
let scanner = PortScanner::new(4);
let mut results = scanner
.scan_host(IpAddr::from([127, 0, 0, 1]), vec![port], 1000)
.await;
.await
.expect("valid port list");

assert_eq!(results.len(), 1);
let result = results.remove(0);
Expand All @@ -100,7 +104,8 @@ async fn test_scan_closed_local_port_records_latency() {
let scanner = PortScanner::new(4);
let mut results = scanner
.scan_host(IpAddr::from([127, 0, 0, 1]), vec![port], 1000)
.await;
.await
.expect("valid port list");

assert_eq!(results.len(), 1);
let result = results.remove(0);
Expand Down
9 changes: 7 additions & 2 deletions crates/netscli-core/src/sweep.rs
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ impl SweepEngine {
scan_timeout_ms: u64,
dns_timeout_ms: u64,
) -> Self {
let concurrency = concurrency.max(1);
let concurrency = concurrency.clamp(1, crate::MAX_CONCURRENCY);
Self {
discover: DiscoverEngine::new_with_timeouts(
concurrency,
Expand Down Expand Up @@ -106,7 +106,7 @@ impl SweepEngine {
let hosts = self
.discover
.scan_subnet_with_progress(subnet, resolve_hostnames, discover_progress)
.await;
.await?;
let total_hosts = hosts.len();
let completed = Arc::new(AtomicUsize::new(0));
let open_hosts = Arc::new(AtomicUsize::new(0));
Expand All @@ -124,9 +124,14 @@ impl SweepEngine {
let progress = progress.clone();
async move {
let host_ip = h.ip;
// The port list is validated once by the caller, so a
// per-host failure here would be the same error repeated
// for every host. Treat it as "no open ports" for this
// entry rather than aborting the whole sweep.
let open_ports = scanner
.scan_host(host_ip, (*ports).clone(), scan_timeout_ms)
.await
.unwrap_or_default()
.into_iter()
.filter(|p| p.open)
.collect();
Expand Down
Loading
Loading