fixed linting with exceptions

kube_hunter/conf/parser.py

@@ -54,7 +54,7 @@ def parser_add_arguments(parser):
         "It supports both in-cluster config (when running as a pod), "
         "and a specific kubectl config file (use --kubeconfig to set this). "
         "By default, when this flag is set, it will use in-cluster config. "
-        "NOTE: this is automatically switched on in --pod mode."
+        "NOTE: this is automatically switched on in --pod mode.",
     )
 
     parser.add_argument(
@@ -63,7 +63,7 @@ def parser_add_arguments(parser):
         metavar="KUBECONFIG",
         default=None,
         help="Specify the kubeconfig file to use for Kubernetes nodes auto discovery "
-        " (to be used in conjuction with the --k8s-auto-discover-nodes flag."
+        " (to be used in conjuction with the --k8s-auto-discover-nodes flag.",
     )
 
     parser.add_argument("--active", action="store_true", help="Enables active hunting")

kube_hunter/modules/discovery/kubernetes_client.py

@@ -11,8 +11,8 @@ def list_all_k8s_cluster_nodes(kube_config=None, client=None):
         else:
             logger.info("Attempting to use in cluster Kubernetes config")
             kubernetes.config.load_incluster_config()
-    except kubernetes.config.config_exception.ConfigException:
-        logger.exception("Failed to initiate Kubernetes client")
+    except kubernetes.config.config_exception.ConfigException as ex:
+        logger.exception(f"Failed to initiate Kubernetes client: {ex}")
         return
 
     try:
@@ -23,5 +23,5 @@ def list_all_k8s_cluster_nodes(kube_config=None, client=None):
         for item in ret.items:
             for addr in item.status.addresses:
                 yield addr.address
-    except:
-        logger.exception("Failed to list nodes from Kubernetes")
+    except Exception as ex:
+        logger.exception(f"Failed to list nodes from Kubernetes: {ex}")

kube_hunter/modules/hunting/mounts.py

@@ -77,15 +77,17 @@ class VarLogMountHunter(Hunter):
             self.publish_event(WriteMountToVarLog(pods=pe_pods))
 
 
-@handler.subscribe(ExposedRunHandler)
+@handler.subscribe_many([ExposedRunHandler, WriteMountToVarLog])
 class ProveVarLogMount(ActiveHunter):
     """Prove /var/log Mount Hunter
     Tries to read /etc/shadow on the host by running commands inside a pod with host mount to /var/log
     """
 
     def __init__(self, event):
-        self.event = event
-        self.base_path = f"https://{self.event.host}:{self.event.port}"
+        self.write_mount_event = self.event.get_by_class(WriteMountToVarLog)
+        self.event = self.write_mount_event
+
+        self.base_path = f"https://{self.write_mount_event.host}:{self.write_mount_event.port}"
 
     def run(self, command, container):
         run_url = KubeletHandlers.RUN.value.format(
@@ -96,20 +98,6 @@ class ProveVarLogMount(ActiveHunter):
         )
         return self.event.session.post(f"{self.base_path}/{run_url}", verify=False).text
 
-    # TODO: replace with multiple subscription to WriteMountToVarLog as well
-    def get_varlog_mounters(self):
-        config = get_config()
-        logger.debug("accessing /pods manually on ProveVarLogMount")
-        pods = self.event.session.get(
-            f"{self.base_path}/" + KubeletHandlers.PODS.value,
-            verify=False,
-            timeout=config.network_timeout,
-        ).json()["items"]
-        for pod in pods:
-            volume = VarLogMountHunter(ExposedPodsHandler(pods=pods)).has_write_mount_to(pod, "/var/log")
-            if volume:
-                yield pod, volume
-
     def mount_path_from_mountname(self, pod, mount_name):
         """returns container name, and container mount path correlated to mount_name"""
         for container in pod["spec"]["containers"]:
@@ -138,7 +126,7 @@ class ProveVarLogMount(ActiveHunter):
         return content
 
     def execute(self):
-        for pod, volume in self.get_varlog_mounters():
+        for pod, volume in self.write_mount_event.pe_pods():
             for container, mount_path in self.mount_path_from_mountname(pod, volume["name"]):
                 logger.debug("Correlated container to mount_name")
                 cont = {

tests/discovery/test_k8s.py

@@ -1,10 +1,9 @@
 from kube_hunter.conf import Config, set_config
 
-set_config(Config())
-
 from kube_hunter.modules.discovery.kubernetes_client import list_all_k8s_cluster_nodes
 from unittest.mock import MagicMock, patch
 
+set_config(Config())
 
 
 def test_client_yields_ips():
@@ -18,7 +17,7 @@ def test_client_yields_ips():
     response.items[1].status.addresses = [MagicMock()]
     response.items[1].status.addresses[0].address = "127.0.0.3"
 
-    with patch('kubernetes.config.load_incluster_config') as m:
+    with patch("kubernetes.config.load_incluster_config") as m:
         output = list(list_all_k8s_cluster_nodes(client=client))
         m.assert_called_once()
 
@@ -26,6 +25,6 @@ def test_client_yields_ips():
 
 
 def test_client_uses_kubeconfig():
-    with patch('kubernetes.config.load_kube_config') as m:
+    with patch("kubernetes.config.load_kube_config") as m:
         list(list_all_k8s_cluster_nodes(kube_config="/location", client=MagicMock()))
         m.assert_called_once_with(config_file="/location")