-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathPin to HUD.py
193 lines (150 loc) · 7.51 KB
/
Pin to HUD.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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
"""
Demonstrate updating the Heads-Up Display from a Playbook using a variety of indicator types, styles, and sizes.
"""
import phantom.rules as phantom
import json
from datetime import datetime, timedelta
##############################
# Start - Global Code Block
def is_ioc(value):
import phantom.utils as phutils
ioc_funcs = [phutils.is_ip, phutils.is_url, phutils.is_email, phutils.is_hash]
for f in ioc_funcs:
if f(value):
return True, f.__name__.split('_')[1]
return False, None
def pin_name_mangle(pin_name, container):
return pin_name + '__{0}'.format(container['id'])
# End - Global Code block
##############################
def on_start(container):
phantom.debug('on_start() called')
pin_1(container=container)
pin_2(container=container)
pin_3(container=container)
pin_4(container=container)
return
def pin_4(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
phantom.debug('pin_4() called')
artifacts = phantom.collect(container=container, datapath='artifacts:*', scope='all')
artifacts = sorted(artifacts, key = lambda x: x['update_time'], reverse=True)
ioc_count = 0
most_recent_ioc = None
ioc_types = set()
for artifact in artifacts:
for key, value in artifact['cef'].iteritems():
value = str(value)
ret, ioc_type = is_ioc(value)
if ret:
if most_recent_ioc is None:
most_recent_ioc = value
ioc_count += 1
ioc_types.add(ioc_type)
pin4_name = pin_name_mangle("pin_4", container)
pin5_name = pin_name_mangle("pin_5", container)
pin6_name = pin_name_mangle("pin_6", container)
pin_id_ioc_cnt = phantom.get_data(pin4_name)
pin_id_ioc_rct = phantom.get_data(pin5_name)
pin_id_ioc_type = phantom.get_data(pin6_name)
if not pin_id_ioc_cnt:
ret_val, message, pin_id_ioc_cnt = phantom.pin(container=container, message="IOC Count", data=str(ioc_count), pin_type="card_medium", pin_style="white")
else:
ret_val, message = phantom.update_pin(pin_id_ioc_cnt, message="IOC Count", data=str(ioc_count), pin_type="card_medium", pin_style="red")
if ret_val:
phantom.save_data(pin_id_ioc_cnt, pin4_name)
if ioc_count:
if not pin_id_ioc_rct:
ret_val, message, pin_id_ioc_rct = phantom.pin(container=container, message="Most Recent IOC", data=most_recent_ioc, pin_type="card_medium", pin_style="purple")
else:
ret_val, message = phantom.update_pin(pin_id_ioc_rct, message="Most Recent IOC", data=most_recent_ioc, pin_type="card_medium", pin_style="purple")
if ret_val:
phantom.save_data(pin_id_ioc_rct, pin5_name)
if not pin_id_ioc_type:
ret_val, message, pin_id_ioc_type = phantom.pin(container=container, message="IOC Types", data=", ".join(ioc_types))
else:
ret_val, message = phantom.update_pin(pin_id_ioc_type, message="IOC Types", data=", ".join(ioc_types))
if ret_val:
phantom.save_data(pin_id_ioc_type, pin6_name)
return
def pin_2(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
import random
phantom.debug('pin_2() called')
# collect data for 'pin_to_hud_6' call
dest_username = filter(lambda x: x[0], phantom.collect2(container=container, datapath=['artifact:*.cef.destinationUserName']))
sorc_username = filter(lambda x: x[0], phantom.collect2(container=container, datapath=['artifact:*.cef.sourceUserName']))
styles = set(["white", "red", "purple"])
pin_name = pin_name_mangle("pin_2", container)
pin_id = phantom.get_data(pin_name)
if not pin_id:
ret_val, message, pin_id = phantom.pin(container=container, message="Affected Users", data=str(len(dest_username) + len(sorc_username)), pin_type="card_medium", pin_style="purple")
phantom.debug("new pin_2")
else:
# Delete and remake this one, for the sake of demonstration
ret_val, message = phantom.delete_pin(pin_id)
ret_val, message, pin_id = phantom.pin(container=container, message="Affected Users", data=str(len(dest_username) + len(sorc_username)), pin_type="card_medium", pin_style=random.sample(styles, 1)[0])
if ret_val:
phantom.save_data(pin_id, pin_name)
# set container properties for:
update_data = {
}
phantom.update(container, update_data)
return
def pin_3(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
import random
phantom.debug('pin_3() called')
# collect data for 'pin_to_hud_6' call
dest_domain = filter(lambda x: x[0], phantom.collect2(container=container, datapath=['artifact:*.cef.destinationDnsDomain']))
pin_name = pin_name_mangle("pin_3", container)
try:
most_rcnt_domain = dest_domain[0][0]
except:
pass
else:
pin_id = phantom.get_data(pin_name)
if not pin_id:
ret_val, message, pin_id = phantom.pin(container=container, message="Most Recent Domain", data=most_rcnt_domain, pin_type="card_medium", pin_style="red")
phantom.debug("new pin_3")
else:
ret_val, message = phantom.update_pin(pin_id, message="Most Recent Domain", data=most_rcnt_domain, pin_type="card_medium", pin_style="red")
if ret_val:
phantom.save_data(pin_id, pin_name)
# set container properties for:
update_data = {
}
phantom.update(container, update_data)
return
def pin_1(action=None, success=None, container=None, results=None, handle=None, filtered_artifacts=None, filtered_results=None):
import random
phantom.debug('pin_1() called')
# collect data for 'pin_to_hud_6' call
dest_ip_artifacts = filter(lambda x: x[0], phantom.collect2(container=container, datapath=['artifact:*.cef.destinationAddress']))
sorc_ip_artifacts = filter(lambda x: x[0], phantom.collect2(container=container, datapath=['artifact:*.cef.sourceAddress']))
styles = set(["white", "red", "purple"])
pin_name = pin_name_mangle("pin_1", container)
pin_id = phantom.get_data(pin_name)
if not pin_id:
ret_val, message, pin_id = phantom.pin(container=container, message="Affected IPs", data=str(len(dest_ip_artifacts) + len(sorc_ip_artifacts)), pin_type="card_medium", pin_style="white")
phantom.debug("new pin_1")
else:
style = random.sample(styles, 1)[0]
phantom.debug(style)
ret_val, message = phantom.update_pin(pin_id, message="Affected IPs", data=str(len(dest_ip_artifacts) + len(sorc_ip_artifacts)), pin_style=style)
if ret_val:
phantom.save_data(pin_id, pin_name)
# set container properties for:
update_data = {
}
phantom.update(container, update_data)
return
def on_finish(container, summary):
phantom.debug('on_finish() called')
# This function is called after all actions are completed.
# summary of all the action and/or all detals of actions
# can be collected here.
# summary_json = phantom.get_summary()
# if 'result' in summary_json:
# for action_result in summary_json['result']:
# if 'action_run_id' in action_result:
# action_results = phantom.get_action_results(action_run_id=action_result['action_run_id'], result_data=False, flatten=False)
# phantom.debug(action_results)
return