github/kube-hunter
1
0
Fork 0
mirror of https://github.com/aquasecurity/kube-hunter.git synced 2026-08-23 22:26:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

replaced traceroute discovery, and refactored the code a bit

Browse Source
This commit is contained in:
Daniel Sagi
2019-06-30 21:57:13 +03:00
parent b5bf168938
commit 5db3f057a8
1 changed files with 78 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/discovery/hosts.py
+78 -62
View File
@@ -10,7 +10,7 @@ import requests
from netaddr import IPNetwork
from __main__ import config
from netifaces import AF_INET, ifaddresses, interfaces
from netifaces import AF_INET, ifaddresses, interfaces, gateways
from ...core.events import handler
from ...core.events.types import Event, NewHostEvent, Vulnerability
@@ -28,7 +28,6 @@ class RunningAsPodEvent(Event):
location = "Local to Pod"
if 'HOSTNAME' in os.environ:
location += "(" + os.environ['HOSTNAME'] + ")"
return location
def get_service_account_file(self, file):
@@ -50,32 +49,71 @@ class HostScanEvent(Event):
self.active = active # flag to specify whether to get actual data from vulnerabilities
self.predefined_hosts = predefined_hosts
class HostDiscoveryHelpers:
@staticmethod
def get_cloud(host):
# for comparing prefixes
class InterfaceTypes(Enum):
LOCALHOST = "127"
class CloudTypes(Enum):
AZURE = "Azure"
class HostDiscoveryUtils:
""" Static class containes util functions for Host discovery processes """
@classmethod
def get_cloud(cls, ip=None):
""" Returns cloud for a given ip address, defaults to the external ip"""
if not host:
host = cls._get_external_ip()
try:
logging.debug("Checking whether the cluster is deployed on azure's cloud")
# azurespeed.com provide their API via HTTP only; the service can be queried with
# HTTPS, but doesn't show a proper certificate. Since no encryption is worse then
# any encryption, we go with the verify=false option for the time being. At least
# this prevents leaking internal IP addresses to passive eavesdropping.
# TODO: find a more secure service to detect cloud IPs
metadata = requests.get("https://www.azurespeed.com/api/region?ipOrUrl={ip}".format(ip=host), verify=False).text
if host:
logging.debug("Checking whether the cluster is deployed on azure's cloud")
# azurespeed.com provide their API via HTTP only; the service can be queried with
# HTTPS, but doesn't show a proper certificate. Since no encryption is worse then
# any encryption, we go with the verify=false option for the time being. At least
# this prevents leaking internal IP addresses to passive eavesdropping.
# TODO: find a more secure service to detect cloud IPs
metadata = requests.get("https://www.azurespeed.com/api/region?ipOrUrl={ip}".format(ip=host), verify=False).text
except requests.ConnectionError as e:
logging.info("- unable to check cloud: {0}".format(e))
return
if "cloud" in metadata:
return json.loads(metadata)["cloud"]
# generator, generating a subnet by given a cidr
@staticmethod
def get_default_gateway():
return gateways()['default'][AF_INET][0]
@staticmethod
def _get_external_ip():
external_ip = None
try:
logging.debug("HostDiscovery hunter attempting to get external IP address")
external_ip = requests.get("http://canhazip.com").text # getting external ip, to determine if cloud cluster
except requests.ConnectionError as e:
logging.debug("unable to determine external IP address: {0}".format(e))
return external_ip
# generator, generating ip addresses from a given cidr
@staticmethod
def generate_subnet(ip, sn="24"):
logging.debug("HostDiscoveryHelpers.generate_subnet {0}/{1}".format(ip, sn))
logging.debug("HostDiscoveryUtils.generate_subnet {0}/{1}".format(ip, sn))
subnet = IPNetwork('{ip}/{sn}'.format(ip=ip, sn=sn))
for ip in IPNetwork(subnet):
logging.debug("HostDiscoveryHelpers.generate_subnet yielding {0}".format(ip))
logging.debug("HostDiscoveryUtils.generate_subnet yielding {0}".format(ip))
yield ip
# generate ip addresses from all internal network interfaces
@staticmethod
def generate_interfaces_subnet(sn='24'):
for ifaceName in interfaces():
for ip in [i['addr'] for i in ifaddresses(ifaceName).setdefault(AF_INET, [])]:
if InterfaceTypes.LOCALHOST.value in ip.__str__():
continue
for ip in HostDiscoveryUtils.generate_subnet(ip, sn):
yield ip
@handler.subscribe(RunningAsPodEvent)
class FromPodHostDiscovery(Discovery):
@@ -86,23 +124,26 @@ class FromPodHostDiscovery(Discovery):
self.event = event
def execute(self):
scan_subnets = list()
# Scan any hosts that the user specified
if config.remote or config.cidr:
self.publish_event(HostScanEvent())
else:
# Discover cluster subnets, we'll scan all these hosts
if self.is_azure_pod():
# TODO: add more api discoveries
# If we have access to azure's api, we extract subnets from there
if self.is_azure_api():
subnets, cloud = self.azure_metadata_discovery()
else:
subnets, cloud = self.traceroute_discovery()
scan_subnets += subnets
for subnet in subnets:
subnets, cloud = self.pod_discovery()
scan_subnets += subnets
for subnet in scan_subnets:
logging.debug("From pod scanning subnet {0}/{1}".format(subnet[0], subnet[1]))
for ip in HostDiscoveryHelpers.generate_subnet(ip=subnet[0], sn=subnet[1]):
for ip in HostDiscoveryUtils.generate_subnet(ip=subnet[0], sn=subnet[1]):
self.publish_event(NewHostEvent(host=ip, cloud=cloud))
def is_azure_pod(self):
def is_azure_api(self):
try:
logging.debug("From pod attempting to access Azure Metadata API")
if requests.get("http://169.254.169.254/metadata/instance?api-version=2017-08-01", headers={"Metadata":"true"}, timeout=5).status_code == 200:
@@ -110,30 +151,25 @@ class FromPodHostDiscovery(Discovery):
except requests.exceptions.ConnectionError:
return False
# for pod scanning
def traceroute_discovery(self):
external_ip = requests.get("http://canhazip.com").text # getting external ip, to determine if cloud cluster
cloud = HostDiscoveryHelpers.get_cloud(external_ip)
logging.getLogger("scapy.runtime").setLevel(logging.ERROR) # disables scapy's warnings
from scapy.all import ICMP, IP, Ether, srp1
node_internal_ip = srp1(Ether() / IP(dst="google.com" , ttl=1) / ICMP(), verbose=0)[IP].src
return [ [node_internal_ip,"24"], ], external_ip
def pod_discovery(self):
# normal option when running as a pod is to scan it's own subnet
# The gateway connects us to the host, and we can discover the
# kubelet from there, other ip's are pods that are running
# next to us
return [(HostDiscoveryUtils.get_default_gateway(), "24")], HostDiscoveryUtils.get_cloud()
# quering azure's interface metadata api | works only from a pod
def azure_metadata_discovery(self):
logging.debug("From pod attempting to access azure's metadata")
machine_metadata = json.loads(requests.get("http://169.254.169.254/metadata/instance?api-version=2017-08-01", headers={"Metadata":"true"}).text)
address, subnet= "", ""
subnets = list()
for interface in machine_metadata["network"]["interface"]:
address, subnet = interface["ipv4"]["subnet"][0]["address"], interface["ipv4"]["subnet"][0]["prefix"]
logging.debug("From pod discovered subnet {0}/{1}".format(address, subnet if not config.quick else "24"))
subnets.append([address,subnet if not config.quick else "24"])
self.publish_event(AzureMetadataApi(cidr="{}/{}".format(address, subnet)))
return subnets, "Azure"
return subnets, CloudTypes.AZURE.value
@handler.subscribe(HostScanEvent)
class HostDiscovery(Discovery):
@@ -150,37 +186,17 @@ class HostDiscovery(Discovery):
except ValueError as e:
logging.error("unable to parse cidr: {0}".format(e))
return
cloud = HostDiscoveryHelpers.get_cloud(ip)
for ip in HostDiscoveryHelpers.generate_subnet(ip, sn=sn):
cloud = HostDiscoveryUtils.get_cloud(ip)
for ip in HostDiscoveryUtils.generate_subnet(ip, sn=sn):
self.publish_event(NewHostEvent(host=ip, cloud=cloud))
elif config.internal:
self.scan_interfaces()
elif len(config.remote) > 0:
for host in config.remote:
self.publish_event(NewHostEvent(host=host, cloud=HostDiscoveryHelpers.get_cloud(host)))
self.publish_event(NewHostEvent(host=host, cloud=HostDiscoveryUtils.get_cloud(host)))
# for normal scanning
def scan_interfaces(self):
try:
logging.debug("HostDiscovery hunter attempting to get external IP address")
external_ip = requests.get("http://canhazip.com").text # getting external ip, to determine if cloud cluster
except requests.ConnectionError as e:
logging.debug("unable to determine local IP address: {0}".format(e))
logging.info("~ default to 127.0.0.1")
external_ip = "127.0.0.1"
cloud = HostDiscoveryHelpers.get_cloud(external_ip)
for ip in self.generate_interfaces_subnet():
cloud = HostDiscoveryUtils.get_cloud()
for ip in HostDiscoveryUtils.generate_interfaces_subnet():
handler.publish_event(NewHostEvent(host=ip, cloud=cloud))
# generate all subnets from all internal network interfaces
def generate_interfaces_subnet(self, sn='24'):
for ifaceName in interfaces():
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):
yield ip
# for comparing prefixes
class InterfaceTypes(Enum):
LOCALHOST = "127"