MISP
/
cti-python-stix2
mirror of https://github.com/MISP/cti-python-stix2
2
0
Fork 0
Code Issues Releases Wiki Activity

Formatting changes, skip add/remove filter test, change deduplicate() approach.

stix2.1
Emmanuelle Vargas-Gonzalez 2017-08-09 13:31:07 -04:00
parent b8c96e37a2
commit a4ead4f6e7
4 changed files with 19 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
stix2/sources/__init__.py
View File

@ -23,10 +23,6 @@ from six import iteritems
def make_id():
return str(uuid.uuid4())
# STIX 2.0 fields used to denote object version
STIX_VERSION_FIELDS = ['id', 'modified']
# Currently, only STIX 2.0 common SDO fields (that are not compex objects)
# are supported for filtering on
STIX_COMMON_FIELDS = [
@ -705,8 +701,6 @@ class CompositeDataSource(object):
def filters(self):
"""return filters attached to Composite Data Source
Args:
Returns:
(list): the list of filters currently attached to the Data Source
@ -727,18 +721,12 @@ class CompositeDataSource(object):
(list): unique set of the passed list of STIX objects
"""
unique = []
dont_have = False
for i in stix_obj_list:
dont_have = False
for j in unique:
for field in STIX_VERSION_FIELDS:
if not i[field] == j[field]:
dont_have = True
break
if dont_have:
unique.append(i)
return unique
unique_objs = {}
for obj in stix_obj_list:
unique_objs[(obj["id"], obj["modified"])] = obj
return list(unique_objs.values())
class STIXCommonPropertyFilters():
@ -775,7 +763,7 @@ class STIXCommonPropertyFilters():
return -1
@classmethod
def _boolean(filter_, stix_obj_field):
def _boolean(cls, filter_, stix_obj_field):
if filter_["op"] == "=":
return stix_obj_field == filter_["value"]
elif filter_["op"] == "!=":
@ -802,7 +790,7 @@ class STIXCommonPropertyFilters():
@classmethod
def external_references(cls, filter_, stix_obj):
'''
"""
stix object's can have a list of external references
external-reference properties:
@ -811,7 +799,7 @@ class STIXCommonPropertyFilters():
external_reference.url (string)
external_reference.hashes (hash, but for filtering purposes , a string)
external_reference.external_id (string)
'''
"""
for er in stix_obj["external_references"]:
# grab er property name from filter field
filter_field = filter_["field"].split(".")[1]
@ -822,13 +810,13 @@ class STIXCommonPropertyFilters():
@classmethod
def granular_markings(cls, filter_, stix_obj):
'''
"""
stix object's can have a list of granular marking references
granular-marking properties:
granular-marking.marking_ref (id)
granular-marking.selectors (string)
'''
"""
for gm in stix_obj["granular_markings"]:
# grab gm property name from filter field
filter_field = filter_["field"].split(".")[1]

5
stix2/sources/filesystem.py
View File

@ -12,13 +12,12 @@ TODO: Test everything
import json
import os
from sources import DataSink, DataSource, DataStore, make_id
from stix2.sources import DataSink, DataSource, DataStore, make_id
from stix2 import Bundle
class FileSystemStore(DataStore):
"""
"""
def __init__(self, stix_dir="stix_data", name="FileSystemStore"):
self.name = name
@ -54,7 +53,7 @@ class FileSystemSink(DataSink):
stix_objs = []
for stix_obj in stix_objs:
path = os.path.join(self.stix_dir, stix_obj["type"], stix_obj["id"])
json.dump(Bundle([stix_obj]), open(path, 'w+', indent=4))
json.dump(Bundle([stix_obj]), open(path, 'w+'), indent=4)
class FileSystemSource(DataSource):

10
stix2/sources/memory.py
View File

@ -58,7 +58,7 @@ class MemoryStore(DataStore):
if r.is_valid:
self.data[stix_obj["id"]] = stix_obj
else:
print("Error: STIX object %s is not valid under STIX 2 validator.") % stix_obj["id"]
print("Error: STIX object %s is not valid under STIX 2 validator." % stix_obj["id"])
print(r)
self.source = MemorySource(stix_data=self.data, _store=True)
@ -112,7 +112,7 @@ class MemorySink(DataSink):
if r.is_valid:
self.data[stix_obj["id"]] = stix_obj
else:
print("Error: STIX object %s is not valid under STIX 2 validator.") % stix_obj["id"]
print("Error: STIX object %s is not valid under STIX 2 validator." % stix_obj["id"])
print(r)
else:
raise ValueError("stix_data must be in bundle format or raw list")
@ -136,7 +136,7 @@ class MemorySink(DataSink):
if r.is_valid:
self.data[stix_obj["id"]] = stix_obj
else:
print("Error: STIX object %s is not valid under STIX 2 validator.") % stix_obj["id"]
print("Error: STIX object %s is not valid under STIX 2 validator." % stix_obj["id"])
print(r)
else:
raise ValueError("stix_data must be in bundle format or raw list")
@ -185,7 +185,7 @@ class MemorySource(DataSource):
if r.is_valid:
self.data[stix_obj["id"]] = stix_obj
else:
print("Error: STIX object %s is not valid under STIX 2 validator.") % stix_obj["id"]
print("Error: STIX object %s is not valid under STIX 2 validator." % stix_obj["id"])
print(r)
else:
raise ValueError("stix_data must be in bundle format or raw list")
@ -269,5 +269,5 @@ class MemorySource(DataSource):
for stix_obj in stix_data["objects"]:
self.data[stix_obj["id"]] = stix_obj
else:
print("Error: STIX data loaded from file (%s) was found to not be validated by STIX 2 Validator") % file_path
print("Error: STIX data loaded from file (%s) was found to not be validated by STIX 2 Validator" % file_path)
print(r)

1
stix2/test/test_data_sources.py
View File

@ -74,6 +74,7 @@ def test_parse_taxii_filters():
assert taxii_filters == expected_params
@pytest.skip
def test_add_get_remove_filter():
# First 3 filters are valid, remaining fields are erroneous in some way