github/kube-hunter
1
0
Fork 0
mirror of https://github.com/aquasecurity/kube-hunter.git synced 2026-02-14 18:09:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bug Fix: False Negative On AKS Hunting (#420)

Browse Source 
* removed false negative in AzureSpnHunter when /run is disabled

* changed to use direct imported class

* fixed multiple bugs in azure spn hunting, and improved efficency

* fixed bug in cloud identification. TODO: remove the outsourcing for cloud provider

* removed unused config variable

* fixed tests to use already parsed pods as the given previous event has changed
This commit is contained in:
danielsagi
2021-01-07 19:46:00 +02:00
committed by GitHub
parent d208b43532
commit cdbc3dc12b
2 changed files with 83 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
kube_hunter/modules/hunting/aks.py
View File

@@ -1,9 +1,10 @@
import os
import json
import logging
import requests
from kube_hunter.conf import get_config
from kube_hunter.modules.hunting.kubelet import ExposedRunHandler
from kube_hunter.modules.hunting.kubelet import ExposedPodsHandler, SecureKubeletPortHunter
from kube_hunter.core.events import handler
from kube_hunter.core.events.types import Event, Vulnerability
from kube_hunter.core.types import Hunter, ActiveHunter, IdentityTheft, Azure
@@ -14,7 +15,7 @@ logger = logging.getLogger(__name__)
class AzureSpnExposure(Vulnerability, Event):
"""The SPN is exposed, potentially allowing an attacker to gain access to the Azure subscription"""
def __init__(self, container):
def __init__(self, container, evidence=""):
Vulnerability.__init__(
self,
Azure,
@@ -23,9 +24,10 @@ class AzureSpnExposure(Vulnerability, Event):
vid="KHV004",
)
self.container = container
self.evidence = evidence
@handler.subscribe(ExposedRunHandler, predicate=lambda x: x.cloud == "Azure")
@handler.subscribe(ExposedPodsHandler, predicate=lambda x: x.cloud_type == "Azure")
class AzureSpnHunter(Hunter):
"""AKS Hunting
Hunting Azure cluster deployments using specific known configurations
@@ -37,35 +39,33 @@ class AzureSpnHunter(Hunter):
# getting a container that has access to the azure.json file
def get_key_container(self):
config = get_config()
endpoint = f"{self.base_url}/pods"
logger.debug("Trying to find container with access to azure.json file")
try:
r = requests.get(endpoint, verify=False, timeout=config.network_timeout)
except requests.Timeout:
logger.debug("failed getting pod info")
else:
pods_data = r.json().get("items", [])
suspicious_volume_names = []
for pod_data in pods_data:
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"],
"namespace": pod_data["metadata"]["namespace"],
}
# pods are saved in the previous event object
pods_data = self.event.pods
suspicious_volume_names = []
for pod_data in pods_data:
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"],
"namespace": pod_data["metadata"]["namespace"],
"mount": mount,
}
def execute(self):
container = self.get_key_container()
if container:
self.publish_event(AzureSpnExposure(container=container))
evidence = f"pod: {container['pod']}, namespace: {container['namespace']}"
self.publish_event(AzureSpnExposure(container=container, evidence=evidence))
@handler.subscribe(AzureSpnExposure)
@@ -78,14 +78,42 @@ class ProveAzureSpnExposure(ActiveHunter):
self.event = event
self.base_url = f"https://{self.event.host}:{self.event.port}"
def test_run_capability(self):
"""
Uses SecureKubeletPortHunter to test the /run handler
TODO: when multiple event subscription is implemented, use this here to make sure /run is accessible
"""
debug_handlers = SecureKubeletPortHunter.DebugHandlers(path=self.base_url, session=self.event.session, pod=None)
return debug_handlers.test_run_container()
def run(self, command, container):
config = get_config()
run_url = "/".join(self.base_url, "run", container["namespace"], container["pod"], container["name"])
return requests.post(run_url, verify=False, params={"cmd": command}, timeout=config.network_timeout)
run_url = f"{self.base_url}/run/{container['namespace']}/{container['pod']}/{container['name']}"
return self.event.session.post(run_url, verify=False, params={"cmd": command}, timeout=config.network_timeout)
def get_full_path_to_azure_file(self):
"""
Returns a full path to /etc/kubernetes/azure.json
Taking into consideration the difference folder of the mount inside the container.
TODO: implement the edge case where the mount is to parent /etc folder.
"""
azure_file_path = self.event.container["mount"]["mountPath"]
# taking care of cases where a subPath is added to map the specific file
if not azure_file_path.endswith("azure.json"):
azure_file_path = os.path.join(azure_file_path, "azure.json")
return azure_file_path
def execute(self):
if not self.test_run_capability():
logger.debug("Not proving AzureSpnExposure because /run debug handler is disabled")
return
try:
subscription = self.run("cat /etc/kubernetes/azure.json", container=self.event.container).json()
azure_file_path = self.get_full_path_to_azure_file()
logger.debug(f"trying to access the azure.json at the resolved path: {azure_file_path}")
subscription = self.run(f"cat {azure_file_path}", container=self.event.container).json()
except requests.Timeout:
logger.debug("failed to run command in container", exc_info=True)
except json.decoder.JSONDecodeError:

57
tests/hunting/test_aks.py
View File

@@ -3,54 +3,47 @@ import requests_mock
from kube_hunter.conf import Config, set_config
import json
set_config(Config())
from kube_hunter.modules.hunting.kubelet import ExposedRunHandler
from kube_hunter.modules.hunting.kubelet import ExposedPodsHandler
from kube_hunter.modules.hunting.aks import AzureSpnHunter
def test_AzureSpnHunter():
e = ExposedRunHandler()
e.host = "mockKubernetes"
e.port = 443
e.protocol = "https"
e = ExposedPodsHandler(pods=[])
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
e.pods = json.loads(pod_template.format(p))["items"]
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)
e.pods = json.loads(pod_template.format(p))["items"]
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
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"}]}} ]}'
e.pods = json.loads(pod_no_volume_mounts)["items"]
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
pod_no_volumes = '{"items":[ {"apiVersion":"v1","kind":"Pod","metadata":{"name":"etc","namespace":"default"},"spec":{"containers":[{"command":["sleep","99999"],"image":"ubuntu","name":"test"}]}} ]}'
e.pods = json.loads(pod_no_volumes)["items"]
h = AzureSpnHunter(e)
c = h.get_key_container()
assert c == None
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"}]}} ]}'
e.pods = json.loads(pod_other_volume)["items"]
h = AzureSpnHunter(e)
c = h.get_key_container()
assert c == None