Merge pull request #238 from mayanayza/feat/open-ports-service

feat: unclaimed ports + additional generic services

Author: mayanayza

backend/src/bin/daemon.rs

@@ -19,8 +19,16 @@ use tower_http::{
 };
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
+fn main() -> anyhow::Result<()> {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .thread_stack_size(4 * 1024 * 1024) // 4MB stack for deep async scanning
+        .enable_all()
+        .build()?;
+
+    runtime.block_on(async_main())
+}
+
+async fn async_main() -> anyhow::Result<()> {
     // Parse CLI and load config
     let cli = DaemonCli::parse();
     let config = AppConfig::load(cli)?;

backend/src/daemon/discovery/service/base.rs

@@ -15,7 +15,7 @@ use crate::{
         discovery::r#impl::types::{DiscoveryType, HostNamingFallback},
         groups::r#impl::base::Group,
         services::{
-            definitions::docker_container::DockerContainer,
+            definitions::{docker_container::DockerContainer, open_ports::OpenPorts},
             r#impl::{
                 base::{
                     DiscoverySessionServiceMatchParams, ServiceMatchBaselineParams,
@@ -462,15 +462,16 @@ pub trait DiscoversNetworkedEntities:
         sorted_service_definitions.sort_by_key(|s| {
             if !ServiceDefinitionExt::is_generic(s) {
                 0 // Highest priority - non-generic services
-            } else if ServiceDefinitionExt::is_generic(s)
-                && s.id() != DockerContainer.id()
-                && s.id() != Gateway.id()
-            {
-                1 // Generic services that aren't Docker Container or Gateway
-            } else {
-                // Docker Containers and Gateways need to go last
+            } else if s.id() == OpenPorts.id() {
+                // Catch-all for open ports, should be dead last
+                3
+            } else if s.id() == DockerContainer.id() || s.id() == Gateway.id() {
+                // Docker Containers and Gateways need to go second to last last
                 // Other generic services should be able to get matched first
                 2
+            } else {
+                // Generic services that aren't Docker Container or Gateway
+                1
             }
         });
 

backend/src/daemon/discovery/service/network.rs

@@ -3,7 +3,7 @@ use crate::daemon::discovery::service::base::{
 };
 use crate::daemon::discovery::types::base::{DiscoveryCriticalError, DiscoverySessionUpdate};
 use crate::daemon::utils::scanner::{
-    arp_scan_host, scan_endpoints, scan_tcp_ports, scan_udp_ports,
+    arp_scan_host, can_arp_scan, scan_endpoints, scan_tcp_ports, scan_udp_ports,
 };
 use crate::server::discovery::r#impl::types::{DiscoveryType, HostNamingFallback};
 use crate::server::hosts::r#impl::{
@@ -195,14 +195,21 @@ impl DiscoveryRunner<NetworkScanDiscovery> {
             .map(|p| p.number())
             .collect();
 
-        // Partition IPs by whether their subnet is interfaced
-        let (interfaced_ips, non_interfaced_ips): (Vec<_>, Vec<_>) =
+        // Check ARP capability once before partitioning
+        let arp_available = can_arp_scan();
+
+        // Partition IPs - only use ARP path if we have capability
+        let (interfaced_ips, non_interfaced_ips): (Vec<_>, Vec<_>) = if arp_available {
             all_ips_with_subnets.into_iter().partition(|(_, subnet)| {
                 subnet_cidr_to_mac
                     .get(&subnet.base.cidr)
                     .and_then(|m| *m)
                     .is_some()
-            });
+            })
+        } else {
+            // No ARP capability - treat all as non-interfaced (port scan only)
+            (Vec::new(), all_ips_with_subnets)
+        };
 
         // =============================================================
         // PHASE 1: Responsiveness check (0-50%)
@@ -374,45 +381,46 @@ impl DiscoveryRunner<NetworkScanDiscovery> {
 
         let phase2_batches_done = Arc::new(AtomicUsize::new(0));
 
+        // In scan_and_process_hosts, box the deep scan futures
         let results = stream::iter(responsive_hosts)
-        .map(|(ip, subnet, mac, phase1_ports)| {
-            let cancel = cancel.clone();
-            let gateway_ips = gateway_ips.clone();
-            let phase2_batches_done = phase2_batches_done.clone();
-
-            async move {
-                let result = self
-                    .deep_scan_host(DeepScanParams{
-                        ip,
-                        subnet: &subnet,
-                        mac,
-                        phase1_ports,
-                        cancel,
-                        port_scan_batch_size: ports_per_host_batch,
-                        gateway_ips: &gateway_ips,
-                        batches_done: &phase2_batches_done,
-                        total_batches,
-                    })
-                    .await;
-
-                match result {
-                    Ok(Some(host)) => Some(host),
-                    Ok(None) => None,
-                    Err(e) => {
-                        if DiscoveryCriticalError::is_critical_error(e.to_string()) {
-                            tracing::error!(ip = %ip, error = %e, "Critical error in deep scan");
-                        } else {
-                            tracing::warn!(ip = %ip, error = %e, "Deep scan failed");
-                        }
-                        None
+    .map(|(ip, subnet, mac, phase1_ports)| {
+        let cancel = cancel.clone();
+        let gateway_ips = gateway_ips.clone();
+        let batches_done = phase2_batches_done.clone();
+
+        Box::pin(async move {
+            let result = self
+                .deep_scan_host(DeepScanParams {
+                    ip,
+                    subnet: &subnet,
+                    mac,
+                    phase1_ports,
+                    cancel,
+                    port_scan_batch_size: ports_per_host_batch,
+                    gateway_ips: &gateway_ips,
+                    batches_done: &batches_done,
+                    total_batches,
+                })
+                .await;
+
+            match result {
+                Ok(Some(host)) => Some(host),
+                Ok(None) => None,
+                Err(e) => {
+                    if DiscoveryCriticalError::is_critical_error(e.to_string()) {
+                        tracing::error!(ip = %ip, error = %e, "Critical error in deep scan");
+                    } else {
+                        tracing::warn!(ip = %ip, error = %e, "Deep scan failed");
                     }
+                    None
                 }
             }
         })
-        .buffer_unordered(deep_scan_concurrency)
-        .filter_map(|x| async { x })
-        .collect::<Vec<Host>>()
-        .await;
+    })
+    .buffer_unordered(deep_scan_concurrency)
+    .filter_map(|x| async { x })
+    .collect::<Vec<Host>>()
+    .await;
 
         self.report_discovery_update(DiscoverySessionUpdate::scanning(100))
             .await?;
@@ -513,7 +521,7 @@ impl DiscoveryRunner<NetworkScanDiscovery> {
             all_tcp_ports.extend(open_ports);
 
             let done = batches_done.fetch_add(1, Ordering::Relaxed) + 1;
-            let pct = (50 + done * 50 / total_batches.max(1)) as u8;
+            let pct = (50 + done * 40 / total_batches.max(1)) as u8; // 50-90%
             let _ = self.report_scanning_progress(pct).await;
         }
 
@@ -539,6 +547,9 @@ impl DiscoveryRunner<NetworkScanDiscovery> {
         .await?;
         open_ports.extend(udp_ports);
 
+        self.report_discovery_update(DiscoverySessionUpdate::scanning(95))
+            .await?;
+
         let mut ports_to_check = open_ports.clone();
         let endpoint_only_ports = Service::endpoint_only_ports();
         ports_to_check.extend(endpoint_only_ports);
@@ -554,6 +565,9 @@ impl DiscoveryRunner<NetworkScanDiscovery> {
         )
         .await?;
 
+        self.report_discovery_update(DiscoverySessionUpdate::scanning(98))
+            .await?;
+
         for endpoint_response in &endpoint_responses {
             let port = endpoint_response.endpoint.port_base;
             if !open_ports.contains(&port) {

backend/src/daemon/utils/scanner.rs

@@ -21,6 +21,7 @@ use rsntp::AsyncSntpClient;
 use snmp2::{AsyncSession, Oid};
 use std::collections::HashMap;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::pin::Pin;
 use std::time::Duration;
 use tokio::net::UdpSocket;
 use tokio::{net::TcpStream, time::timeout};
@@ -57,22 +58,22 @@ where
 {
     let mut results = Vec::new();
     let mut item_iter = items.into_iter();
-    let mut futures = FuturesUnordered::new();
 
-    // Fill initial batch
+    let mut futures: FuturesUnordered<Pin<Box<dyn Future<Output = Option<O>> + Send>>> =
+        FuturesUnordered::new();
+
     for _ in 0..batch_size {
         if cancel.is_cancelled() {
             break;
         }
 
         if let Some(item) = item_iter.next() {
-            futures.push(scan_fn(item));
+            futures.push(Box::pin(scan_fn(item)));
         } else {
             break;
         }
     }
 
-    // Process results and maintain constant parallelism
     while let Some(result) = futures.next().await {
         if cancel.is_cancelled() {
             break;
@@ -82,11 +83,9 @@ where
             results.push(output);
         }
 
-        // Immediately add next item(s) to maintain batch size
-        // Keep adding until we're back at batch_size or out of items
         while futures.len() < batch_size && !cancel.is_cancelled() {
             if let Some(item) = item_iter.next() {
-                futures.push(scan_fn(item));
+                futures.push(Box::pin(scan_fn(item)));
             } else {
                 break;
             }
@@ -96,6 +95,52 @@ where
     results
 }
 
+/// Check if ARP scanning is available (requires elevated privileges on some OSes)
+pub fn can_arp_scan() -> bool {
+    // Try to open a datalink channel on any suitable interface
+    let interfaces = datalink::interfaces();
+
+    let suitable_interface = interfaces
+        .into_iter()
+        .find(|iface| iface.is_up() && !iface.is_loopback() && iface.mac.is_some());
+
+    let Some(interface) = suitable_interface else {
+        tracing::debug!("No suitable interface found for ARP capability check");
+        return false;
+    };
+
+    let config = pnet::datalink::Config {
+        read_timeout: Some(Duration::from_millis(100)),
+        ..Default::default()
+    };
+
+    match datalink::channel(&interface, config) {
+        Ok(_) => {
+            tracing::debug!(interface = %interface.name, "ARP scanning available");
+            true
+        }
+        Err(e) => {
+            let err_str = e.to_string().to_lowercase();
+            if err_str.contains("permission")
+                || err_str.contains("operation not permitted")
+                || err_str.contains("access denied")
+                || err_str.contains("requires root")
+            {
+                tracing::info!(
+                    error = %e,
+                    "ARP scanning unavailable (insufficient privileges), falling back to port scanning"
+                );
+            } else {
+                tracing::warn!(
+                    error = %e,
+                    "ARP scanning unavailable, falling back to port scanning"
+                );
+            }
+            false
+        }
+    }
+}
+
 /// Send ARP request to a single IP and wait for response
 /// Returns the MAC address if the host responds, None otherwise
 pub async fn arp_scan_host(

backend/src/server/daemons/service.rs

@@ -122,12 +122,31 @@ impl DaemonService {
             path: "/api/discovery/initiate".to_string(),
         };
 
+        tracing::info!(
+            daemon_id = %daemon_id,
+            endpoint = %endpoint,
+            session_id = %request.session_id,
+            "Attempting to send discovery request to daemon"
+        );
+
         let response = self
             .client
             .post(format!("{}", endpoint))
             .json(&request)
             .send()
-            .await?;
+            .await
+            .map_err(|e| {
+                tracing::error!(
+                    daemon_id = %daemon_id,
+                    endpoint = %endpoint,
+                    error = %e,
+                    error_debug = ?e,
+                    is_connect = %e.is_connect(),
+                    is_timeout = %e.is_timeout(),
+                    "Failed to connect to daemon"
+                );
+                e
+            })?;
 
         if !response.status().is_success() {
             anyhow::bail!(

backend/src/server/services/definitions/ampq.rs

@@ -0,0 +1,31 @@
+use crate::server::hosts::r#impl::ports::PortBase;
+use crate::server::services::definitions::{ServiceDefinitionFactory, create_service};
+use crate::server::services::r#impl::categories::ServiceCategory;
+use crate::server::services::r#impl::definitions::ServiceDefinition;
+use crate::server::services::r#impl::patterns::Pattern;
+
+#[derive(Default, Clone, Eq, PartialEq, Hash)]
+pub struct Amqp;
+
+impl ServiceDefinition for Amqp {
+    fn name(&self) -> &'static str {
+        "AMQP"
+    }
+    fn description(&self) -> &'static str {
+        "Advanced Message Queuing Protocol"
+    }
+    fn category(&self) -> ServiceCategory {
+        ServiceCategory::MessageQueue
+    }
+    fn discovery_pattern(&self) -> Pattern<'_> {
+        Pattern::AnyOf(vec![
+            Pattern::Port(PortBase::AMQP),
+            Pattern::Port(PortBase::AMQPTls),
+        ])
+    }
+    fn is_generic(&self) -> bool {
+        true
+    }
+}
+
+inventory::submit!(ServiceDefinitionFactory::new(create_service::<Amqp>));

backend/src/server/services/definitions/docker_container.rs

@@ -46,6 +46,7 @@ impl ServiceDefinition for DockerContainer {
                             _ => true,
                         })
                 },
+                |_| Vec::new(),
                 "No other services with this container's ID have been matched",
                 "A service with this container's ID has already been matched",
                 MatchConfidence::Low,