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

changed module hirerarchy, fixed exit problem and changed to using cmd2

Browse Source
This commit is contained in:
Daniel Sagi
2021-06-17 14:40:18 +00:00
parent 6a4e6040f7
commit 2f6badd32f
10 changed files with 104 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kube_hunter/console/__init__.py
+2 -3
View File
@@ -1,5 +1,4 @@
from . import general
from . import manager
from . import env
from . import models
from . import discover
from . import show
from . import auth
kube_hunter/console/auth/__init__.py
+1
View File
@@ -0,0 +1 @@
from .auth import AuthSubConsole
kube_hunter/console/auth/auth.py
+37
View File
@@ -0,0 +1,37 @@
import cmd2
class AuthSubConsole(cmd2.Cmd):
def __init__(self, env):
self.env = env
self.prompt = environment.get_prompt(sub_console="env/auth")
def do_set(self, auth_index):
"""Manages the auth database, Usage:
auth + = <auth_token> // adds a new auth
auth 0 = <auth_token> // edits an already set auth index
"""
pass
def do_show(self, auth_index):
"""Show current collected auths"""
if auth_index:
try:
auth_index = int(auth_index)
except:
print("ERROR: Auth index should be a integer")
return
print(environment.current_auth.get_auth(auth_index).raw_token)
else:
print(environment.current_auth.get_table())
def do_exit(self, arg):
return True
def postcmd(self, stop, line):
self.prompt = self.prompt.get_prompt(sub_console="env/auth")
if stop:
return True
def emptyline(self):
pass
kube_hunter/console/models.py → kube_hunter/console/auth/types.py
+2 -53
View File
@@ -1,49 +1,3 @@
import json
import socket
import base64
from prettytable import ALL, PrettyTable
""" Models for console """
""" Basic Models """
class Container:
""" Basic model for Container objects """
name = ""
def __str__(self):
return self.name
class Pod:
""" Basic model for Pod objects """
ip_address = ""
name = ""
namespace = ""
containers = []
def __str__(self):
return f"{self.namespace}/{self.name}"
def incluster_update(self, pod_event):
"""
uses pod_event and other techniques to get full data on the incluster pod env data
"""
self.namespace = pod_event.namespace
# hostname will almost always will be the pod's name
self.name = socket.gethostname()
""" Cloud Models """
class Cloud:
name = None
def __repr__(self):
return self.name
class UnknownCloud(Cloud):
name = "Unknown Cloud"
""" Auth models"""
class Auth:
def parse_token(self, token):
@@ -92,7 +46,7 @@ class AuthStore:
def get_auth(self, index):
return self.auths[index]
def __repr__(self):
def get_table(self):
auth_table = PrettyTable(["index", "Name", "Selected"], hrules=ALL)
auth_table.align = "l"
auth_table.padding_width = 1
@@ -105,9 +59,4 @@ class AuthStore:
selected_mark = "*"
auth_table.add_row([i, auth.sub, selected_mark])
return auth_table
return auth_table
kube_hunter/console/discover.py
-24
View File
@@ -1,24 +0,0 @@
import cmd
import logging
from kube_hunter.console.env import environment
from kube_hunter.modules.discovery.hosts import RunningAsPodEvent
class DiscoverSubConsole(cmd.Cmd):
prompt = environment.get_prompt(sub_console="discover")
def do_local(self, arg):
# pod discovery
pass
def do_exit(self, arg):
return True
def postcmd(self, stop, line):
if stop:
return True
self.prompt = environment.get_prompt(sub_console="discover")
# binds EOF to exit the shell as well
do_EOF = do_exit
kube_hunter/console/env.py
-50
View File
@@ -1,50 +0,0 @@
from colorama import init, Fore, Style
from kube_hunter.console import models
# initializes colorma
init()
class ImmersedEnvironment:
"""
ImmersedEnvironment keeps track of the current console run state.
"""
auths = []
pods = []
is_inside_cloud = False
is_inside_container = False
is_inside_pod = False
current_cloud = models.UnknownCloud()
current_pod = models.Pod()
current_auth = models.AuthStore()
current_container = models.Container()
def get_prompt(self, sub_console=""):
"""
Parses current env state to picture a short description of where we are right now
General format is `(cloud) -> (run_unit) kube-hunter $`
"""
arrow = "->"
prompt_prefix = f" kube-hunter{' [' + sub_console + ']' if sub_console else ''} $ "
# add colores unly
cloud = f"({Fore.BLUE}{self.current_cloud}{Style.RESET_ALL})"
pod = f"({Fore.MAGENTA}{self.current_pod}{Style.RESET_ALL})"
container = f"(container: {Fore.CYAN}{self.current_container}{Style.RESET_ALL})"
container_in_pod = f"({Fore.MAGENTA}{self.current_pod}/{{}}{Style.RESET_ALL})"
env_description = cloud
if self.is_inside_pod:
if len(self.current_pod.containers):
env_description += f" {arrow} {container_in_pod.format(self.current_pod.containers[0])}"
else:
env_description += f" {arrow} {pod}"
elif self.is_inside_container:
env_description += f" {arrow} {container}"
return f"{env_description}{prompt_prefix}"
environment = ImmersedEnvironment()
kube_hunter/console/general/__init__.py
View File
kube_hunter/console/general/types.py
+32
View File
@@ -0,0 +1,32 @@
class Container:
""" Basic model for Container objects """
name = ""
def __str__(self):
return self.name
class Pod:
""" Basic model for Pod objects """
ip_address = ""
name = ""
namespace = ""
containers = []
def __str__(self):
return f"{self.namespace}/{self.name}"
def incluster_update(self, pod_event):
"""
uses pod_event and other techniques to get full data on the incluster pod env data
"""
self.namespace = pod_event.namespace
# hostname will almost always will be the pod's name
self.name = socket.gethostname()
class Cloud:
def __init__(self, name):
self.name = name
class UnknownCloud(Cloud):
def __init__(self):
super(UnknownCloud, self).__init__("Unknown Cloud")
kube_hunter/console/manager.py
+30 -39
View File
@@ -1,43 +1,27 @@
import cmd
import cmd2
import logging
from kube_hunter.console.env import environment
from kube_hunter.console.discover import DiscoverSubConsole
from kube_hunter.console.show import ShowSubConsole
from kube_hunter.console.env import EnvSubConsole, ImmersedEnvironment
from kube_hunter.modules.discovery.hosts import RunningAsPodEvent
class KubeHunterMainConsole(cmd2.Cmd):
def __init__(self, env=None):
kube_hunter_logo = r"""
__ __ __ __
/\ \ /\ \ /\ \ /\ \__
\ \ \/'\ __ _\ \ \____ __ \ \ \___ __ __ ___\ \ ,_\ __ _ __
\ \ , < /\ \/\ \ \ '__`\ /'__`\ ______\ \ _ `\/\ \/\ \/' _ `\ \ \/ /'__`/\`'__\
\ \ \\`\\ \ \_\ \ \ \L\ /\ __//\______\ \ \ \ \ \ \_\ /\ \/\ \ \ \_/\ __\ \ \/
\ \_\ \_\ \____/\ \_,__\ \____\/______/\ \_\ \_\ \____\ \_\ \_\ \__\ \____\ \_\
\/_/\/_/\/___/ \/___/ \/____/ \/_/\/_/\/___/ \/_/\/_/\/__/\/____/\/_/
"""
self.intro = f'{kube_hunter_logo}\n\nWelcome to kube-hunter Immeresed Console. Type help or ? to list commands.\n'
self.env = env
self.prompt = self.env.get_prompt()
def start_console():
KubeHunterMainConsole().cmdloop()
class KubeHunterMainConsole(cmd.Cmd):
kube_hunter_logo = r"""
__ __ __ __
/\ \ /\ \ /\ \ /\ \__
\ \ \/'\ __ _\ \ \____ __ \ \ \___ __ __ ___\ \ ,_\ __ _ __
\ \ , < /\ \/\ \ \ '__`\ /'__`\ ______\ \ _ `\/\ \/\ \/' _ `\ \ \/ /'__`/\`'__\
\ \ \\`\\ \ \_\ \ \ \L\ /\ __//\______\ \ \ \ \ \ \_\ /\ \/\ \ \ \_/\ __\ \ \/
\ \_\ \_\ \____/\ \_,__\ \____\/______/\ \_\ \_\ \____\ \_\ \_\ \__\ \____\ \_\
\/_/\/_/\/___/ \/___/ \/____/ \/_/\/_/\/___/ \/_/\/_/\/__/\/____/\/_/
"""
intro = f'{kube_hunter_logo}\n\nWelcome to kube-hunter Immeresed Console. Type help or ? to list commands.\n'
prompt = environment.get_prompt()
discover_cmd = DiscoverSubConsole()
show_cmd = ShowSubConsole()
def do_show(self, arg):
def do_env(self, arg):
'Show your environment data collected so far'
self.show_cmd.cmdloop()
def do_discover(self, arg):
'Depends on your environment, lets you discover nearby Services/Pods/Clusters'
self.discover_cmd.cmdloop()
def do_hunt(self, arg):
'Depends on your environment, lets you hunt nearby Services/Pods/Clusters'
pass
EnvSubConsole(self.env).cmdloop()
def do_interactive(self, arg):
import ipdb
@@ -47,13 +31,15 @@ class KubeHunterMainConsole(cmd.Cmd):
"""Try to determine you are based on local files and mounts"""
pod_event = RunningAsPodEvent()
if pod_event.auth_token:
environment.current_pod.incluster_update(pod_event)
environment.current_auth.new_auth(pod_event.auth_token)
environment.is_inside_pod = True
self.env.current_auth.new_auth(pod_event.auth_token)
self.env.current_pod.incluster_update(pod_event)
self.env.is_inside_pod = True
def postcmd(self, stop, line):
self.prompt = environment.get_prompt()
self.prompt = self.env.get_prompt()
if stop:
return True
def do_exit(self, arg):
'exists shell'
return True
@@ -64,3 +50,8 @@ class KubeHunterMainConsole(cmd.Cmd):
# binds EOF to exit the shell as well
do_EOF = do_exit
def start_console():
environment = ImmersedEnvironment()
a = KubeHunterMainConsole(environment)
a.cmdloop()
kube_hunter/console/show.py
-34
View File
@@ -1,34 +0,0 @@
import cmd
import logging
from kube_hunter.console.env import environment
class ShowSubConsole(cmd.Cmd):
prompt = environment.get_prompt(sub_console="show")
def do_auth(self, auth_index):
"""Displays the auth database, to show raw data of an auth entry pass it's index as a parameter."""
if auth_index:
try:
auth_index = int(auth_index)
except:
print("ERROR: Auth index should be a number")
return
print(environment.current_auth.get_auth(auth_index).raw_token)
else:
print(environment.current_auth.__repr__())
def do_exit(self, arg):
return True
def postcmd(self, stop, line):
self.prompt = environment.get_prompt(sub_console="show")
if stop:
return True
def emptyline(self):
pass
# binds EOF to exit the shell as well
do_EOF = do_exit