Support ignoring IPs (#332)

* Support ignoring IPs

Closes #296

kube_hunter/__main__.py

@@ -33,7 +33,11 @@ def interactive_set_config():
     elif choice == "2":
         config.interface = True
     elif choice == "3":
-        config.cidr = input("CIDR (example - 192.168.1.0/24): ").replace(" ", "")
+        config.cidr = (
+            input("CIDR separated by a ',' (example - 192.168.0.0/16,!192.168.0.8/32,!192.168.1.0/24): ")
+            .replace(" ", "")
+            .split(",")
+        )
     else:
         return False
     return True

kube_hunter/conf/parser.py

@@ -18,7 +18,12 @@ def parse_args():
         "--include-patched-versions", action="store_true", help="Don't skip patched versions when scanning",
     )
 
-    parser.add_argument("--cidr", type=str, help="Set an ip range to scan, example: 192.168.0.0/16")
+    parser.add_argument(
+        "--cidr",
+        type=str,
+        help="Set an IP range to scan/ignore, example: '192.168.0.0/24,!192.168.0.8/32,!192.168.0.16/32'",
+    )
+
     parser.add_argument(
         "--mapping", action="store_true", help="Outputs only a mapping of the cluster's nodes",
     )
@@ -54,4 +59,7 @@ def parse_args():
 
     parser.add_argument("--network-timeout", type=float, default=5.0, help="network operations timeout")
 
-    return parser.parse_args()
+    args = parser.parse_args()
+    if args.cidr:
+        args.cidr = args.cidr.replace(" ", "").split(",")
+    return args

kube_hunter/modules/discovery/hosts.py

@@ -1,9 +1,10 @@
 import os
 import logging
 import requests
+import itertools
 
 from enum import Enum
-from netaddr import IPNetwork, IPAddress
+from netaddr import IPNetwork, IPAddress, AddrFormatError
 from netifaces import AF_INET, ifaddresses, interfaces
 from scapy.all import ICMP, IP, Ether, srp1
 
@@ -61,12 +62,27 @@ def __init__(self, pod=False, active=False, predefined_hosts=None):
 class HostDiscoveryHelpers:
     # generator, generating a subnet by given a cidr
     @staticmethod
-    def generate_subnet(ip, sn="24"):
-        logger.debug(f"HostDiscoveryHelpers.generate_subnet {ip}/{sn}")
-        subnet = f"{ip}/{sn}"
-        for ip in IPNetwork(subnet):
-            logger.debug(f"HostDiscoveryHelpers.generate_subnet yielding {ip}")
-            yield ip
+    def filter_subnet(subnet, ignore=None):
+        for ip in subnet:
+            if ignore and any(ip in s for s in ignore):
+                logger.debug(f"HostDiscoveryHelpers.filter_subnet ignoring {ip}")
+            else:
+                yield ip
+
+    @staticmethod
+    def generate_hosts(cidrs):
+        ignore = list()
+        scan = list()
+        for cidr in cidrs:
+            try:
+                if cidr.startswith("!"):
+                    ignore.append(IPNetwork(cidr[1:]))
+                else:
+                    scan.append(IPNetwork(cidr))
+            except AddrFormatError as e:
+                raise ValueError(f"Unable to parse CIDR {cidr}") from e
+
+        return itertools.chain.from_iterable(HostDiscoveryHelpers.filter_subnet(sb, ignore=ignore) for sb in scan)
 
 
 @handler.subscribe(RunningAsPodEvent)
@@ -97,7 +113,7 @@ def execute(self):
                 if self.event.kubeservicehost and self.event.kubeservicehost in IPNetwork(f"{ip}/{mask}"):
                     should_scan_apiserver = False
                 logger.debug(f"From pod scanning subnet {ip}/{mask}")
-                for ip in HostDiscoveryHelpers.generate_subnet(ip, mask):
+                for ip in IPNetwork(f"{ip}/{mask}"):
                     self.publish_event(NewHostEvent(host=ip, cloud=cloud))
             if should_scan_apiserver:
                 self.publish_event(NewHostEvent(host=IPAddress(self.event.kubeservicehost), cloud=cloud))
@@ -163,12 +179,7 @@ def __init__(self, event):
 
     def execute(self):
         if config.cidr:
-            try:
-                ip, sn = config.cidr.split("/")
-            except ValueError:
-                logger.exception(f'Unable to parse CIDR "{config.cidr}"')
-                return
-            for ip in HostDiscoveryHelpers.generate_subnet(ip, sn=sn):
+            for ip in HostDiscoveryHelpers.generate_hosts(config.cidr):
                 self.publish_event(NewHostEvent(host=ip))
         elif config.interface:
             self.scan_interfaces()
@@ -187,7 +198,7 @@ def generate_interfaces_subnet(self, sn="24"):
             for ip in [i["addr"] for i in ifaddresses(ifaceName).setdefault(AF_INET, [])]:
                 if not self.event.localhost and InterfaceTypes.LOCALHOST.value in ip.__str__():
                     continue
-                for ip in HostDiscoveryHelpers.generate_subnet(ip, sn):
+                for ip in IPNetwork(f"{ip}/{sn}"):
                     yield ip
 
 

tests/discovery/test_hosts.py

@@ -1,10 +1,13 @@
 import requests_mock
+import pytest
 
+from netaddr import IPNetwork, IPAddress
 from kube_hunter.modules.discovery.hosts import (
     FromPodHostDiscovery,
     RunningAsPodEvent,
     HostScanEvent,
     AzureMetadataApi,
+    HostDiscoveryHelpers,
 )
 from kube_hunter.core.events.types import NewHostEvent
 from kube_hunter.core.events import handler
@@ -70,3 +73,34 @@ def __init__(self, event):
 class testAzureMetadataApi(object):
     def __init__(self, event):
         assert config.azure
+
+
+class TestDiscoveryUtils:
+    @staticmethod
+    def test_generate_hosts_valid_cidr():
+        test_cidr = "192.168.0.0/24"
+        expected = set(IPNetwork(test_cidr))
+
+        actual = set(HostDiscoveryHelpers.generate_hosts([test_cidr]))
+
+        assert actual == expected
+
+    @staticmethod
+    def test_generate_hosts_valid_ignore():
+        remove = IPAddress("192.168.1.8")
+        scan = "192.168.1.0/24"
+        expected = set(ip for ip in IPNetwork(scan) if ip != remove)
+
+        actual = set(HostDiscoveryHelpers.generate_hosts([scan, f"!{str(remove)}"]))
+
+        assert actual == expected
+
+    @staticmethod
+    def test_generate_hosts_invalid_cidr():
+        with pytest.raises(ValueError):
+            list(HostDiscoveryHelpers.generate_hosts(["192..2.3/24"]))
+
+    @staticmethod
+    def test_generate_hosts_invalid_ignore():
+        with pytest.raises(ValueError):
+            list(HostDiscoveryHelpers.generate_hosts(["192.168.1.8", "!29.2..1/24"]))