kube-hunter/kube_hunter/modules/hunting/cves.py

import logging
from packaging import version

from kube_hunter.conf import config
from kube_hunter.core.events import handler
from kube_hunter.core.events.types import Vulnerability, Event, K8sVersionDisclosure
from kube_hunter.core.types import (
    Hunter,
    KubernetesCluster,
    RemoteCodeExec,
    PrivilegeEscalation,
    DenialOfService,
    KubectlClient,
)
from kube_hunter.modules.discovery.kubectl import KubectlClientEvent

logger = logging.getLogger(__name__)


class ServerApiVersionEndPointAccessPE(Vulnerability, Event):
    """Node is vulnerable to critical CVE-2018-1002105"""

    def __init__(self, evidence):
        Vulnerability.__init__(
            self,
            KubernetesCluster,
            name="Critical Privilege Escalation CVE",
            category=PrivilegeEscalation,
            vid="KHV022",
        )
        self.evidence = evidence


class ServerApiVersionEndPointAccessDos(Vulnerability, Event):
    """Node not patched for CVE-2019-1002100. Depending on your RBAC settings,
     a crafted json-patch could cause a Denial of Service."""

    def __init__(self, evidence):
        Vulnerability.__init__(
            self,
            KubernetesCluster,
            name="Denial of Service to Kubernetes API Server",
            category=DenialOfService,
            vid="KHV023",
        )
        self.evidence = evidence


class PingFloodHttp2Implementation(Vulnerability, Event):
    """Node not patched for CVE-2019-9512. an attacker could cause a
    Denial of Service by sending specially crafted HTTP requests."""

    def __init__(self, evidence):
        Vulnerability.__init__(
            self, KubernetesCluster, name="Possible Ping Flood Attack", category=DenialOfService, vid="KHV024",
        )
        self.evidence = evidence


class ResetFloodHttp2Implementation(Vulnerability, Event):
    """Node not patched for CVE-2019-9514. an attacker could cause a
    Denial of Service by sending specially crafted HTTP requests."""

    def __init__(self, evidence):
        Vulnerability.__init__(
            self, KubernetesCluster, name="Possible Reset Flood Attack", category=DenialOfService, vid="KHV025",
        )
        self.evidence = evidence


class ServerApiClusterScopedResourcesAccess(Vulnerability, Event):
    """Api Server not patched for CVE-2019-11247.
    API server allows access to custom resources via wrong scope"""

    def __init__(self, evidence):
        Vulnerability.__init__(
            self,
            KubernetesCluster,
            name="Arbitrary Access To Cluster Scoped Resources",
            category=PrivilegeEscalation,
            vid="KHV026",
        )
        self.evidence = evidence


class IncompleteFixToKubectlCpVulnerability(Vulnerability, Event):
    """The kubectl client is vulnerable to CVE-2019-11246,
    an attacker could potentially execute arbitrary code on the client's machine"""

    def __init__(self, binary_version):
        Vulnerability.__init__(
            self, KubectlClient, "Kubectl Vulnerable To CVE-2019-11246", category=RemoteCodeExec, vid="KHV027",
        )
        self.binary_version = binary_version
        self.evidence = "kubectl version: {}".format(self.binary_version)


class KubectlCpVulnerability(Vulnerability, Event):
    """The kubectl client is vulnerable to CVE-2019-1002101,
    an attacker could potentially execute arbitrary code on the client's machine"""

    def __init__(self, binary_version):
        Vulnerability.__init__(
            self, KubectlClient, "Kubectl Vulnerable To CVE-2019-1002101", category=RemoteCodeExec, vid="KHV028",
        )
        self.binary_version = binary_version
        self.evidence = "kubectl version: {}".format(self.binary_version)


class CveUtils:
    @staticmethod
    def get_base_release(full_ver):
        # if LegacyVersion, converting manually to a base version
        if type(full_ver) == version.LegacyVersion:
            return version.parse(".".join(full_ver._version.split(".")[:2]))
        return version.parse(".".join(map(str, full_ver._version.release[:2])))

    @staticmethod
    def to_legacy(full_ver):
        # converting version to version.LegacyVersion
        return version.LegacyVersion(".".join(map(str, full_ver._version.release)))

    @staticmethod
    def to_raw_version(v):
        if type(v) != version.LegacyVersion:
            return ".".join(map(str, v._version.release))
        return v._version

    @staticmethod
    def version_compare(v1, v2):
        """Function compares two versions, handling differences with conversion to LegacyVersion"""
        # getting raw version, while striping 'v' char at the start. if exists.
        # removing this char lets us safely compare the two version.
        v1_raw = CveUtils.to_raw_version(v1).strip("v")
        v2_raw = CveUtils.to_raw_version(v2).strip("v")
        new_v1 = version.LegacyVersion(v1_raw)
        new_v2 = version.LegacyVersion(v2_raw)

        return CveUtils.basic_compare(new_v1, new_v2)

    @staticmethod
    def basic_compare(v1, v2):
        return (v1 > v2) - (v1 < v2)

    @staticmethod
    def is_downstream_version(version):
        return any(c in version for c in "+-~")

    @staticmethod
    def is_vulnerable(fix_versions, check_version, ignore_downstream=False):
        """Function determines if a version is vulnerable,
        by comparing to given fix versions by base release"""
        if ignore_downstream and CveUtils.is_downstream_version(check_version):
            return False

        vulnerable = False
        check_v = version.parse(check_version)
        base_check_v = CveUtils.get_base_release(check_v)

        # default to classic compare, unless the check_version is legacy.
        version_compare_func = CveUtils.basic_compare
        if type(check_v) == version.LegacyVersion:
            version_compare_func = CveUtils.version_compare

        if check_version not in fix_versions:
            # comparing ease base release for a fix
            for fix_v in fix_versions:
                fix_v = version.parse(fix_v)
                base_fix_v = CveUtils.get_base_release(fix_v)

                # if the check version and the current fix has the same base release
                if base_check_v == base_fix_v:
                    # when check_version is legacy, we use a custom compare func, to handle differences between versions.
                    if version_compare_func(check_v, fix_v) == -1:
                        # determine vulnerable if smaller and with same base version
                        vulnerable = True
                        break

        # if we did't find a fix in the fix releases, checking if the version is smaller that the first fix
        if not vulnerable and version_compare_func(check_v, version.parse(fix_versions[0])) == -1:
            vulnerable = True

        return vulnerable


@handler.subscribe_once(K8sVersionDisclosure)
class K8sClusterCveHunter(Hunter):
    """K8s CVE Hunter
    Checks if Node is running a Kubernetes version vulnerable to
    specific important CVEs
    """

    def __init__(self, event):
        self.event = event

    def execute(self):
        logger.debug(f"Checking known CVEs for k8s API version: {self.event.version}")
        cve_mapping = {
            ServerApiVersionEndPointAccessPE: ["1.10.11", "1.11.5", "1.12.3"],
            ServerApiVersionEndPointAccessDos: ["1.11.8", "1.12.6", "1.13.4"],
            ResetFloodHttp2Implementation: ["1.13.10", "1.14.6", "1.15.3"],
            PingFloodHttp2Implementation: ["1.13.10", "1.14.6", "1.15.3"],
            ServerApiClusterScopedResourcesAccess: ["1.13.9", "1.14.5", "1.15.2"],
        }
        for vulnerability, fix_versions in cve_mapping.items():
            if CveUtils.is_vulnerable(fix_versions, self.event.version, not config.include_patched_versions):
                self.publish_event(vulnerability(self.event.version))


@handler.subscribe(KubectlClientEvent)
class KubectlCVEHunter(Hunter):
    """Kubectl CVE Hunter
    Checks if the kubectl client is vulnerable to specific important CVEs
    """

    def __init__(self, event):
        self.event = event

    def execute(self):
        cve_mapping = {
            KubectlCpVulnerability: ["1.11.9", "1.12.7", "1.13.5", "1.14.0"],
            IncompleteFixToKubectlCpVulnerability: ["1.12.9", "1.13.6", "1.14.2"],
        }
        logger.debug(f"Checking known CVEs for kubectl version: {self.event.version}")
        for vulnerability, fix_versions in cve_mapping.items():
            if CveUtils.is_vulnerable(fix_versions, self.event.version, not config.include_patched_versions):
                self.publish_event(vulnerability(binary_version=self.event.version))