-
Notifications
You must be signed in to change notification settings - Fork 76
/
main.py
170 lines (125 loc) · 5.51 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
import json
import subprocess
import argparse
import sys
from art import text2art
import termcolor
from clint.textui import colored, puts, indent
import time
mitre_tactics = ["privilege_escalation", "discovery", "command_and_control", "credential_access", "persistence",
"collection", "defense_evasion", "execution", "reconnaissance", "lateral_movement", "initial_access"]
def kubectl_subproc(kubectl_command):
k_proc = subprocess.run(kubectl_command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout = k_proc.stdout
stderr = k_proc.stderr
return stdout, stderr
def kubectl_print(k_out, k_err, k_mode, k_lead):
with indent(2):
if k_out:
puts(colored.white('\n'))
if k_mode == 'active':
puts(colored.green('✔ command output: \n'))
else:
puts(colored.green('✔ found, printing output:\n'))
with indent(2):
puts(colored.green(k_out.decode()))
if k_lead:
puts(colored.cyan("Leading to technique id: %s" % k_lead))
elif k_err:
puts(colored.white('\n'))
puts(colored.red(k_err.decode()))
else:
puts(colored.white('\n'))
puts(colored.red('✘ none found\n'))
def get_mitre_tactics():
print_logo()
with indent(2):
puts(colored.white("Supported MITRE ATT&CK Tactics:"))
print("")
for mitre_tactic in mitre_tactics:
with indent(2):
puts(colored.red("+ %s" % mitre_tactic))
def run_kubectl(rk_technique, rk_scan_mode):
technique_command = rk_technique.get('command')
technique_id = rk_technique['id']
technique_leading_to = rk_technique['leading_to']
technique_mode = rk_technique['mode']
technique_args = rk_technique['args']
technique_arg_list = rk_technique.get('arg_list')
technique_multistep = rk_technique.get('multistep')
technique_steps = rk_technique.get('commands')
puts(colored.white('\n'))
with indent(4):
puts(colored.yellow("ID: "), newline=False), puts(colored.white("%s" % technique_id))
puts(colored.yellow("Technique: "), newline=False), puts(colored.white("%s" % rk_technique['name']))
puts(colored.yellow("Command: "), newline=False), puts(colored.white("%s" % technique_command))
if rk_scan_mode == 'all' or technique_mode == rk_scan_mode:
if not technique_args:
if not technique_multistep:
out, err = kubectl_subproc(technique_command)
kubectl_print(out, err, technique_mode, technique_leading_to)
else:
for cmd_step in technique_steps:
out, err = kubectl_subproc(cmd_step)
kubectl_print(out, err, technique_mode, technique_leading_to)
time.sleep(2)
else:
with indent(4):
puts(colored.white('\n'))
puts(colored.red("✘ The command requires specific parameters:"))
puts(colored.white('\n'))
for arg in technique_arg_list:
x = input(" %s:" % arg)
technique_command = technique_command.replace("$%s" % arg, x)
puts(colored.green("✔ command updated, running: ", technique_command))
out, err = kubectl_subproc(technique_command)
kubectl_print(out, err, technique_mode, technique_leading_to)
time.sleep(2)
else:
with indent(4):
puts(colored.red("✘ This command mode does not match your scan mode."))
time.sleep(2)
def print_logo():
text_art = text2art("RED KUBE")
print(termcolor.colored(text_art, 'red'))
puts(colored.white(" +++ WELCOME TO RED-KUBE +++\n\n"))
def cleanup():
out, err = kubectl_subproc("kubectl delete pods trivy awscli")
with indent(2):
puts(colored.green(out.decode()))
puts(colored.red(err.decode()))
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument('--mode', action='store', dest='mode', type=str,
help='scan mode (passive/active/all)', required=False, default='passive')
parser.add_argument('--tactic', action='store', dest='tactic', type=str,
help='specific tactic', required=False)
parser.add_argument('--show_tactics', action='store_true', help='show tactics')
parser.add_argument('--cleanup', action='store_true', required=False)
cmd_args = parser.parse_args()
scan_tactic = ''
scan_mode = 'passive'
if cmd_args.mode:
scan_mode = cmd_args.mode
if cmd_args.tactic:
scan_tactic = cmd_args.tactic
if cmd_args.show_tactics:
get_mitre_tactics()
sys.exit()
if cmd_args.cleanup:
cleanup()
sys.exit()
if scan_tactic in mitre_tactics:
print_logo()
scan_tactic = scan_tactic
with indent(4):
puts(colored.red("MITRE ATT&CK Tactic %s chosen " % scan_tactic))
with open('attacks/%s.json' % scan_tactic) as tactic_file:
tactic_data = json.load(tactic_file)
for technique in tactic_data:
run_kubectl(technique, scan_mode)
time.sleep(1)
else:
print_logo()
puts(colored.red("Please choose a tactic using --tactic TACTIC_NAME"))
sys.exit()