discovery node - add port detection and subnet filter for passive scan (#1225)

Author: noursaidi

misc/discoverynode/src/nmap.py

@@ -8,6 +8,7 @@
 import logging
 import datetime
 import argparse
+import json
 
 state = mock.MagicMock()
 
@@ -22,7 +23,7 @@
 parser.add_argument("targets", help="nmap target", type=str)
 args = parser.parse_args()
 
-a = udmi.discovery.ether.EtherDiscovery(state, print)
+a = udmi.discovery.ether.EtherDiscovery(state, lambda x: print(x.to_json()))
 a.controller(
   {
     "discovery": {

misc/discoverynode/src/passive.py

@@ -7,6 +7,8 @@
 import sys
 import logging
 import datetime
+import json
+import argparse
 
 state = mock.MagicMock()
 
@@ -17,7 +19,11 @@
 stderr.setLevel(logging.WARNING)
 logging.root.setLevel(logging.INFO)
 
-a = udmi.discovery.passive.PassiveNetworkDiscovery(state, print)
+parser = argparse.ArgumentParser(description="subnet cidr")
+parser.add_argument("subnet_filter", help="subnet_filter", type=str)
+args = parser.parse_args()
+
+a = udmi.discovery.passive.PassiveNetworkDiscovery(state, lambda x: print(x.to_json()), subnet_filter = args.subnet_filter)
 a.controller(
   {
     "discovery": {

misc/discoverynode/src/tests/test_integration.py

@@ -135,7 +135,7 @@ def test_nmap():
             "timestamp": timestamp_now(),
             "discovery": {
                 "families": {
-                    "ether": {"generation": timestamp_now(), "depth": "ports", "addrs": ["192.168.12.1/24"]}
+                    "ether": {"generation": timestamp_now(), "depth": "services", "addrs": ["192.168.12.1/24"]}
                 }
             },
         })

misc/discoverynode/src/udmi/core.py

@@ -175,7 +175,9 @@ def enable_discovery(self,*,bacnet=True,vendor=True,ipv4=True,ether=True):
 
     if ipv4:
       passive_discovery = udmi.discovery.passive.PassiveNetworkDiscovery(
-          self.state, self.publish_discovery
+          self.state,
+          self.publish_discovery,
+          subnet_filter = self.config.get("ip", {}).get("subnet_filter")
       )
 
       self.add_config_route(

misc/discoverynode/src/udmi/discovery/ether.py

@@ -61,7 +61,7 @@ def magic_bathometer(self, depth: str) -> str:
     match depth:
       case "ping":
         return "ping"
-      case "ports":
+      case "ports" | "services":
         return "nmap"
       case _:
         raise RuntimeError(f"unmatched depth {depth}")
@@ -147,20 +147,15 @@ def nmap_stop_discovery(self):
   def nmap_runner(self):
     OUTPUT_FILE = "nmaplocalhost.xml"
 
+    cmd = ["/usr/bin/nmap"]
+    if self.config.depth == "services":
+      cmd.extend(["--script", "banner", "-A"])
+    cmd.extend(["-p-", "-T3"])
+    cmd.extend(self.config.addrs)
+    cmd.extend(["-oX", OUTPUT_FILE, "--stats-every", "5s"])
+
     with subprocess.Popen(
-        [
-            "/usr/bin/nmap",
-            "--script",
-            "banner",
-            "-p-",
-            "-T4",
-            "-A",
-            *self.config.addrs,
-            "-oX",
-            OUTPUT_FILE,
-            "--stats-every",
-            "5s",
-        ],
+        cmd,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
         encoding="utf-8",
@@ -178,6 +173,7 @@ def nmap_runner(self):
             return
 
     logging.info("nmap scan complete, parsing results")
+
     for host in nmap.results_reader(OUTPUT_FILE):
       event = udmi.schema.discovery_event.DiscoveryEvent(
           generation=self.generation,

misc/discoverynode/src/udmi/discovery/passive.py

@@ -40,10 +40,11 @@ class PassiveNetworkDiscovery(discovery.DiscoveryController):
 
   family = "ipv4"
 
-  def __init__(self, state, publisher, *, interface=None):
+  def __init__(self, state, publisher, *, interface=None, subnet_filter=None):
 
     self.queue = queue.SimpleQueue()
-    self.interface = None
+    self.interface = interface
+    self.subnet_filter = subnet_filter
     self.addresses_seen = set()
     self.device_records = set()
     self.devices_records_published = set()
@@ -90,7 +91,6 @@ def start_discovery(self):
     self.devices_records_published.clear()
     self.device_records.clear()
     self.addresses_seen.clear()
-
     # Turn-on the queue worker to consume sniffed packets
     self.queue_thread = threading.Thread(
         target=self.queue_worker, args=[], daemon=True
@@ -103,8 +103,34 @@ def start_discovery(self):
     )
     self.service_thread.start()
 
+    if subnet := self.subnet_filter:
+      try:
+        iface = ipaddress.ip_interface(subnet)
+        network = iface.network
+        bpf_filter = f"ip and src net {network} and (dst net {network} or broadcast or multicast)"
+        # Exclude network and broadcast addresses from being discovered as source
+        bpf_filter += f" and not src host {network.network_address}"
+        if network.broadcast_address:
+          bpf_filter += f" and not src host {network.broadcast_address}"
+        # Assume first host is gateway and exclude it
+        if first_host := next(network.hosts(), None):
+          bpf_filter += f" and not src host {first_host}"
+        # Exclude our own IP if provided in the subnet filter
+        if iface.ip != network.network_address:
+          bpf_filter += f" and not src host {iface.ip}"
+      except ValueError:
+        logging.warning("Invalid subnet filter: %s", subnet)
+        bpf_filter = f"ip and src net {subnet} and (dst net {subnet} or broadcast or multicast)"
+    else:
+      bpf_filter = PRIVATE_IP_BPF_FILTER
+    logging.info("Using BPF filter: %s", bpf_filter)
+
     self.sniffer = scapy.sendrecv.AsyncSniffer(
-        prn=self.queue.put, store=False, iface=self.interface, started_callback=self.scapy_is_go, filter=PRIVATE_IP_BPF_FILTER
+        prn=self.queue.put,
+        store=False,
+        iface=self.interface,
+        started_callback=self.scapy_is_go,
+        filter=bpf_filter,
     )
 
     self.sniffer.start()
@@ -175,8 +201,10 @@ def queue_worker(self):
         if scapy.layers.inet.IP in item:
           self.ip_packets_seen += 1
 
-          # A packet "sees" two devices - the source and the destination
-          for x in ["src", "dst"]:
+          # A packet "sees" two devices - the source and the destination.
+          # However, we only care about the source, as the destination may
+          # not be online (e.g. if we are pinging it).
+          for x in ["src"]:
 
             if x == "src":
               if scapy.layers.inet.UDP in item and (