forked from splunk-soar-connectors/digitalshadows
-
Notifications
You must be signed in to change notification settings - Fork 2
/
digital_shadows_connector.py
191 lines (158 loc) · 7.91 KB
/
digital_shadows_connector.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
# File: digital_shadows_connector.py
#
# Licensed under Apache 2.0 (https://www.apache.org/licenses/LICENSE-2.0.txt)
#
import json
import phantom.app as phantom
from phantom.base_connector import BaseConnector
from digital_shadows_consts import DS_ACTION_NOT_SUPPORTED
from ds_databreach_connector import DSDataBreachConnector
from ds_incidents_connector import DSIncidentsConnector
from ds_intelligence_incidents_connector import DSIntelligenceIncidentsConnector
from ds_on_poll_connector import DSOnPollConnector
from ds_search_entities_connector import DSSearchEntitiesConnector
from ds_test_connectivity_connector import DSTestConnectivityConnector
class DigitalShadowsConnector(BaseConnector):
def __init__(self):
super(DigitalShadowsConnector, self).__init__()
def test_connectivity(self):
self.save_progress("Testing connectivity")
self.save_progress("test")
test_connectivity_connector = DSTestConnectivityConnector(self)
return test_connectivity_connector.test_connectivity()
def get_incident_by_id(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
incidents_connector = DSIncidentsConnector(self)
return incidents_connector.get_incident_by_id(param)
def get_incident_review_by_id(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
incidents_connector = DSIncidentsConnector(self)
return incidents_connector.get_incident_review_by_id(param)
def get_incident_list(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
incidents_connector = DSIncidentsConnector(self)
return incidents_connector.get_incident_list(param)
def post_incident_review(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
incidents_connector = DSIncidentsConnector(self)
return incidents_connector.post_incident_review(param)
def get_intelligence_incident_by_id(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
intelligence_incidents_connector = DSIntelligenceIncidentsConnector(self)
return intelligence_incidents_connector.get_intelligence_incident_by_id(param)
def get_intel_incident_ioc_by_id(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
intelligence_incidents_connector = DSIntelligenceIncidentsConnector(self)
return intelligence_incidents_connector.get_intel_incident_ioc_by_id(param)
def get_intelligence_incident(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
intelligence_incidents_connector = DSIntelligenceIncidentsConnector(self)
return intelligence_incidents_connector.get_intelligence_incident(param)
def get_data_breach(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.get_data_breach(param)
def get_data_breach_by_id(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.get_data_breach_by_id(param)
def get_data_breach_record(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.get_data_breach_record(param)
def get_data_breach_record_by_id(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.get_data_breach_record_by_id(param)
def get_data_breach_record_by_username(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.get_data_breach_record_by_username(param)
def get_data_breach_record_reviews(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.get_data_breach_record_reviews(param)
def post_breach_record_review(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
databreach_connector = DSDataBreachConnector(self)
return databreach_connector.post_breach_record_review(param)
def search_entities(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
search_entities_connector = DSSearchEntitiesConnector(self)
return search_entities_connector.search_entities(param)
def on_poll(self, param):
self.save_progress("Entered the function")
self.save_progress("processing")
on_poll_connector = DSOnPollConnector(self)
return on_poll_connector.on_poll(param)
def handle_action(self, param):
# get action which is needed to run
action_id = self.get_action_identifier()
if param:
self.save_progress("Ingesting handle action in: {}".format(param))
if action_id == 'test_connectivity':
return self.test_connectivity()
elif action_id == 'get_incident_by_id':
return self.get_incident_by_id(param)
elif action_id == 'get_incident_review_by_id':
return self.get_incident_review_by_id(param)
elif action_id == 'get_incident_list':
return self.get_incident_list(param)
elif action_id == 'post_incident_review':
return self.post_incident_review(param)
elif action_id == 'get_intelligence_incident_by_id':
return self.get_intelligence_incident_by_id(param)
elif action_id == 'get_intel_incident_ioc_by_id':
return self.get_intel_incident_ioc_by_id(param)
elif action_id == 'get_intelligence_incident':
return self.get_intelligence_incident(param)
elif action_id == 'get_data_breach':
return self.get_data_breach(param)
elif action_id == 'get_data_breach_by_id':
return self.get_data_breach_by_id(param)
elif action_id == 'get_data_breach_record':
return self.get_data_breach_record(param)
elif action_id == 'get_data_breach_record_by_id':
return self.get_data_breach_record_by_id(param)
elif action_id == 'get_data_breach_record_by_username':
return self.get_data_breach_record_by_username(param)
elif action_id == 'get_data_breach_record_reviews':
return self.get_data_breach_record_reviews(param)
elif action_id == 'post_breach_record_review':
return self.post_breach_record_review(param)
elif action_id == 'search_entities':
return self.search_entities(param)
elif action_id == 'on_poll':
return self.on_poll(param)
else:
self.save_progress(DS_ACTION_NOT_SUPPORTED.format(action_id))
return self.set_status(phantom.APP_ERROR, DS_ACTION_NOT_SUPPORTED.format(action_id))
if __name__ == '__main__':
import sys
if len(sys.argv) < 2:
print("No test json specified as input")
sys.exit(0)
with open(sys.argv[1]) as f:
in_json = f.read()
in_json = json.loads(in_json)
print(json.dumps(in_json, indent=4))
connector = DigitalShadowsConnector()
connector.print_progress_message = True
ret_val = connector._handle_action(json.dumps(in_json), None)
print(json.dumps(json.loads(ret_val), indent=4))
sys.exit(0)