MISP / MISP-STIX-Converter / 139

pending completion

Build # 139

Build Type

Pull #29

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/MISP/MISP-STIX-Converter/commit/<a class=hub.com/MISP/MISP-STIX-Converter/commit/ce41b15c3f06f64e143cd0f20bcab4c5f2c2f951">ce41b15c3<a href="https://github.com/MISP/MISP-STIX-Converter/commit/ce41b15c3f06f64e143cd0f20bcab4c5f2c2f951">&quot;&gt;Fix for &amp;quot;AttributeError: can&amp;#39;t set attribute&amp;quot;

In the latest VM version availables (</a><a class="double-link" href="https://github.com/MISP/MISP-STIX-Converter/commit/<a class="double-link" href="https://github.com/MISP/MISP-STIX-Converter/commit/ccb3118940f9fcdf356126bed4370f3e7b5900a2">ccb311894</a>">ccb311894</a><a href="https://github.com/MISP/MISP-STIX-Converter/commit/ce41b15c3f06f64e143cd0f20bcab4c5f2c2f951">) STIX import does not work.
Maybe caused by a change in another library (from pymisp import mispevent).

The error reported
&quot;&quot;&quot;
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.6/dist-packages/misp_modules/__init__.py&quot;, line 197, in post
    response = yield tornado.gen.with_timeout(timeout, self.run_request(jsonpayload))
  File &quot;/usr/local/lib/python3.6/dist-packages/tornado/gen.py&quot;, line 1055, in run
    value = future.result()
  File &quot;/usr/local/lib/python3.6/dist-packages/tornado/concurrent.py&quot;, line 238, in result
    raise_exc_info(self._exc_info)
  File &quot;&lt;string&gt;&quot;, line 4, in raise_exc_info
  File &quot;/usr/lib/python3.6/concurrent/futures/thread.py&quot;, line 56, in run
    result = self.fn(*self.args, **self.kwargs)
  File &quot;/usr/local/lib/python3.6/dist-packages/misp_modules/__init__.py&quot;, line 185, in run_request
    response = mhandlers[x[&#39;module&#39;]].handler(q=jsonpayload)
  File &quot;modules/import_mod/stiximport.py&quot;, line 35, in handler
    pkg = stix.load_stix(package)
  File &quot;/usr/local/lib/python3.6/dist-packages/pymisp/tools/stix.py&quot;, line 18, in load_stix
    threat_level_id=threat_level_id, analysis=analysis)
  File &quot;/usr/local/lib/python3.6/dist-packages/misp_stix_converter/converters/buildMISPAttribute.py&quot;, line 273, in buildEvent
    event.attributes = uniqueAttributes
AttributeError: can't set attribute
"""

Pull Request Pull Request #29: Fix for "AttributeError: can't set attribute"

Run Details

116 of 287 branches covered (40.42%)

3 of 3 new or added lines in 1 file covered. (100.0%)

311 of 666 relevant lines covered (46.7%)

1.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.3

/misp_stix_converter/converters/buildMISPAttribute.py

#!/usr/bin/env python3
import re
import cybox
import logging
import hashlib
import ast
from pymisp import mispevent
from misp_stix_converter.converters.lint_roller import lintRoll
from stix.core import STIXPackage


# Cybox cybox don't we all love cybox children
from cybox.objects import email_message_object, file_object, address_object, socket_address_object
from cybox.objects import domain_name_object, hostname_object, uri_object
from cybox.objects import mutex_object, whois_object
from cybox.objects import as_object, http_session_object
from cybox.objects import pipe_object, network_packet_object, win_registry_key_object
from cybox.objects import x509_certificate_object, win_executable_file_object, win_process_object


# Just a little containment file for STIX -> MISP conversion
ipre = re.compile("([0-9]{1,3}.){3}[0-9]{1,3}")
log = logging.getLogger("__main__")


def uniq(lst):
    return_list = []
    for elem in lst:
        if elem not in return_list:
            return_list.append(elem)
    return return_list


def ast_eval(node):
    try:
        node = ast.literal_eval(node)
        if isinstance(node, list):
            node = uniq(node)
        return node
    except ValueError:
        return str(node)
    except SyntaxError:
        return str(node)


def parseRelated(obj, mispEvent, pkg):
    """Dedicated to parse related object
    Support for first level "importRelated.related_objects"
    """

    for i in obj.related_object:
        type_ = type(i.properties)
        # Here comes the sun (and of course the fun) for related objects! (DB)
        if type_ == address_object.Address:
            buildAddressAttribute(i.properties, mispEvent, pkg)

        elif type_ == domain_name_object.DomainName:
            buildDomainNameAttribute(i.properties, mispEvent, pkg)

        elif type_ == hostname_object.Hostname:
            buildHostnameAttribute(i.properties, mispEvent, pkg)

        elif type_ == socket_address_object.SocketAddress:
            buildAddressAttribute(i.properties, mispEvent, pkg, True)

        elif type_ == uri_object.URI:
            buildURIAttribute(i.properties, mispEvent, pkg)

        elif type_ == file_object.File:
            buildFileAttribute(i.properties, mispEvent, pkg)

        elif type_ == email_message_object.EmailMessage:
            buildEmailMessageAttribute(i.properties, mispEvent, pkg)

    return mispEvent


def parseAttachment(obj, mispEvent, pkg):
    """
    Limited support for email attachments, just consider this a work in progress
    """

    for i in obj:
        # Assuming any email attached file is a... file!
        buildFileAttribute(i.properties, mispEvent, pkg)

    return mispEvent


def buildFileAttribute(obj, mispEvent, pkg, importRelated=False):
    """
    All you can get by a File object in a single method
    TODO: all possible attributes are not yet parsed
    """

    if obj.file_name:
        mispEvent.add_attribute('filename', ast_eval(str(obj.file_name)), comment=pkg.title or None)

    # Added support to file_extension
    # No MISP object available for file extension
    # I propose to use pattern-in-file. Suggestions are welcome! (DB)
    if obj.file_extension:
        mispEvent.add_attribute('pattern-in-file', ast_eval(str(obj.file_extension)), comment=pkg.title or None)

    if obj.size_in_bytes:
        mispEvent.add_attribute('size-in-bytes', ast_eval(str(obj.size_in_bytes)), comment=pkg.title or None)

    if obj.md5:
        # We actually have to check the length
        # An actual report had supposed md5s of length 31. Silly.
        if len(str(obj.md5)) == 32:
            mispEvent.add_attribute('md5', ast_eval(str(obj.md5)), comment=pkg.title or None)

    if obj.sha1:
        if len(str(obj.sha1)) == 40:
            mispEvent.add_attribute('sha1', ast_eval(str(obj.sha1)), comment=pkg.title or None)

    if obj.sha256:
        if len(str(obj.sha256)) == 64:
            mispEvent.add_attribute('sha256', ast_eval(str(obj.sha256)), comment=pkg.title or None)

    # Added support for SHA512 (DB)
    if obj.sha512:
        if len(str(obj.sha512)) == 128:
            mispEvent.add_attribute('sha512', ast_eval(str(obj.sha512)), comment=pkg.title or None)

    if importRelated and pkg.object_.related_objects:
        parseRelated(pkg.object_.related_objects, mispEvent, pkg)

    return mispEvent


def buildAddressAttribute(obj, mispEvent, pkg, importRelated=False):

    if obj.is_source:
        mispEvent.add_attribute('ip-src', ast_eval(str(obj.address_value)), comment=pkg.title or None)

    elif obj.is_destination:
        mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value)), comment=pkg.title or None)
    else:
        # We don't know, first check if it's an IP range
        if hasattr(obj, "condition") and obj.condition:
            if obj.condition == "InclusiveBetween":
                # Ok, so it's a range. hm. Shall we add them seperately#comma#or together?
                mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value[0])))
                mispEvent.add_attribute('ip-dst', ast_eval(str(obj.add_attribute[1])))
            elif obj.condition == "Equals":
                mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value)), comment=pkg.title or None)
        else:
            # Don't have anything to go on
            mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value)), comment=pkg.title or None)

    if importRelated and pkg.object_.related_objects:
        parseRelated(pkg.object_.related_objects, mispEvent, pkg)

    return mispEvent


def buildEmailMessageAttribute(obj, mispEvent, pkg, importRelated=False):
    if obj.header:
        # We have a header, can check for to/from etc etc
        if obj.header.from_:
            mispEvent.add_attribute('email-src',
                                    ast_eval(str(obj.header.from_.address_value)),
                                    comment=pkg.title or None)
        if obj.header.to:
            for mail in obj.header.to:
                mispEvent.add_attribute('email-dst', ast_eval(mail.address_value), comment=pkg.title or None)
        if obj.header.subject:
            mispEvent.add_attribute('email-subject', ast_eval(str(obj.header.subject)), comment=pkg.title or None)

    if obj.attachments and pkg.object_.related_objects:
        parseAttachment(pkg.object_.related_objects, mispEvent, pkg)

    elif importRelated and pkg.object_.related_objects:
        parseRelated(pkg.object_.related_objects, mispEvent, pkg)

    return mispEvent


def buildDomainNameAttribute(obj, mispEvent, pkg, importRelated=False):
    mispEvent.add_attribute('domain', ast_eval(str(obj.value)), comment=pkg.title or None)

    if importRelated and pkg.object_.related_objects:
        parseRelated(pkg.object_.related_objects, mispEvent, pkg)

    return mispEvent


def buildHostnameAttribute(obj, mispEvent, pkg, importRelated=False):
    mispEvent.add_attribute('hostname', ast_eval(str(obj.hostname_value)), comment=pkg.title or None)

    if importRelated and pkg.object_.related_objects:
        parseRelated(pkg.object_.related_objects, mispEvent, pkg)

    return mispEvent


def buildURIAttribute(obj, mispEvent, pkg, importRelated=False):
    mispEvent.add_attribute('url', ast_eval(str(obj.value)), comment=pkg.title or None)

    if importRelated and pkg.object_.related_objects:
        parseRelated(pkg.object_.related_objects, mispEvent, pkg)

    return mispEvent


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 buildEvent(pkg, **kwargs):
    log.info("Building Event...")
    if not pkg.stix_header:
        title = "STIX Import"
    else:
        if not pkg.stix_header.title:
            title = "STIX Import"
        else:
            title = pkg.stix_header.title
    log.info("Using title %s", title)

    log.debug("Seting up MISPEvent...")
    event = mispevent.MISPEvent()
    event.distribution = kwargs.get("distribution", 0)
    event.threat_level_id = kwargs.get("threat_level_id", 3)
    event.analysis = kwargs.get("analysis", 0)
    event.info = title

    if hasattr(pkg, "description"):
        log.debug("Found description %s", pkg.description)
        event.add_attribute("comment", pkg.description)

    log.debug("Beginning to Lint_roll...")
    ids = []
    to_process = []
    for obj in lintRoll(pkg):
        if isinstance(obj, cybox.core.observable.Observable):
            if obj.id_ not in ids:
                ids.append(obj.id_)
                to_process.append(obj)

    log.debug("Processing %s object...", len(to_process))
    for obj in to_process:
        log.debug("Working on %s...", obj)
        # This will find literally every object ever.
        try:
            event = buildAttribute(obj, event)
        except Exception as ex:
            log.exception(ex)
    # Now make sure we only have unique items
    log.debug("Making sure we only have Unique attributes...")
    
    uniqueAttribValues = []

    for attrindex, attrib in enumerate(event.attributes):
        if attrib.value not in uniqueAttribValues:
            uniqueAttribValues.append(attrib.value)
        else:
            log.debug("Removed duplicated attribute in package: "+attrib.value)
            event.attributes.pop(attrindex)

    log.debug("Finished parsing attributes.")
    return event


def buildAttribute(pkg, mispEvent):
    try:
        # Check if the object is a cybox observable
        if isinstance(pkg, cybox.core.observable.Observable):
            if hasattr(pkg, "object_") and pkg.object_:

                obj = pkg.object_.properties

                # It's a proper object!
                type_ = type(obj)
                # Here comes the fun!
                if type_ == address_object.Address:
                    # Now script uses buildAddressAttribute (DB)
                    buildAddressAttribute(obj, mispEvent, pkg, True)

                elif type_ == domain_name_object.DomainName:
                    # Now script uses buildDomainNameAttribute (DB)
                    buildDomainNameAttribute(obj, mispEvent, pkg, True)

                elif type_ == hostname_object.Hostname:
                    # Now script uses buildHostnameAttribute
                    buildHostnameAttribute(obj, mispEvent, pkg, True)

                elif type_ == socket_address_object.SocketAddress:
                    if obj.ip_address:
                        buildAddressAttribute(obj.ip_address, mispEvent, pkg, True)
                    if obj.hostname:
                        buildHostnameAttribute(obj.hostname, mispEvent, pkg, True)

                elif type_ == uri_object.URI:
                    # Now script uses buildURIAttribute (DB)
                    buildURIAttribute(obj, mispEvent, pkg, True)

                elif type_ == file_object.File:
                    # Now script uses buildFileAttribute (DB)
                    buildFileAttribute(obj, mispEvent, pkg, True)

                elif type_ == email_message_object.EmailMessage:
                    # Now script uses buildEmailMessageAttribute (DB)
                    buildEmailMessageAttribute(obj, mispEvent, pkg, True)

                elif type_ == mutex_object.Mutex:
                    mispEvent.add_attribute('mutex', ast_eval(str(obj.name)), comment=pkg.title or None)
                elif type_ == whois_object.WhoisEntry:
                    pass
                elif type_ == win_registry_key_object.WinRegistryKey:
                    pass
                elif type_ == network_packet_object.NetworkPacket:
                    pass
                elif type_ == http_session_object.HTTPSession:
                    pass
                elif type_ == pipe_object.Pipe:
                    mispEvent.add_attribute('named pipe', ast_eval(str(obj.name)), comment=pkg.title or None)
                elif type_ == as_object.AS:
                    mispEvent.add_attribute('AS', ast_eval(str(obj.number)),
                                            comment=pkg.title or ast_eval(str(obj.name)) or None)
                elif type_ == win_executable_file_object.WinExecutableFile:
                    pass
                elif type_ == win_process_object.WinProcess:
                    pass
                elif type_ == x509_certificate_object.X509Certificate:
                    pass
                else:
                    log.debug("Type not syncing %s", type_)
            else:
                pass
        else:
            pass  # Other objects. TODO.
    except Exception as ex:
        log.error(ex)
        log.exception(ex, exc_info=True)

    return mispEvent

1	#!/usr/bin/env python3
2	import re	3✔
3	import cybox	3✔
4	import logging	3✔
5	import hashlib	3✔
6	import ast	3✔
7	from pymisp import mispevent	3✔
8	from misp_stix_converter.converters.lint_roller import lintRoll	3✔
9	from stix.core import STIXPackage	3✔
10
11
12	# Cybox cybox don't we all love cybox children
13	from cybox.objects import email_message_object, file_object, address_object, socket_address_object	3✔
14	from cybox.objects import domain_name_object, hostname_object, uri_object	3✔
15	from cybox.objects import mutex_object, whois_object	3✔
16	from cybox.objects import as_object, http_session_object	3✔
17	from cybox.objects import pipe_object, network_packet_object, win_registry_key_object	3✔
18	from cybox.objects import x509_certificate_object, win_executable_file_object, win_process_object	3✔
19
20
21	# Just a little containment file for STIX -> MISP conversion
22	ipre = re.compile("([0-9]{1,3}.){3}[0-9]{1,3}")	3✔
23	log = logging.getLogger("__main__")	3✔
24
25
26	def uniq(lst):	3✔
27	return_list = []	×
28	for elem in lst:	×
29	if elem not in return_list:	×
30	return_list.append(elem)	×
31	return return_list	×
32
33
34	def ast_eval(node):	3✔
35	try:	3✔
36	node = ast.literal_eval(node)	3✔
37	if isinstance(node, list):	3!
38	node = uniq(node)	×
39	return node	3✔
40	except ValueError:	3✔
41	return str(node)	3✔
42	except SyntaxError:	3✔
43	return str(node)	3✔
44
45
46	def parseRelated(obj, mispEvent, pkg):	3✔
47	"""Dedicated to parse related object
48	Support for first level "importRelated.related_objects"
49	"""
50
51	for i in obj.related_object:	×
52	type_ = type(i.properties)	×
53	# Here comes the sun (and of course the fun) for related objects! (DB)
54	if type_ == address_object.Address:	×
55	buildAddressAttribute(i.properties, mispEvent, pkg)	×
56
57	elif type_ == domain_name_object.DomainName:	×
58	buildDomainNameAttribute(i.properties, mispEvent, pkg)	×
59
60	elif type_ == hostname_object.Hostname:	×
61	buildHostnameAttribute(i.properties, mispEvent, pkg)	×
62
63	elif type_ == socket_address_object.SocketAddress:	×
64	buildAddressAttribute(i.properties, mispEvent, pkg, True)	×
65
66	elif type_ == uri_object.URI:	×
67	buildURIAttribute(i.properties, mispEvent, pkg)	×
68
69	elif type_ == file_object.File:	×
70	buildFileAttribute(i.properties, mispEvent, pkg)	×
71
72	elif type_ == email_message_object.EmailMessage:	×
73	buildEmailMessageAttribute(i.properties, mispEvent, pkg)	×
74
75	return mispEvent	×
76
77
78	def parseAttachment(obj, mispEvent, pkg):	3✔
79	"""
80	Limited support for email attachments, just consider this a work in progress
81	"""
82
83	for i in obj:	3✔
84	# Assuming any email attached file is a... file!
85	buildFileAttribute(i.properties, mispEvent, pkg)	3✔
86
87	return mispEvent	3✔
88
89
90	def buildFileAttribute(obj, mispEvent, pkg, importRelated=False):	3✔
91	"""
92	All you can get by a File object in a single method
93	TODO: all possible attributes are not yet parsed
94	"""
95
96	if obj.file_name:	3✔
97	mispEvent.add_attribute('filename', ast_eval(str(obj.file_name)), comment=pkg.title or None)	3✔
98
99	# Added support to file_extension
100	# No MISP object available for file extension
101	# I propose to use pattern-in-file. Suggestions are welcome! (DB)
102	if obj.file_extension:	3✔
103	mispEvent.add_attribute('pattern-in-file', ast_eval(str(obj.file_extension)), comment=pkg.title or None)	3✔
104
105	if obj.size_in_bytes:	3✔
106	mispEvent.add_attribute('size-in-bytes', ast_eval(str(obj.size_in_bytes)), comment=pkg.title or None)	3✔
107
108	if obj.md5:	3✔
109	# We actually have to check the length
110	# An actual report had supposed md5s of length 31. Silly.
111	if len(str(obj.md5)) == 32:	3!
112	mispEvent.add_attribute('md5', ast_eval(str(obj.md5)), comment=pkg.title or None)	3✔
113
114	if obj.sha1:	3!
115	if len(str(obj.sha1)) == 40:	×
116	mispEvent.add_attribute('sha1', ast_eval(str(obj.sha1)), comment=pkg.title or None)	×
117
118	if obj.sha256:	3✔
119	if len(str(obj.sha256)) == 64:	3!
120	mispEvent.add_attribute('sha256', ast_eval(str(obj.sha256)), comment=pkg.title or None)	3✔
121
122	# Added support for SHA512 (DB)
123	if obj.sha512:	3!
124	if len(str(obj.sha512)) == 128:	×
125	mispEvent.add_attribute('sha512', ast_eval(str(obj.sha512)), comment=pkg.title or None)	×
126
127	if importRelated and pkg.object_.related_objects:	3!
128	parseRelated(pkg.object_.related_objects, mispEvent, pkg)	×
129
130	return mispEvent	3✔
131
132
133	def buildAddressAttribute(obj, mispEvent, pkg, importRelated=False):	3✔
134
135	if obj.is_source:	3!
136	mispEvent.add_attribute('ip-src', ast_eval(str(obj.address_value)), comment=pkg.title or None)	×
137
138	elif obj.is_destination:	3!
139	mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value)), comment=pkg.title or None)	×
140	else:
141	# We don't know, first check if it's an IP range
142	if hasattr(obj, "condition") and obj.condition:	3✔
143	if obj.condition == "InclusiveBetween":	3!
144	# Ok, so it's a range. hm. Shall we add them seperately#comma#or together?
145	mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value[0])))	×
146	mispEvent.add_attribute('ip-dst', ast_eval(str(obj.add_attribute[1])))	×
147	elif obj.condition == "Equals":	3!
148	mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value)), comment=pkg.title or None)	3✔
149	else:
150	# Don't have anything to go on
151	mispEvent.add_attribute('ip-dst', ast_eval(str(obj.address_value)), comment=pkg.title or None)	3✔
152
153	if importRelated and pkg.object_.related_objects:	3!
154	parseRelated(pkg.object_.related_objects, mispEvent, pkg)	×
155
156	return mispEvent	3✔
157
158
159	def buildEmailMessageAttribute(obj, mispEvent, pkg, importRelated=False):	3✔
160	if obj.header:	3✔
161	# We have a header, can check for to/from etc etc
162	if obj.header.from_:	3!
163	mispEvent.add_attribute('email-src',	×
164	ast_eval(str(obj.header.from_.address_value)),
165	comment=pkg.title or None)
166	if obj.header.to:	3!
167	for mail in obj.header.to:	×
168	mispEvent.add_attribute('email-dst', ast_eval(mail.address_value), comment=pkg.title or None)	×
169	if obj.header.subject:	3!
170	mispEvent.add_attribute('email-subject', ast_eval(str(obj.header.subject)), comment=pkg.title or None)	3✔
171
172	if obj.attachments and pkg.object_.related_objects:	3✔
173	parseAttachment(pkg.object_.related_objects, mispEvent, pkg)	3✔
174
175	elif importRelated and pkg.object_.related_objects:	3!
176	parseRelated(pkg.object_.related_objects, mispEvent, pkg)	×
177
178	return mispEvent	3✔
179
180
181	def buildDomainNameAttribute(obj, mispEvent, pkg, importRelated=False):	3✔
182	mispEvent.add_attribute('domain', ast_eval(str(obj.value)), comment=pkg.title or None)	×
183
184	if importRelated and pkg.object_.related_objects:	×
185	parseRelated(pkg.object_.related_objects, mispEvent, pkg)	×
186
187	return mispEvent	×
188
189
190	def buildHostnameAttribute(obj, mispEvent, pkg, importRelated=False):	3✔
191	mispEvent.add_attribute('hostname', ast_eval(str(obj.hostname_value)), comment=pkg.title or None)	×
192
193	if importRelated and pkg.object_.related_objects:	×
194	parseRelated(pkg.object_.related_objects, mispEvent, pkg)	×
195
196	return mispEvent	×
197
198
199	def buildURIAttribute(obj, mispEvent, pkg, importRelated=False):	3✔
200	mispEvent.add_attribute('url', ast_eval(str(obj.value)), comment=pkg.title or None)	3✔
201
202	if importRelated and pkg.object_.related_objects:	3!
203	parseRelated(pkg.object_.related_objects, mispEvent, pkg)	×
204
205	return mispEvent	3✔
206
207
208	def identifyHash(hsh):	3✔
209	"""
210	What's that hash!?
211	"""
212
213	possible_hashes = []	×
214
215	hashes = [x for x in hashlib.algorithms_guaranteed]	×
216
217	for h in hashes:	×
218	if len(str(hsh)) == len(hashlib.new(h).hexdigest()):	×
219	possible_hashes.append(h)	×
220	possible_hashes.append("filename\|{}".format(h))	×
221	return possible_hashes	×
222
223
224	def buildEvent(pkg, **kwargs):	3✔
225	log.info("Building Event...")	3✔
226	if not pkg.stix_header:	3✔
227	title = "STIX Import"	3✔
228	else:
229	if not pkg.stix_header.title:	3✔
230	title = "STIX Import"	3✔
231	else:
232	title = pkg.stix_header.title	3✔
233	log.info("Using title %s", title)	3✔
234
235	log.debug("Seting up MISPEvent...")	3✔
236	event = mispevent.MISPEvent()	3✔
237	event.distribution = kwargs.get("distribution", 0)	3✔
238	event.threat_level_id = kwargs.get("threat_level_id", 3)	3✔
239	event.analysis = kwargs.get("analysis", 0)	3✔
240	event.info = title	3✔
241
242	if hasattr(pkg, "description"):	3!
243	log.debug("Found description %s", pkg.description)	×
244	event.add_attribute("comment", pkg.description)	×
245
246	log.debug("Beginning to Lint_roll...")	3✔
247	ids = []	3✔
248	to_process = []	3✔
249	for obj in lintRoll(pkg):	3✔
250	if isinstance(obj, cybox.core.observable.Observable):	3✔
251	if obj.id_ not in ids:	3✔
252	ids.append(obj.id_)	3✔
253	to_process.append(obj)	3✔
254
255	log.debug("Processing %s object...", len(to_process))	3✔
256	for obj in to_process:	3✔
257	log.debug("Working on %s...", obj)	3✔
258	# This will find literally every object ever.
259	try:	3✔
260	event = buildAttribute(obj, event)	3✔
261	except Exception as ex:	×
262	log.exception(ex)	×
263	# Now make sure we only have unique items
264	log.debug("Making sure we only have Unique attributes...")	3✔
265
266	uniqueAttribValues = []	3✔
267
268	for attrindex, attrib in enumerate(event.attributes):	3✔
269	if attrib.value not in uniqueAttribValues:	3✔
270	uniqueAttribValues.append(attrib.value)	3✔
271	else:
272	log.debug("Removed duplicated attribute in package: "+attrib.value)	3✔
273	event.attributes.pop(attrindex)	3✔
274
275	log.debug("Finished parsing attributes.")	3✔
276	return event	3✔
277
278
279	def buildAttribute(pkg, mispEvent):	3✔
280	try:	3✔
281	# Check if the object is a cybox observable
282	if isinstance(pkg, cybox.core.observable.Observable):	3!
283	if hasattr(pkg, "object_") and pkg.object_:	3!
284
285	obj = pkg.object_.properties	3✔
286
287	# It's a proper object!
288	type_ = type(obj)	3✔
289	# Here comes the fun!
290	if type_ == address_object.Address:	3✔
291	# Now script uses buildAddressAttribute (DB)
292	buildAddressAttribute(obj, mispEvent, pkg, True)	3✔
293
294	elif type_ == domain_name_object.DomainName:	3!
295	# Now script uses buildDomainNameAttribute (DB)
296	buildDomainNameAttribute(obj, mispEvent, pkg, True)	×
297
298	elif type_ == hostname_object.Hostname:	3!
299	# Now script uses buildHostnameAttribute
300	buildHostnameAttribute(obj, mispEvent, pkg, True)	×
301
302	elif type_ == socket_address_object.SocketAddress:	3!
303	if obj.ip_address:	×
304	buildAddressAttribute(obj.ip_address, mispEvent, pkg, True)	×
305	if obj.hostname:	×
306	buildHostnameAttribute(obj.hostname, mispEvent, pkg, True)	×
307
308	elif type_ == uri_object.URI:	3✔
309	# Now script uses buildURIAttribute (DB)
310	buildURIAttribute(obj, mispEvent, pkg, True)	3✔
311
312	elif type_ == file_object.File:	3✔
313	# Now script uses buildFileAttribute (DB)
314	buildFileAttribute(obj, mispEvent, pkg, True)	3✔
315
316	elif type_ == email_message_object.EmailMessage:	3✔
317	# Now script uses buildEmailMessageAttribute (DB)
318	buildEmailMessageAttribute(obj, mispEvent, pkg, True)	3✔
319
320	elif type_ == mutex_object.Mutex:	3!
321	mispEvent.add_attribute('mutex', ast_eval(str(obj.name)), comment=pkg.title or None)	×
322	elif type_ == whois_object.WhoisEntry:	3!
323	pass	×
324	elif type_ == win_registry_key_object.WinRegistryKey:	3!
325	pass	×
326	elif type_ == network_packet_object.NetworkPacket:	3!
327	pass	×
328	elif type_ == http_session_object.HTTPSession:	3!
329	pass	×
330	elif type_ == pipe_object.Pipe:	3!
331	mispEvent.add_attribute('named pipe', ast_eval(str(obj.name)), comment=pkg.title or None)	×
332	elif type_ == as_object.AS:	3!
333	mispEvent.add_attribute('AS', ast_eval(str(obj.number)),	×
334	comment=pkg.title or ast_eval(str(obj.name)) or None)
335	elif type_ == win_executable_file_object.WinExecutableFile:	3!
336	pass	×
337	elif type_ == win_process_object.WinProcess:	3!
338	pass	×
339	elif type_ == x509_certificate_object.X509Certificate:	3!
340	pass	3✔
341	else:
342	log.debug("Type not syncing %s", type_)	×
343	else:
344	pass	3✔
345	else:
346	pass # Other objects. TODO.	3✔
347	except Exception as ex:	×
348	log.error(ex)	×
349	log.exception(ex, exc_info=True)	×
350
351	return mispEvent	3✔

MISP / MISP-STIX-Converter / 139

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous