import logging from packaging import version from kube_hunter.conf import get_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 = f"kubectl version: {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 = f"kubectl version: {self.binary_version}" class CveUtils: @staticmethod def get_base_release(full_ver): # if LegacyVersion, converting manually to a base version if isinstance(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 not isinstance(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 isinstance(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): config = get_config() 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): config = get_config() 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))