mirror of
https://github.com/aquasecurity/kube-hunter.git
synced 2026-04-27 04:47:05 +00:00
added missing env module, and fixed auth subconsole to use self.env
This commit is contained in:
1
.gitignore
vendored
1
.gitignore
vendored
@@ -9,6 +9,7 @@ venv/
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
env/
|
||||
!kube_hunter/console/env/
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
|
||||
50
kube_hunter/console/env.py
Normal file
50
kube_hunter/console/env.py
Normal file
@@ -0,0 +1,50 @@
|
||||
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()
|
||||
2
kube_hunter/console/env/__init__.py
vendored
Normal file
2
kube_hunter/console/env/__init__.py
vendored
Normal file
@@ -0,0 +1,2 @@
|
||||
from .env import EnvSubConsole
|
||||
from .types import ImmersedEnvironment
|
||||
31
kube_hunter/console/env/env.py
vendored
Normal file
31
kube_hunter/console/env/env.py
vendored
Normal file
@@ -0,0 +1,31 @@
|
||||
import cmd2
|
||||
|
||||
from kube_hunter.console.auth import AuthSubConsole
|
||||
|
||||
|
||||
class EnvSubConsole(cmd2.Cmd):
|
||||
"""EnvSubConsole
|
||||
In charge of managing and viewing the entire current environment state
|
||||
Includes: Auth database..
|
||||
"""
|
||||
def __init__(self, env):
|
||||
self.env = env
|
||||
self.prompt = self.env.get_prompt(sub_command="env")
|
||||
|
||||
def do_auth(self, arg):
|
||||
AuthSubConsole(self.env).cmdloop()
|
||||
|
||||
def postcmd(self, stop, line):
|
||||
self.prompt = self.env.get_prompt(sub_command="env")
|
||||
if stop:
|
||||
return True
|
||||
|
||||
def do_exit(self, arg):
|
||||
'exists shell'
|
||||
return True
|
||||
|
||||
def emptyline(self):
|
||||
pass
|
||||
|
||||
# binds EOF to exit the shell as well
|
||||
do_EOF = do_exit
|
||||
50
kube_hunter/console/env/types.py
vendored
Normal file
50
kube_hunter/console/env/types.py
vendored
Normal file
@@ -0,0 +1,50 @@
|
||||
from colorama import init, Fore, Style
|
||||
|
||||
from kube_hunter.console.general import types as GeneralTypes
|
||||
from kube_hunter.console.auth import types as AuthTypes
|
||||
|
||||
# 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 = GeneralTypes.UnknownCloud()
|
||||
current_pod = GeneralTypes.Pod()
|
||||
current_container = GeneralTypes.Container()
|
||||
|
||||
current_auth = AuthTypes.AuthStore()
|
||||
|
||||
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}"
|
||||
Reference in New Issue
Block a user