forked from MISP/misp-modules
/
stiximport.py
executable file
·256 lines (199 loc) · 6.2 KB
/
stiximport.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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
import json
from stix.core import STIXPackage
import re
import base64
import hashlib
import tempfile
misperrors = {'error': 'Error'}
userConfig = {}
inputSource = ['file']
moduleinfo = {'version': '0.1', 'author': 'Hannah Ward',
'description': 'Import some stix stuff',
'module-type': ['import']}
moduleconfig = ["max_size"]
def handler(q=False):
# Just in case we have no data
if q is False:
return False
# The return value
r = {'results': []}
# Load up that JSON
q = json.loads(q)
# It's b64 encoded, so decode that stuff
package = str(base64.b64decode(q.get("data", None)), 'utf-8')
# If something really weird happened
if not package:
return json.dumps({"success": 0})
# Get the maxsize from the config
# Default to 10MB
# (I believe the max_size arg is given in bytes)
# Check if we were given a configuration
memsize = q.get("config", None)
# If we were, find out if there's a memsize field
if memsize:
memsize = memsize.get("max_size", 10 * 1024)
else:
memsize = 10 * 1024
# Load up the package into STIX
package = loadPackage(package, memsize)
# Build all the observables
if package.observables:
for obs in package.observables:
r["results"].append(buildObservable(obs))
# And now the threat actors
if package.threat_actors:
for ta in package.threat_actors:
r["results"].append(buildActor(ta))
# Aaaand the indicators
if package.indicators:
for ind in package.indicators:
r["results"] += buildIndicator(ind)
# Are you seeing a pattern?
if package.exploit_targets:
for et in package.exploit_targets:
r["results"].append(buildExploitTarget(et))
# LOADING STUFF
if package.campaigns:
for cpn in package.campaigns:
r["results"].append(buildCampaign(cpn))
# Clean up results
# Don't send on anything that didn't have a value
r["results"] = [x for x in r["results"] if isinstance(x, dict) and len(x["values"]) != 0]
return r
# Quick and dirty regex for IP addresses
ipre = re.compile("([0-9]{1,3}.){3}[0-9]{1,3}")
def buildCampaign(cpn):
"""
Extract a campaign name
"""
return {"values": [cpn.title], "types": ["campaign-name"]}
def buildExploitTarget(et):
"""
Extract CVEs from exploit targets
"""
r = {"values": [], "types": ["vulnerability"]}
if et.vulnerabilities:
for v in et.vulnerabilities:
if v.cve_id:
r["values"].append(v.cve_id)
return r
def identifyHash(hsh):
"""
What's that hash!?
"""
possible_hashes = []
hashes = [x for x in hashlib.algorithms_guaranteed]
for h in hashes:
if len(str(hsh)) == len(hashlib.new(h).hexdigest()):
possible_hashes.append(h)
possible_hashes.append("filename|{}".format(h))
return possible_hashes
def buildIndicator(ind):
"""
Extract hashes
and other fun things
like that
"""
r = []
# Try to get hashes. I hate stix
if ind.observables:
for i in ind.observables:
if i.observable_composition:
for j in i.observable_composition.observables:
r.append(buildObservable(j))
r.append(buildObservable(i))
return r
def buildActor(ta):
"""
Extract the name
and comment of a
threat actor
"""
r = {"values": [ta.title], "types": ["threat-actor"]}
return r
def buildObservable(o):
"""
Take a STIX observable
and extract the value
and category
"""
# Life is easier with json
if not isinstance(o, dict):
o = json.loads(o.to_json())
# Make a new record to store values in
r = {"values": []}
# Get the object properties. This contains all the
# fun stuff like values
if "observable_composition" in o:
# May as well be useless
return r
props = o["object"]["properties"]
# If it has an address_value field, it's gonna be an address
# Kinda obvious really
if "address_value" in props:
# We've got ourselves a nice little address
value = props["address_value"]
if isinstance(value, dict):
# Sometimes it's embedded in a dictionary
value = value["value"]
# Is it an IP?
if ipre.match(str(value)):
# Yes!
r["values"].append(value)
r["types"] = ["ip-src", "ip-dst"]
else:
# Probably a domain yo
r["values"].append(value)
r["types"] = ["domain", "hostname"]
if "hashes" in props:
for hsh in props["hashes"]:
r["values"].append(hsh["simple_hash_value"]["value"])
r["types"] = identifyHash(hsh["simple_hash_value"]["value"])
elif "xsi:type" in props:
# Cybox. Ew.
try:
type_ = props["xsi:type"]
val = props["value"]
if type_ == "LinkObjectType":
r["types"] = ["link"]
r["values"].append(val)
else:
print("Ignoring {}".format(type_))
except:
pass
return r
def loadPackage(data, memsize=1024):
# Write the stix package to a tmp file
temp = tempfile.SpooledTemporaryFile(max_size=int(memsize), mode="w+")
temp.write(data)
# Back to the beginning so we can read it again
temp.seek(0)
try:
# Try loading it into every format we know of
try:
package = STIXPackage().from_xml(temp)
except:
# We have to seek back again
temp.seek(0)
package = STIXPackage().from_json(temp)
except Exception:
print("Failed to load package")
raise ValueError("COULD NOT LOAD STIX PACKAGE!")
temp.close()
return package
def introspection():
modulesetup = {}
try:
userConfig
modulesetup['userConfig'] = userConfig
except NameError:
pass
try:
inputSource
modulesetup['inputSource'] = inputSource
except NameError:
pass
return modulesetup
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo