Updated kramdown

Updated kramdown to a newer patched version, the old version was not patched to CVE-2020-14001

.github/workflows/lint.yml

@@ -0,0 +1,12 @@
+name: Lint
+
+on: [push, pull_request]
+
+jobs:
+  build:
+    runs-on: ubuntu-20.04
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+      - uses: pre-commit/action@v2.0.0

.github/workflows/release.yml

@@ -0,0 +1,52 @@
+on:
+  push:
+    # Sequence of patterns matched against refs/tags
+    tags:
+    - 'v*' # Push events to matching v*, i.e. v1.0, v20.15.10
+    
+name: Upload Release Asset
+
+jobs:
+  build:
+    name: Upload Release Asset
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.9'
+      
+      - name: Install dependencies
+        run: |
+          python -m pip install -U pip
+          python -m pip install -r requirements-dev.txt
+      
+      - name: Build project
+        shell: bash
+        run: |
+          make pyinstaller
+      
+      - name: Create Release
+        id: create_release
+        uses: actions/create-release@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          tag_name: ${{ github.ref }}
+          release_name: Release ${{ github.ref }}
+          draft: false
+          prerelease: false
+          
+      - name: Upload Release Asset
+        id: upload-release-asset 
+        uses: actions/upload-release-asset@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }}
+          asset_path: ./dist/kube-hunter
+          asset_name: kube-hunter-linux-x86_64-${{ github.ref }}
+          asset_content_type: application/octet-stream

.github/workflows/test.yml

@@ -0,0 +1,54 @@
+name: Test
+
+on: [push, pull_request]
+
+env:
+  FORCE_COLOR: 1
+
+jobs:
+  build:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.6", "3.7", "3.8", "3.9"]
+        os: [ubuntu-20.04, ubuntu-18.04, ubuntu-16.04]
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Get pip cache dir
+        id: pip-cache
+        run: |
+          echo "::set-output name=dir::$(pip cache dir)"
+
+      - name: Cache
+        uses: actions/cache@v2
+        with:
+          path: ${{ steps.pip-cache.outputs.dir }}
+          key:
+            ${{ matrix.os }}-${{ matrix.python-version }}-${{ hashFiles('requirements-dev.txt') }}
+          restore-keys: |
+            ${{ matrix.os }}-${{ matrix.python-version }}-
+
+      - name: Install dependencies
+        run: |
+          python -m pip install -U pip
+          python -m pip install -U wheel
+          python -m pip install -r requirements.txt
+          python -m pip install -r requirements-dev.txt
+
+      - name: Test
+        shell: bash
+        run: |
+          make test
+
+      - name: Upload coverage
+        uses: codecov/codecov-action@v1
+        with:
+          name: ${{ matrix.os }} Python ${{ matrix.python-version }}

.gitignore

@@ -24,6 +24,7 @@ var/
 *.egg
 *.spec
 .eggs
+pip-wheel-metadata
 
 # Directory Cache Files
 .DS_Store

.travis.yml

@@ -1,20 +0,0 @@
-group: travis_latest
-language: python
-cache: pip
-python:
-    - "3.6"
-    - "3.7"
-    - "3.8"
-install:
-  - pip install -r requirements.txt
-  - pip install -r requirements-dev.txt
-before_script:
-  - make lint-check
-script:
-  - make test
-after_success:
-  -  bash <(curl -s https://codecov.io/bash)
-notifications:
-  email:
-    on_success: change
-    on_failure: always

README.md

@@ -1,6 +1,6 @@
 ![kube-hunter](https://github.com/aquasecurity/kube-hunter/blob/master/kube-hunter.png)
 
-[![Build Status](https://travis-ci.org/aquasecurity/kube-hunter.svg?branch=master)](https://travis-ci.org/aquasecurity/kube-hunter)
+[![Build Status](https://github.com/aquasecurity/kube-hunter/workflows/Test/badge.svg)](https://github.com/aquasecurity/kube-hunter/actions)
 [![codecov](https://codecov.io/gh/aquasecurity/kube-hunter/branch/master/graph/badge.svg)](https://codecov.io/gh/aquasecurity/kube-hunter)
 [![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
 [![License](https://img.shields.io/github/license/aquasecurity/kube-hunter)](https://github.com/aquasecurity/kube-hunter/blob/master/LICENSE)
@@ -34,6 +34,7 @@ Table of Contents
       * [Prerequisites](#prerequisites)
    * [Container](#container)
    * [Pod](#pod)
+* [Contribution](#contribution)
          
 ## Hunting
 
@@ -174,5 +175,8 @@ The example `job.yaml` file defines a Job that will run kube-hunter in a pod, us
 * Find the pod name with `kubectl describe job kube-hunter`
 * View the test results with `kubectl logs <pod name>`
 
+## Contribution 
+To read the contribution guidelines, <a href="https://github.com/aquasecurity/kube-hunter/blob/master/CONTRIBUTING.md"> Click here </a>
+
 ## License
 This repository is available under the [Apache License 2.0](https://github.com/aquasecurity/kube-hunter/blob/master/LICENSE).

docs/Gemfile.lock

@@ -67,7 +67,7 @@ GEM
       jekyll-theme-time-machine (= 0.1.1)
       jekyll-titles-from-headings (= 0.5.1)
       jemoji (= 0.10.2)
-      kramdown (= 1.17.0)
+      kramdown (>= 2.3.0)
       liquid (= 4.0.0)
       listen (= 3.1.5)
       mercenary (~> 0.3)

docs/_kb/KHV005.md

@@ -12,7 +12,7 @@ Kubernetes API was accessed with Pod Service Account or without Authentication (
 
 ## Remediation
 
-Secure acess to your Kubernetes API.
+Secure access to your Kubernetes API.
 
 It is recommended to explicitly specify a Service Account for all of your workloads (`serviceAccountName` in `Pod.Spec`), and manage their permissions according to the least privilege principal.
 
@@ -21,4 +21,4 @@ Consider opting out automatic mounting of SA token using `automountServiceAccoun
 
 ## References
 
-- [Configure Service Accounts for Pods](https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/)
+- [Configure Service Accounts for Pods](https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/)

docs/_kb/KHV052.md

@@ -0,0 +1,23 @@
+---
+vid: KHV052
+title: Exposed Pods
+categories: [Information Disclosure]
+---
+
+# {{ page.vid }} - {{ page.title }}
+
+## Issue description
+
+An attacker could view sensitive information about pods that are bound to a Node using the exposed /pods endpoint
+This can be done either by accessing the readonly port (default 10255), or from the secure kubelet port (10250)
+
+## Remediation
+
+Ensure kubelet is protected using `--anonymous-auth=false` kubelet flag. Allow only legitimate users using `--client-ca-file` or `--authentication-token-webhook` kubelet flags. This is usually done by the installer or cloud provider.
+
+Disable the readonly port by using `--read-only-port=0` kubelet flag.
+
+## References
+
+- [Kubelet configuration](https://kubernetes.io/docs/reference/command-line-tools-reference/kubelet/)
+- [Kubelet authentication/authorization](https://kubernetes.io/docs/reference/command-line-tools-reference/kubelet-authentication-authorization/)

kube_hunter/__main__.py

@@ -18,6 +18,7 @@ config = Config(
     cidr=args.cidr,
     include_patched_versions=args.include_patched_versions,
     interface=args.interface,
+    log_file=args.log_file,
     mapping=args.mapping,
     network_timeout=args.network_timeout,
     pod=args.pod,
@@ -25,7 +26,7 @@ config = Config(
     remote=args.remote,
     statistics=args.statistics,
 )
-setup_logger(args.log)
+setup_logger(args.log, args.log_file)
 set_config(config)
 
 # Running all other registered plugins before execution
@@ -72,13 +73,13 @@ def list_hunters():
     print("\nPassive Hunters:\n----------------")
     for hunter, docs in handler.passive_hunters.items():
         name, doc = hunter.parse_docs(docs)
-        print("* {}\n  {}\n".format(name, doc))
+        print(f"* {name}\n  {doc}\n")
 
     if config.active:
         print("\n\nActive Hunters:\n---------------")
         for hunter, docs in handler.active_hunters.items():
             name, doc = hunter.parse_docs(docs)
-            print("* {}\n  {}\n".format(name, doc))
+            print(f"* {name}\n  {doc}\n")
 
 
 hunt_started_lock = threading.Lock()

kube_hunter/conf/__init__.py

@@ -13,6 +13,7 @@ class Config:
     - interface: Interface scanning mode
     - list_hunters: Print a list of existing hunters
     - log_level: Log level
+    - log_file: Log File path
     - mapping: Report only found components
     - network_timeout: Timeout for network operations
     - pod: From pod scanning mode
@@ -27,6 +28,7 @@ class Config:
     dispatcher: Optional[Any] = None
     include_patched_versions: bool = False
     interface: bool = False
+    log_file: Optional[str] = None
     mapping: bool = False
     network_timeout: float = 5.0
     pod: bool = False

kube_hunter/conf/logging.py

@@ -1,6 +1,5 @@
 import logging
 
-
 DEFAULT_LEVEL = logging.INFO
 DEFAULT_LEVEL_NAME = logging.getLevelName(DEFAULT_LEVEL)
 LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s %(message)s"
@@ -10,7 +9,7 @@ logging.getLogger("scapy.runtime").setLevel(logging.CRITICAL)
 logging.getLogger("scapy.loading").setLevel(logging.CRITICAL)
 
 
-def setup_logger(level_name):
+def setup_logger(level_name, logfile):
     # Remove any existing handlers
     # Unnecessary in Python 3.8 since `logging.basicConfig` has `force` parameter
     for h in logging.getLogger().handlers[:]:
@@ -22,6 +21,9 @@ def setup_logger(level_name):
     else:
         log_level = getattr(logging, level_name.upper(), None)
         log_level = log_level if isinstance(log_level, int) else None
-        logging.basicConfig(level=log_level or DEFAULT_LEVEL, format=LOG_FORMAT)
+        if logfile is None:
+            logging.basicConfig(level=log_level or DEFAULT_LEVEL, format=LOG_FORMAT)
+        else:
+            logging.basicConfig(filename=logfile, level=log_level or DEFAULT_LEVEL, format=LOG_FORMAT)
         if not log_level:
             logging.warning(f"Unknown log level '{level_name}', using {DEFAULT_LEVEL_NAME}")

kube_hunter/conf/parser.py

@@ -56,6 +56,13 @@ def parser_add_arguments(parser):
         help="Set log level, options are: debug, info, warn, none",
     )
 
+    parser.add_argument(
+        "--log-file",
+        type=str,
+        default=None,
+        help="Path to a log file to output all logs to",
+    )
+
     parser.add_argument(
         "--report",
         type=str,

kube_hunter/core/events/handler.py

@@ -14,7 +14,7 @@ logger = logging.getLogger(__name__)
 # Inherits Queue object, handles events asynchronously
 class EventQueue(Queue):
     def __init__(self, num_worker=10):
-        super(EventQueue, self).__init__()
+        super().__init__()
         self.passive_hunters = dict()
         self.active_hunters = dict()
         self.all_hunters = dict()

kube_hunter/modules/discovery/hosts.py

@@ -5,8 +5,7 @@ import requests
 
 from enum import Enum
 from netaddr import IPNetwork, IPAddress, AddrFormatError
-from netifaces import AF_INET, ifaddresses, interfaces
-from scapy.all import ICMP, IP, Ether, srp1
+from netifaces import AF_INET, ifaddresses, interfaces, gateways
 
 from kube_hunter.conf import get_config
 from kube_hunter.core.events import handler
@@ -37,7 +36,7 @@ class RunningAsPodEvent(Event):
         try:
             with open(f"/var/run/secrets/kubernetes.io/serviceaccount/{file}") as f:
                 return f.read()
-        except IOError:
+        except OSError:
             pass
 
 
@@ -53,7 +52,7 @@ class AzureMetadataApi(Vulnerability, Event):
             vid="KHV003",
         )
         self.cidr = cidr
-        self.evidence = "cidr: {}".format(cidr)
+        self.evidence = f"cidr: {cidr}"
 
 
 class HostScanEvent(Event):
@@ -109,7 +108,7 @@ class FromPodHostDiscovery(Discovery):
             if self.is_azure_pod():
                 subnets, cloud = self.azure_metadata_discovery()
             else:
-                subnets = self.traceroute_discovery()
+                subnets = self.gateway_discovery()
 
             should_scan_apiserver = False
             if self.event.kubeservicehost:
@@ -141,14 +140,9 @@ class FromPodHostDiscovery(Discovery):
             return False
 
     # for pod scanning
-    def traceroute_discovery(self):
-        config = get_config()
-        node_internal_ip = srp1(
-            Ether() / IP(dst="1.1.1.1", ttl=1) / ICMP(),
-            verbose=0,
-            timeout=config.network_timeout,
-        )[IP].src
-        return [[node_internal_ip, "24"]]
+    def gateway_discovery(self):
+        """ Retrieving default gateway of pod, which is usually also a contact point with the host """
+        return [[gateways()["default"][AF_INET][0], "24"]]
 
     # querying azure's interface metadata api | works only from a pod
     def azure_metadata_discovery(self):

kube_hunter/modules/hunting/aks.py

@@ -46,11 +46,16 @@ class AzureSpnHunter(Hunter):
             logger.debug("failed getting pod info")
         else:
             pods_data = r.json().get("items", [])
+            suspicious_volume_names = []
             for pod_data in pods_data:
-                for container in pod_data["spec"]["containers"]:
-                    for mount in container["volumeMounts"]:
-                        path = mount["mountPath"]
+                for volume in pod_data["spec"].get("volumes", []):
+                    if volume.get("hostPath"):
+                        path = volume["hostPath"]["path"]
                         if "/etc/kubernetes/azure.json".startswith(path):
+                            suspicious_volume_names.append(volume["name"])
+                for container in pod_data["spec"]["containers"]:
+                    for mount in container.get("volumeMounts", []):
+                        if mount["name"] in suspicious_volume_names:
                             return {
                                 "name": container["name"],
                                 "pod": pod_data["metadata"]["name"],

kube_hunter/modules/hunting/apiserver.py

@@ -56,16 +56,19 @@ class ServerApiHTTPAccess(Vulnerability, Event):
 
 
 class ApiInfoDisclosure(Vulnerability, Event):
+    """Information Disclosure depending upon RBAC permissions and Kube-Cluster Setup"""
+
     def __init__(self, evidence, using_token, name):
+        category = InformationDisclosure
         if using_token:
-            name += " using service account token"
+            name += " using default service account token"
         else:
             name += " as anonymous user"
         Vulnerability.__init__(
             self,
             KubernetesCluster,
             name=name,
-            category=InformationDisclosure,
+            category=category,
             vid="KHV007",
         )
         self.evidence = evidence
@@ -343,7 +346,7 @@ class AccessApiServer(Hunter):
             else:
                 self.publish_event(ServerApiAccess(api, self.with_token))
 
-        namespaces = self.get_items("{path}/api/v1/namespaces".format(path=self.path))
+        namespaces = self.get_items(f"{self.path}/api/v1/namespaces")
         if namespaces:
             self.publish_event(ListNamespaces(namespaces, self.with_token))
 
@@ -371,7 +374,7 @@ class AccessApiServerWithToken(AccessApiServer):
     """
 
     def __init__(self, event):
-        super(AccessApiServerWithToken, self).__init__(event)
+        super().__init__(event)
         assert self.event.auth_token
         self.headers = {"Authorization": f"Bearer {self.event.auth_token}"}
         self.category = InformationDisclosure

kube_hunter/modules/hunting/arp.py

@@ -43,7 +43,7 @@ class ArpSpoofHunter(ActiveHunter):
     def detect_l3_on_host(self, arp_responses):
         """ returns True for an existence of an L3 network plugin """
         logger.debug("Attempting to detect L3 network plugin using ARP")
-        unique_macs = list(set(response[ARP].hwsrc for _, response in arp_responses))
+        unique_macs = list({response[ARP].hwsrc for _, response in arp_responses})
 
         # if LAN addresses not unique
         if len(unique_macs) == 1:

kube_hunter/modules/hunting/certificates.py

@@ -8,11 +8,13 @@ from kube_hunter.core.events import handler
 from kube_hunter.core.events.types import Vulnerability, Event, Service
 
 logger = logging.getLogger(__name__)
-email_pattern = re.compile(rb"([a-z0-9]+@[a-z0-9]+\.[a-z0-9]+)")
+email_pattern = re.compile(rb"([a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)")
 
 
 class CertificateEmail(Vulnerability, Event):
-    """Certificate includes an email address"""
+    """The Kubernetes API Server advertises a public certificate for TLS.
+    This certificate includes an email address, that may provide additional information for an attacker on your
+    organization, or be abused for further email based attacks."""
 
     def __init__(self, email):
         Vulnerability.__init__(
@@ -23,7 +25,7 @@ class CertificateEmail(Vulnerability, Event):
             vid="KHV021",
         )
         self.email = email
-        self.evidence = "email: {}".format(self.email)
+        self.evidence = f"email: {self.email}"
 
 
 @handler.subscribe(Service)

kube_hunter/modules/hunting/cves.py

@@ -104,7 +104,7 @@ class IncompleteFixToKubectlCpVulnerability(Vulnerability, Event):
             vid="KHV027",
         )
         self.binary_version = binary_version
-        self.evidence = "kubectl version: {}".format(self.binary_version)
+        self.evidence = f"kubectl version: {self.binary_version}"
 
 
 class KubectlCpVulnerability(Vulnerability, Event):
@@ -120,7 +120,7 @@ class KubectlCpVulnerability(Vulnerability, Event):
             vid="KHV028",
         )
         self.binary_version = binary_version
-        self.evidence = "kubectl version: {}".format(self.binary_version)
+        self.evidence = f"kubectl version: {self.binary_version}"
 
 
 class CveUtils:

kube_hunter/modules/hunting/dns.py

@@ -25,7 +25,7 @@ class PossibleDnsSpoofing(Vulnerability, Event):
             vid="KHV030",
         )
         self.kubedns_pod_ip = kubedns_pod_ip
-        self.evidence = "kube-dns at: {}".format(self.kubedns_pod_ip)
+        self.evidence = f"kube-dns at: {self.kubedns_pod_ip}"
 
 
 # Only triggered with RunningAsPod base event

kube_hunter/modules/hunting/kubelet.py

@@ -35,10 +35,7 @@ class ExposedPodsHandler(Vulnerability, Event):
 
     def __init__(self, pods):
         Vulnerability.__init__(
-            self,
-            component=Kubelet,
-            name="Exposed Pods",
-            category=InformationDisclosure,
+            self, component=Kubelet, name="Exposed Pods", category=InformationDisclosure, vid="KHV052"
         )
         self.pods = pods
         self.evidence = f"count: {len(self.pods)}"
@@ -84,7 +81,7 @@ class ExposedRunningPodsHandler(Vulnerability, Event):
             vid="KHV038",
         )
         self.count = count
-        self.evidence = "{} running pods".format(self.count)
+        self.evidence = f"{self.count} running pods"
 
 
 class ExposedExecHandler(Vulnerability, Event):
@@ -347,27 +344,23 @@ class SecureKubeletPortHunter(Hunter):
 
         # need further investigation on websockets protocol for further implementation
         def test_port_forward(self):
-            config = get_config()
-            headers = {
-                "Upgrade": "websocket",
-                "Connection": "Upgrade",
-                "Sec-Websocket-Key": "s",
-                "Sec-Websocket-Version": "13",
-                "Sec-Websocket-Protocol": "SPDY",
-            }
-            pf_url = self.path + KubeletHandlers.PORTFORWARD.value.format(
-                pod_namespace=self.pod["namespace"],
-                pod_id=self.pod["name"],
-                port=80,
-            )
-            self.session.get(
-                pf_url,
-                headers=headers,
-                verify=False,
-                stream=True,
-                timeout=config.network_timeout,
-            ).status_code == 200
+            pass
             # TODO: what to return?
+            # Example starting code:
+            #
+            # config = get_config()
+            # headers = {
+            #     "Upgrade": "websocket",
+            #     "Connection": "Upgrade",
+            #     "Sec-Websocket-Key": "s",
+            #     "Sec-Websocket-Version": "13",
+            #     "Sec-Websocket-Protocol": "SPDY",
+            # }
+            # pf_url = self.path + KubeletHandlers.PORTFORWARD.value.format(
+            #     pod_namespace=self.pod["namespace"],
+            #     pod_id=self.pod["name"],
+            #     port=80,
+            # )
 
         # executes one command and returns output
         def test_run_container(self):
@@ -378,8 +371,9 @@ class SecureKubeletPortHunter(Hunter):
                 container_name="test",
                 cmd="",
             )
-            # if we get a Method Not Allowed, we know we passed Authentication and Authorization.
-            return self.session.get(run_url, verify=False, timeout=config.network_timeout).status_code == 405
+            # if we get this message, we know we passed Authentication and Authorization, and that the endpoint is enabled.
+            status_code = self.session.post(run_url, verify=False, timeout=config.network_timeout).status_code
+            return status_code == requests.codes.NOT_FOUND
 
         # returns list of currently running pods
         def test_running_pods(self):
@@ -532,7 +526,7 @@ class ProveAnonymousAuth(ActiveHunter):
 
     def __init__(self, event):
         self.event = event
-        self.base_url = "https://{host}:10250/".format(host=self.event.host)
+        self.base_url = f"https://{self.event.host}:10250/"
 
     def get_request(self, url, verify=False):
         config = get_config()
@@ -571,7 +565,7 @@ class ProveAnonymousAuth(ActiveHunter):
         return ProveAnonymousAuth.has_no_error(result) and ProveAnonymousAuth.has_no_exception(result)
 
     def cat_command(self, run_request_url, full_file_path):
-        return self.post_request(run_request_url, {"cmd": "cat {}".format(full_file_path)})
+        return self.post_request(run_request_url, {"cmd": f"cat {full_file_path}"})
 
     def process_container(self, run_request_url):
         service_account_token = self.cat_command(run_request_url, "/var/run/secrets/kubernetes.io/serviceaccount/token")
@@ -608,7 +602,7 @@ class ProveAnonymousAuth(ActiveHunter):
                 for container_data in pod_data["spec"]["containers"]:
                     container_name = container_data["name"]
 
-                    run_request_url = self.base_url + "run/{}/{}/{}".format(pod_namespace, pod_id, container_name)
+                    run_request_url = self.base_url + f"run/{pod_namespace}/{pod_id}/{container_name}"
 
                     extracted_data = self.process_container(run_request_url)
 
@@ -617,11 +611,11 @@ class ProveAnonymousAuth(ActiveHunter):
                         environment_variables = extracted_data["environment_variables"]
 
                         temp_message += (
-                            "\n\nPod namespace: {}".format(pod_namespace)
-                            + "\n\nPod ID: {}".format(pod_id)
-                            + "\n\nContainer name: {}".format(container_name)
-                            + "\n\nService account token: {}".format(service_account_token)
-                            + "\nEnvironment variables: {}".format(environment_variables)
+                            f"\n\nPod namespace: {pod_namespace}"
+                            + f"\n\nPod ID: {pod_id}"
+                            + f"\n\nContainer name: {container_name}"
+                            + f"\n\nService account token: {service_account_token}"
+                            + f"\nEnvironment variables: {environment_variables}"
                         )
 
                         first_check = container_data.get("securityContext", {}).get("privileged")
@@ -646,7 +640,7 @@ class ProveAnonymousAuth(ActiveHunter):
             if temp_message:
                 message = "The following containers have been successfully breached." + temp_message
 
-                self.event.evidence = "{}".format(message)
+                self.event.evidence = f"{message}"
 
             if exposed_existing_privileged_containers:
                 self.publish_event(
@@ -666,7 +660,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
 
     def __init__(self, event, seconds_to_wait_for_os_command=1):
         self.event = event
-        self.base_url = "https://{host}:10250/".format(host=self.event.host)
+        self.base_url = f"https://{self.event.host}:10250/"
         self.seconds_to_wait_for_os_command = seconds_to_wait_for_os_command
         self.number_of_rm_attempts = 5
         self.number_of_rmdir_attempts = 5
@@ -685,7 +679,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
             return "Exception: " + str(ex)
 
     def cat_command(self, run_request_url, full_file_path):
-        return self.post_request(run_request_url, {"cmd": "cat {}".format(full_file_path)})
+        return self.post_request(run_request_url, {"cmd": f"cat {full_file_path}"})
 
     def clean_attacked_exposed_existing_privileged_container(
         self,
@@ -701,7 +695,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
 
         self.rm_command(
             run_request_url,
-            "{}/etc/cron.daily/{}".format(directory_created, file_created),
+            f"{directory_created}/etc/cron.daily/{file_created}",
             number_of_rm_attempts,
             seconds_to_wait_for_os_command,
         )
@@ -729,9 +723,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
     def rm_command(self, run_request_url, file_to_remove, number_of_rm_attempts, seconds_to_wait_for_os_command):
         if self.check_file_exists(run_request_url, file_to_remove):
             for _ in range(number_of_rm_attempts):
-                command_execution_outcome = self.post_request(
-                    run_request_url, {"cmd": "rm -f {}".format(file_to_remove)}
-                )
+                command_execution_outcome = self.post_request(run_request_url, {"cmd": f"rm -f {file_to_remove}"})
 
                 if seconds_to_wait_for_os_command:
                     time.sleep(seconds_to_wait_for_os_command)
@@ -758,10 +750,10 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
         return False
 
     def chmod_command(self, run_request_url, permissions, file):
-        return self.post_request(run_request_url, {"cmd": "chmod {} {}".format(permissions, file)})
+        return self.post_request(run_request_url, {"cmd": f"chmod {permissions} {file}"})
 
     def touch_command(self, run_request_url, file_to_create):
-        return self.post_request(run_request_url, {"cmd": "touch {}".format(file_to_create)})
+        return self.post_request(run_request_url, {"cmd": f"touch {file_to_create}"})
 
     def attack_exposed_existing_privileged_container(
         self, run_request_url, directory_created, number_of_rm_attempts, seconds_to_wait_for_os_command, file_name=None
@@ -769,7 +761,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
         if file_name is None:
             file_name = "kube-hunter" + str(uuid.uuid1())
 
-        file_name_with_path = "{}/etc/cron.daily/{}".format(directory_created, file_name)
+        file_name_with_path = f"{directory_created}/etc/cron.daily/{file_name}"
 
         file_created = self.touch_command(run_request_url, file_name_with_path)
 
@@ -797,9 +789,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
     ):
         if self.check_directory_exists(run_request_url, directory_to_remove):
             for _ in range(number_of_rmdir_attempts):
-                command_execution_outcome = self.post_request(
-                    run_request_url, {"cmd": "rmdir {}".format(directory_to_remove)}
-                )
+                command_execution_outcome = self.post_request(run_request_url, {"cmd": f"rmdir {directory_to_remove}"})
 
                 if seconds_to_wait_for_os_command:
                     time.sleep(seconds_to_wait_for_os_command)
@@ -826,7 +816,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
         return False
 
     def ls_command(self, run_request_url, file_or_directory):
-        return self.post_request(run_request_url, {"cmd": "ls {}".format(file_or_directory)})
+        return self.post_request(run_request_url, {"cmd": f"ls {file_or_directory}"})
 
     def umount_command(
         self,
@@ -844,7 +834,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
             for _ in range(number_of_umount_attempts):
                 # Ref: http://man7.org/linux/man-pages/man2/umount.2.html
                 command_execution_outcome = self.post_request(
-                    run_request_url, {"cmd": "umount {} {}".format(file_system_or_partition, directory)}
+                    run_request_url, {"cmd": f"umount {file_system_or_partition} {directory}"}
                 )
 
                 if seconds_to_wait_for_os_command:
@@ -875,16 +865,16 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
 
     def mount_command(self, run_request_url, file_system_or_partition, directory):
         # Ref: http://man7.org/linux/man-pages/man1/mkdir.1.html
-        return self.post_request(run_request_url, {"cmd": "mount {} {}".format(file_system_or_partition, directory)})
+        return self.post_request(run_request_url, {"cmd": f"mount {file_system_or_partition} {directory}"})
 
     def mkdir_command(self, run_request_url, directory_to_create):
         # Ref: http://man7.org/linux/man-pages/man1/mkdir.1.html
-        return self.post_request(run_request_url, {"cmd": "mkdir {}".format(directory_to_create)})
+        return self.post_request(run_request_url, {"cmd": f"mkdir {directory_to_create}"})
 
     def findfs_command(self, run_request_url, file_system_or_partition_type, file_system_or_partition):
         # Ref: http://man7.org/linux/man-pages/man8/findfs.8.html
         return self.post_request(
-            run_request_url, {"cmd": "findfs {}{}".format(file_system_or_partition_type, file_system_or_partition)}
+            run_request_url, {"cmd": f"findfs {file_system_or_partition_type}{file_system_or_partition}"}
         )
 
     def get_root_values(self, command_line):
@@ -943,9 +933,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
                             )
 
                             if ProveAnonymousAuth.has_no_error_nor_exception(mounted_file_system_or_partition):
-                                host_name = self.cat_command(
-                                    run_request_url, "{}/etc/hostname".format(directory_created)
-                                )
+                                host_name = self.cat_command(run_request_url, f"{directory_created}/etc/hostname")
 
                                 if ProveAnonymousAuth.has_no_error_nor_exception(host_name):
                                     return {
@@ -979,7 +967,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
             pod_id = exposed_existing_privileged_containers["pod_id"]
             container_name = exposed_existing_privileged_containers["container_name"]
 
-            run_request_url = self.base_url + "run/{}/{}/{}".format(pod_namespace, pod_id, container_name)
+            run_request_url = self.base_url + f"run/{pod_namespace}/{pod_id}/{container_name}"
 
             is_exposed_existing_privileged_container_privileged = self.process_exposed_existing_privileged_container(
                 run_request_url,
@@ -1029,7 +1017,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
                 + temp_message
             )
 
-            self.event.evidence = "{}".format(message)
+            self.event.evidence = f"{message}"
         else:
             message = (
                 "The following exposed existing privileged containers"
@@ -1038,7 +1026,7 @@ class MaliciousIntentViaSecureKubeletPort(ActiveHunter):
                 + temp_message
             )
 
-            self.event.evidence = "{}".format(message)
+            self.event.evidence = f"{message}"
 
 
 @handler.subscribe(ExposedRunHandler)
@@ -1145,11 +1133,16 @@ class ProveSystemLogs(ActiveHunter):
             f"{self.base_url}/" + KubeletHandlers.LOGS.value.format(path="audit/audit.log"),
             verify=False,
             timeout=config.network_timeout,
-        ).text
-        logger.debug(f"Audit log of host {self.event.host}: {audit_logs[:10]}")
-        # iterating over proctitles and converting them into readable strings
-        proctitles = []
-        for proctitle in re.findall(r"proctitle=(\w+)", audit_logs):
-            proctitles.append(bytes.fromhex(proctitle).decode("utf-8").replace("\x00", " "))
-        self.event.proctitles = proctitles
-        self.event.evidence = f"audit log: {proctitles}"
+        )
+
+        # TODO: add more methods for proving system logs
+        if audit_logs.status_code == requests.status_codes.codes.OK:
+            logger.debug(f"Audit log of host {self.event.host}: {audit_logs.text[:10]}")
+            # iterating over proctitles and converting them into readable strings
+            proctitles = []
+            for proctitle in re.findall(r"proctitle=(\w+)", audit_logs.text):
+                proctitles.append(bytes.fromhex(proctitle).decode("utf-8").replace("\x00", " "))
+            self.event.proctitles = proctitles
+            self.event.evidence = f"audit log: {proctitles}"
+        else:
+            self.event.evidence = "Could not parse system logs"

kube_hunter/modules/hunting/mounts.py

@@ -32,7 +32,7 @@ class WriteMountToVarLog(Vulnerability, Event):
             vid="KHV047",
         )
         self.pods = pods
-        self.evidence = "pods: {}".format(", ".join((pod["metadata"]["name"] for pod in self.pods)))
+        self.evidence = "pods: {}".format(", ".join(pod["metadata"]["name"] for pod in self.pods))
 
 
 class DirectoryTraversalWithKubelet(Vulnerability, Event):
@@ -47,7 +47,7 @@ class DirectoryTraversalWithKubelet(Vulnerability, Event):
             category=PrivilegeEscalation,
         )
         self.output = output
-        self.evidence = "output: {}".format(self.output)
+        self.evidence = f"output: {self.output}"
 
 
 @handler.subscribe(ExposedPodsHandler)

kube_hunter/modules/report/base.py

@@ -7,6 +7,9 @@ from kube_hunter.modules.report.collector import (
     vulnerabilities_lock,
 )
 
+BASE_KB_LINK = "https://avd.aquasec.com/"
+FULL_KB_LINK = "https://avd.aquasec.com/kube-hunter/{vid}/"
+
 
 class BaseReporter:
     def get_nodes(self):
@@ -38,6 +41,7 @@ class BaseReporter:
                     "vulnerability": vuln.get_name(),
                     "description": vuln.explain(),
                     "evidence": str(vuln.evidence),
+                    "avd_reference": FULL_KB_LINK.format(vid=vuln.get_vid().lower()),
                     "hunter": vuln.hunter.get_name(),
                 }
                 for vuln in vulnerabilities
@@ -63,6 +67,4 @@ class BaseReporter:
         if statistics:
             report["hunter_statistics"] = self.get_hunter_statistics()
 
-        report["kburl"] = "https://aquasecurity.github.io/kube-hunter/kb/{vid}"
-
         return report

kube_hunter/modules/report/plain.py

@@ -1,6 +1,6 @@
 from prettytable import ALL, PrettyTable
 
-from kube_hunter.modules.report.base import BaseReporter
+from kube_hunter.modules.report.base import BaseReporter, BASE_KB_LINK
 from kube_hunter.modules.report.collector import (
     services,
     vulnerabilities,
@@ -9,9 +9,8 @@ from kube_hunter.modules.report.collector import (
     vulnerabilities_lock,
 )
 
-EVIDENCE_PREVIEW = 40
+EVIDENCE_PREVIEW = 100
 MAX_TABLE_WIDTH = 20
-KB_LINK = "https://github.com/aquasecurity/kube-hunter/tree/master/docs/_kb"
 
 
 class PlainReporter(BaseReporter):
@@ -60,7 +59,7 @@ class PlainReporter(BaseReporter):
             if service.event_id not in id_memory:
                 nodes_table.add_row(["Node/Master", service.host])
                 id_memory.add(service.event_id)
-        nodes_ret = "\nNodes\n{}\n".format(nodes_table)
+        nodes_ret = f"\nNodes\n{nodes_table}\n"
         services_lock.release()
         return nodes_ret
 
@@ -114,7 +113,7 @@ class PlainReporter(BaseReporter):
         return (
             "\nVulnerabilities\n"
             "For further information about a vulnerability, search its ID in: \n"
-            f"{KB_LINK}\n{vuln_table}\n"
+            f"{BASE_KB_LINK}\n{vuln_table}\n"
         )
 
     def hunters_table(self):

pyinstaller_hooks/hook-prettytable.py

@@ -0,0 +1,3 @@
+from PyInstaller.utils.hooks import collect_all
+
+datas, binaries, hiddenimports = collect_all("prettytable")

setup.cfg

@@ -22,6 +22,8 @@ classifiers =
     Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3 :: Only
     Topic :: Security
 
 [options]

setup.py

@@ -41,6 +41,8 @@ class PyInstallerCommand(Command):
         cfg.read("setup.cfg")
         command = [
             "pyinstaller",
+            "--additional-hooks-dir",
+            "pyinstaller_hooks",
             "--clean",
             "--onefile",
             "--name",

tests/conf/test_logging.py

@@ -11,12 +11,13 @@ def test_setup_logger_level():
         ("NOTEXISTS", logging.INFO),
         ("BASIC_FORMAT", logging.INFO),
     ]
+    logFile = None
     for level, expected in test_cases:
-        setup_logger(level)
+        setup_logger(level, logFile)
         actual = logging.getLogger().getEffectiveLevel()
         assert actual == expected, f"{level} level should be {expected} (got {actual})"
 
 
 def test_setup_logger_none():
-    setup_logger("NONE")
+    setup_logger("NONE", None)
     assert logging.getLogger().manager.disable == logging.CRITICAL

tests/discovery/test_apiserver.py

@@ -123,7 +123,7 @@ def test_InsecureApiServer():
 
 # We should only generate an ApiServer event for a response that looks like it came from a Kubernetes node
 @handler.subscribe(ApiServer)
-class testApiServer(object):
+class testApiServer:
     def __init__(self, event):
         print("Event")
         assert event.host == "mockKubernetes"

tests/discovery/test_hosts.py

@@ -90,7 +90,7 @@ class TestDiscoveryUtils:
     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)
+        expected = {ip for ip in IPNetwork(scan) if ip != remove}
 
         actual = set(HostDiscoveryHelpers.generate_hosts([scan, f"!{str(remove)}"]))
 

tests/hunting/test_aks.py

@@ -0,0 +1,56 @@
+# flake8: noqa: E402
+import requests_mock
+
+from kube_hunter.conf import Config, set_config
+
+set_config(Config())
+
+from kube_hunter.modules.hunting.kubelet import ExposedRunHandler
+from kube_hunter.modules.hunting.aks import AzureSpnHunter
+
+
+def test_AzureSpnHunter():
+    e = ExposedRunHandler()
+    e.host = "mockKubernetes"
+    e.port = 443
+    e.protocol = "https"
+
+    pod_template = '{{"items":[ {{"apiVersion":"v1","kind":"Pod","metadata":{{"name":"etc","namespace":"default"}},"spec":{{"containers":[{{"command":["sleep","99999"],"image":"ubuntu","name":"test","volumeMounts":[{{"mountPath":"/mp","name":"v"}}]}}],"volumes":[{{"hostPath":{{"path":"{}"}},"name":"v"}}]}}}} ]}}'
+
+    bad_paths = ["/", "/etc", "/etc/", "/etc/kubernetes", "/etc/kubernetes/azure.json"]
+    good_paths = ["/yo", "/etc/yo", "/etc/kubernetes/yo.json"]
+
+    for p in bad_paths:
+        with requests_mock.Mocker() as m:
+            m.get("https://mockKubernetes:443/pods", text=pod_template.format(p))
+            h = AzureSpnHunter(e)
+            c = h.get_key_container()
+            assert c
+
+    for p in good_paths:
+        with requests_mock.Mocker() as m:
+            m.get("https://mockKubernetes:443/pods", text=pod_template.format(p))
+            h = AzureSpnHunter(e)
+            c = h.get_key_container()
+            assert c == None
+
+    with requests_mock.Mocker() as m:
+        pod_no_volume_mounts = '{"items":[ {"apiVersion":"v1","kind":"Pod","metadata":{"name":"etc","namespace":"default"},"spec":{"containers":[{"command":["sleep","99999"],"image":"ubuntu","name":"test"}],"volumes":[{"hostPath":{"path":"/whatever"},"name":"v"}]}} ]}'
+        m.get("https://mockKubernetes:443/pods", text=pod_no_volume_mounts)
+        h = AzureSpnHunter(e)
+        c = h.get_key_container()
+        assert c == None
+
+    with requests_mock.Mocker() as m:
+        pod_no_volumes = '{"items":[ {"apiVersion":"v1","kind":"Pod","metadata":{"name":"etc","namespace":"default"},"spec":{"containers":[{"command":["sleep","99999"],"image":"ubuntu","name":"test"}]}} ]}'
+        m.get("https://mockKubernetes:443/pods", text=pod_no_volumes)
+        h = AzureSpnHunter(e)
+        c = h.get_key_container()
+        assert c == None
+
+    with requests_mock.Mocker() as m:
+        pod_other_volume = '{"items":[ {"apiVersion":"v1","kind":"Pod","metadata":{"name":"etc","namespace":"default"},"spec":{"containers":[{"command":["sleep","99999"],"image":"ubuntu","name":"test","volumeMounts":[{"mountPath":"/mp","name":"v"}]}],"volumes":[{"emptyDir":{},"name":"v"}]}} ]}'
+        m.get("https://mockKubernetes:443/pods", text=pod_other_volume)
+        h = AzureSpnHunter(e)
+        c = h.get_key_container()
+        assert c == None

tests/hunting/test_apiserver_hunter.py

@@ -122,7 +122,7 @@ def test_AccessApiServer():
 
 
 @handler.subscribe(ListNamespaces)
-class test_ListNamespaces(object):
+class test_ListNamespaces:
     def __init__(self, event):
         print("ListNamespaces")
         assert event.evidence == ["hello"]
@@ -135,7 +135,7 @@ class test_ListNamespaces(object):
 
 
 @handler.subscribe(ListPodsAndNamespaces)
-class test_ListPodsAndNamespaces(object):
+class test_ListPodsAndNamespaces:
     def __init__(self, event):
         print("ListPodsAndNamespaces")
         assert len(event.evidence) == 2
@@ -158,7 +158,7 @@ class test_ListPodsAndNamespaces(object):
 
 # Should never see this because the API call in the test returns 403 status code
 @handler.subscribe(ListRoles)
-class test_ListRoles(object):
+class test_ListRoles:
     def __init__(self, event):
         print("ListRoles")
         assert 0
@@ -169,7 +169,7 @@ class test_ListRoles(object):
 # Should only see this when we have a token because the API call returns an empty list of items
 # in the test where we have no token
 @handler.subscribe(ListClusterRoles)
-class test_ListClusterRoles(object):
+class test_ListClusterRoles:
     def __init__(self, event):
         print("ListClusterRoles")
         assert event.auth_token == "so-secret"
@@ -178,7 +178,7 @@ class test_ListClusterRoles(object):
 
 
 @handler.subscribe(ServerApiAccess)
-class test_ServerApiAccess(object):
+class test_ServerApiAccess:
     def __init__(self, event):
         print("ServerApiAccess")
         if event.category == UnauthenticatedAccess:
@@ -191,7 +191,7 @@ class test_ServerApiAccess(object):
 
 
 @handler.subscribe(ApiServerPassiveHunterFinished)
-class test_PassiveHunterFinished(object):
+class test_PassiveHunterFinished:
     def __init__(self, event):
         print("PassiveHunterFinished")
         assert event.namespaces == ["hello"]
@@ -276,12 +276,12 @@ def test_AccessApiServerActive():
 
 
 @handler.subscribe(CreateANamespace)
-class test_CreateANamespace(object):
+class test_CreateANamespace:
     def __init__(self, event):
         assert "abcde" in event.evidence
 
 
 @handler.subscribe(DeleteANamespace)
-class test_DeleteANamespace(object):
+class test_DeleteANamespace:
     def __init__(self, event):
         assert "2019-02-26" in event.evidence

tests/hunting/test_certificates.py

@@ -37,6 +37,6 @@ rceJuGsnJEQ=
 
 
 @handler.subscribe(CertificateEmail)
-class test_CertificateEmail(object):
+class test_CertificateEmail:
     def __init__(self, event):
         assert event.email == b"build@nodejs.org0"

tests/hunting/test_cvehunting.py

@@ -41,7 +41,7 @@ def test_K8sCveHunter():
 
 
 @handler.subscribe(ServerApiVersionEndPointAccessPE)
-class test_CVE_2018_1002105(object):
+class test_CVE_2018_1002105:
     def __init__(self, event):
         global cve_counter
         cve_counter += 1

tests/hunting/test_kubelet.py

@@ -270,7 +270,7 @@ def test_proveanonymousauth_connectivity_issues():
 
 
 @handler.subscribe(ExposedExistingPrivilegedContainersViaSecureKubeletPort)
-class ExposedPrivilegedContainersViaAnonymousAuthEnabledInSecureKubeletPortEventCounter(object):
+class ExposedPrivilegedContainersViaAnonymousAuthEnabledInSecureKubeletPortEventCounter:
     def __init__(self, event):
         global counter
         counter += 1
@@ -371,9 +371,9 @@ def test_attack_exposed_existing_privileged_container_success():
         run_url = url + "run/kube-hunter-privileged/kube-hunter-privileged-deployment-86dc79f945-sjjps/ubuntu?cmd="
         directory_created = "/kube-hunter-mock_" + str(uuid.uuid1())
         file_name = "kube-hunter-mock" + str(uuid.uuid1())
-        file_name_with_path = "{}/etc/cron.daily/{}".format(directory_created, file_name)
+        file_name_with_path = f"{directory_created}/etc/cron.daily/{file_name}"
 
-        session_mock.post(run_url + urllib.parse.quote("touch {}".format(file_name_with_path), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"touch {file_name_with_path}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("chmod {} {}".format("755", file_name_with_path), safe=""), text=""
         )
@@ -395,12 +395,12 @@ def test_attack_exposed_existing_privileged_container_failure_when_touch():
     with requests_mock.Mocker(session=class_being_tested.event.session) as session_mock:
         directory_created = "/kube-hunter-mock_" + str(uuid.uuid1())
         file_name = "kube-hunter-mock" + str(uuid.uuid1())
-        file_name_with_path = "{}/etc/cron.daily/{}".format(directory_created, file_name)
+        file_name_with_path = f"{directory_created}/etc/cron.daily/{file_name}"
 
         url = "https://localhost:10250/"
         run_url = url + "run/kube-hunter-privileged/kube-hunter-privileged-deployment-86dc79f945-sjjps/ubuntu?cmd="
         session_mock.post(
-            run_url + urllib.parse.quote("touch {}".format(file_name_with_path), safe=""),
+            run_url + urllib.parse.quote(f"touch {file_name_with_path}", safe=""),
             text="Operation not permitted",
         )
 
@@ -420,11 +420,11 @@ def test_attack_exposed_existing_privileged_container_failure_when_chmod():
     with requests_mock.Mocker(session=class_being_tested.event.session) as session_mock:
         directory_created = "/kube-hunter-mock_" + str(uuid.uuid1())
         file_name = "kube-hunter-mock" + str(uuid.uuid1())
-        file_name_with_path = "{}/etc/cron.daily/{}".format(directory_created, file_name)
+        file_name_with_path = f"{directory_created}/etc/cron.daily/{file_name}"
 
         url = "https://localhost:10250/"
         run_url = url + "run/kube-hunter-privileged/kube-hunter-privileged-deployment-86dc79f945-sjjps/ubuntu?cmd="
-        session_mock.post(run_url + urllib.parse.quote("touch {}".format(file_name_with_path), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"touch {file_name_with_path}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("chmod {} {}".format("755", file_name_with_path), safe=""),
             text="Permission denied",
@@ -547,12 +547,12 @@ def test_process_exposed_existing_privileged_container_success():
 
         session_mock.post(run_url + urllib.parse.quote("cat /proc/cmdline", safe=""), text=cat_proc_cmdline)
         session_mock.post(run_url + urllib.parse.quote("findfs LABEL=Mock", safe=""), text="/dev/mock_fs")
-        session_mock.post(run_url + urllib.parse.quote("mkdir {}".format(directory_created), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"mkdir {directory_created}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("mount {} {}".format("/dev/mock_fs", directory_created), safe=""), text=""
         )
         session_mock.post(
-            run_url + urllib.parse.quote("cat {}/etc/hostname".format(directory_created), safe=""), text="mockhostname"
+            run_url + urllib.parse.quote(f"cat {directory_created}/etc/hostname", safe=""), text="mockhostname"
         )
 
         return_value = class_being_tested.process_exposed_existing_privileged_container(
@@ -619,9 +619,7 @@ def test_process_exposed_existing_privileged_container_failure_when_mkdir():
 
         session_mock.post(run_url + urllib.parse.quote("cat /proc/cmdline", safe=""), text=cat_proc_cmdline)
         session_mock.post(run_url + urllib.parse.quote("findfs LABEL=Mock", safe=""), text="/dev/mock_fs")
-        session_mock.post(
-            run_url + urllib.parse.quote("mkdir {}".format(directory_created), safe=""), text="Permission denied"
-        )
+        session_mock.post(run_url + urllib.parse.quote(f"mkdir {directory_created}", safe=""), text="Permission denied")
 
         return_value = class_being_tested.process_exposed_existing_privileged_container(
             url + "run/kube-hunter-privileged/kube-hunter-privileged-deployment-86dc79f945-sjjps/ubuntu",
@@ -644,7 +642,7 @@ def test_process_exposed_existing_privileged_container_failure_when_mount():
 
         session_mock.post(run_url + urllib.parse.quote("cat /proc/cmdline", safe=""), text=cat_proc_cmdline)
         session_mock.post(run_url + urllib.parse.quote("findfs LABEL=Mock", safe=""), text="/dev/mock_fs")
-        session_mock.post(run_url + urllib.parse.quote("mkdir {}".format(directory_created), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"mkdir {directory_created}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("mount {} {}".format("/dev/mock_fs", directory_created), safe=""),
             text="Permission denied",
@@ -671,12 +669,12 @@ def test_process_exposed_existing_privileged_container_failure_when_cat_hostname
 
         session_mock.post(run_url + urllib.parse.quote("cat /proc/cmdline", safe=""), text=cat_proc_cmdline)
         session_mock.post(run_url + urllib.parse.quote("findfs LABEL=Mock", safe=""), text="/dev/mock_fs")
-        session_mock.post(run_url + urllib.parse.quote("mkdir {}".format(directory_created), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"mkdir {directory_created}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("mount {} {}".format("/dev/mock_fs", directory_created), safe=""), text=""
         )
         session_mock.post(
-            run_url + urllib.parse.quote("cat {}/etc/hostname".format(directory_created), safe=""),
+            run_url + urllib.parse.quote(f"cat {directory_created}/etc/hostname", safe=""),
             text="Permission denied",
         )
 
@@ -699,18 +697,18 @@ def test_maliciousintentviasecurekubeletport_success():
         run_url = url + "run/kube-hunter-privileged/kube-hunter-privileged-deployment-86dc79f945-sjjps/ubuntu?cmd="
         directory_created = "/kube-hunter-mock_" + str(uuid.uuid1())
         file_name = "kube-hunter-mock" + str(uuid.uuid1())
-        file_name_with_path = "{}/etc/cron.daily/{}".format(directory_created, file_name)
+        file_name_with_path = f"{directory_created}/etc/cron.daily/{file_name}"
 
         session_mock.post(run_url + urllib.parse.quote("cat /proc/cmdline", safe=""), text=cat_proc_cmdline)
         session_mock.post(run_url + urllib.parse.quote("findfs LABEL=Mock", safe=""), text="/dev/mock_fs")
-        session_mock.post(run_url + urllib.parse.quote("mkdir {}".format(directory_created), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"mkdir {directory_created}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("mount {} {}".format("/dev/mock_fs", directory_created), safe=""), text=""
         )
         session_mock.post(
-            run_url + urllib.parse.quote("cat {}/etc/hostname".format(directory_created), safe=""), text="mockhostname"
+            run_url + urllib.parse.quote(f"cat {directory_created}/etc/hostname", safe=""), text="mockhostname"
         )
-        session_mock.post(run_url + urllib.parse.quote("touch {}".format(file_name_with_path), safe=""), text="")
+        session_mock.post(run_url + urllib.parse.quote(f"touch {file_name_with_path}", safe=""), text="")
         session_mock.post(
             run_url + urllib.parse.quote("chmod {} {}".format("755", file_name_with_path), safe=""), text=""
         )