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

Some changes. More fixes coming soon, hopefully

master
Desai, Kartikey H 2019-09-06 18:08:27 -04:00
parent 8f773fd556
commit 5b6592e2dc
5 changed files with 73 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
stix2/base.py
View File

@ -357,11 +357,12 @@ class _Observable(_STIXBase):
return return
from .properties import ObjectReferenceProperty from .properties import ObjectReferenceProperty
if isinstance(prop, ObjectReferenceProperty): if prop_name.endswith('_ref'):
if prop_name.endswith('_ref'): if isinstance(prop, ObjectReferenceProperty):
ref = kwargs[prop_name] ref = kwargs[prop_name]
self._check_ref(ref, prop, prop_name) self._check_ref(ref, prop, prop_name)
elif prop_name.endswith('_refs'): elif prop_name.endswith('_refs'):
if isinstance(prop.contained, ObjectReferenceProperty):
for ref in kwargs[prop_name]: for ref in kwargs[prop_name]:
self._check_ref(ref, prop, prop_name) self._check_ref(ref, prop, prop_name)
@ -392,7 +393,12 @@ class _Observable(_STIXBase):
if streamlined_obj_vals: if streamlined_obj_vals:
data = canonicalize(streamlined_obj_vals, utf8=False) data = canonicalize(streamlined_obj_vals, utf8=False)
return required_prefix + six.text_type(uuid.uuid5(namespace, data)) # print (str(type(data)))
try:
return required_prefix + six.text_type(uuid.uuid5(namespace, data))
except UnicodeDecodeError:
return required_prefix + six.text_type(uuid.uuid5(namespace, six.binary_type(data)))
# return required_prefix + six.text_type(uuid.uuid5(namespace, data))
# We return None if there are no values specified for any of the id-contributing-properties # We return None if there are no values specified for any of the id-contributing-properties
return None return None

22
stix2/datastore/filters.py
View File

@ -98,6 +98,15 @@ class Filter(collections.namedtuple('Filter', ['property', 'op', 'value'])):
print (filter_value) print (filter_value)
print ('\n') print ('\n')
print ('$$$$$') print ('$$$$$')
pass
else:
# print ('222222222')
# print (stix_obj_property)
# print ('\n')
# print (filter_value)
# print ('\n')
# print ('222222222')
pass
return stix_obj_property == filter_value return stix_obj_property == filter_value
elif self.op == "!=": elif self.op == "!=":
return stix_obj_property != filter_value return stix_obj_property != filter_value
@ -107,12 +116,21 @@ class Filter(collections.namedtuple('Filter', ['property', 'op', 'value'])):
if isinstance(filter_value, dict): if isinstance(filter_value, dict):
boolB = filter_value in stix_obj_property.values() boolB = filter_value in stix_obj_property.values()
if boolB is False: if boolB is False:
print ('$$$$$') print ('@@@@@@')
print (filter_value) print (filter_value)
print ('\n') print ('\n')
print (stix_obj_property.values()) print (stix_obj_property.values())
print ('\n') print ('\n')
print ('$$$$$') print ('@@@@@@@')
pass
else:
# print ('55555555555')
# print (filter_value)
# print ('\n')
# print (stix_obj_property.values())
# print ('\n')
# print ('55555555555')
pass
return filter_value in stix_obj_property.values() return filter_value in stix_obj_property.values()
else: else:
return filter_value in stix_obj_property return filter_value in stix_obj_property

2
stix2/properties.py
View File

@ -457,7 +457,7 @@ class ReferenceProperty(Property):
elif self.valid_types == ["only_SCO"]: elif self.valid_types == ["only_SCO"]:
self.valid_types = STIX2_OBJ_MAPS['v21']['observables'].keys() self.valid_types = STIX2_OBJ_MAPS['v21']['observables'].keys()
elif self.valid_types == ["only_SCO_&_SRO"]: elif self.valid_types == ["only_SCO_&_SRO"]:
self.valid_types = STIX2_OBJ_MAPS['v21']['observables'].keys() + ['relationship', 'sighting'] self.valid_types = list(STIX2_OBJ_MAPS['v21']['observables'].keys()) + ['relationship', 'sighting']
if possible_prefix[:-2] in self.valid_types: if possible_prefix[:-2] in self.valid_types:
required_prefix = possible_prefix required_prefix = possible_prefix

2
stix2/test/v21/test_datastore_filters.py
View File

@ -111,7 +111,7 @@ filters = [
Filter("object_marking_refs", "=", "marking-definition--613f2e26-0000-4000-8000-b8e91df99dc9"), Filter("object_marking_refs", "=", "marking-definition--613f2e26-0000-4000-8000-b8e91df99dc9"),
Filter("granular_markings.selectors", "in", "description"), Filter("granular_markings.selectors", "in", "description"),
Filter("external_references.source_name", "=", "CVE"), Filter("external_references.source_name", "=", "CVE"),
Filter("objects", "=", {"0": {"type": "file", "spec_version": "2.1", "id": "file--42a7175a-42cc-508f-8fa7-23b330aff876", "name": "HAL 9000.exe"}}), Filter("objects", "=", {"0": {"type": "file", "id": "file--42a7175a-42cc-508f-8fa7-23b330aff876", "name": "HAL 9000.exe", "spec_version": "2.1"}}),
Filter("objects", "contains", {"type": "file", "id": "file--42a7175a-42cc-508f-8fa7-23b330aff876", "name": "HAL 9000.exe", "spec_version": "2.1"}), Filter("objects", "contains", {"type": "file", "id": "file--42a7175a-42cc-508f-8fa7-23b330aff876", "name": "HAL 9000.exe", "spec_version": "2.1"}),
Filter("labels", "contains", "heartbleed"), Filter("labels", "contains", "heartbleed"),
] ]

92
stix2/test/v21/test_observed_data.py
View File

@ -364,17 +364,21 @@ def test_parse_autonomous_system_valid(data):
"type": "email-addr", "type": "email-addr",
"value": "john@example.com", "value": "john@example.com",
"display_name": "John Doe", "display_name": "John Doe",
"belongs_to_ref": "0" "belongs_to_ref": "user-account--fc07c1af-6b11-41f8-97a4-47920d866a91"
}""", }""",
], ],
) )
def test_parse_email_address(data): def test_parse_email_address(data):
odata = stix2.parse_observable(data, {"0": "user-account"}, version='2.1') odata = stix2.parse_observable(data, version='2.1')
assert odata.type == "email-addr" assert odata.type == "email-addr"
odata_str = re.compile('"belongs_to_ref": "0"', re.DOTALL).sub('"belongs_to_ref": "3"', data) odata_str = re.compile(
with pytest.raises(stix2.exceptions.InvalidObjRefError): '"belongs_to_ref": "user-account--fc07c1af-6b11-41f8-97a4-47920d866a91"', re.DOTALL,
stix2.parse_observable(odata_str, {"0": "user-account"}, version='2.1') ).sub(
'"belongs_to_ref": "mutex--9be6365f-b89c-48c0-9340-6953f6595718"', data,
)
with pytest.raises(stix2.exceptions.InvalidValueError):
stix2.parse_observable(odata_str, version='2.1')
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -385,12 +389,12 @@ def test_parse_email_address(data):
"is_multipart": true, "is_multipart": true,
"content_type": "multipart/mixed", "content_type": "multipart/mixed",
"date": "2016-06-19T14:20:40.000Z", "date": "2016-06-19T14:20:40.000Z",
"from_ref": "1", "from_ref": "email-addr--d4ef7e1f-086d-5ff4-bce4-312ddc3eae76",
"to_refs": [ "to_refs": [
"2" "email-addr--8b0eb924-208c-5efd-80e5-84e2d610e54b"
], ],
"cc_refs": [ "cc_refs": [
"3" "email-addr--1766f860-5cf3-5697-8789-35f1242663d5"
], ],
"subject": "Check out this picture of a cat!", "subject": "Check out this picture of a cat!",
"additional_header_fields": { "additional_header_fields": {
@ -407,12 +411,12 @@ def test_parse_email_address(data):
{ {
"content_type": "image/png", "content_type": "image/png",
"content_disposition": "attachment; filename=\\"tabby.png\\"", "content_disposition": "attachment; filename=\\"tabby.png\\"",
"body_raw_ref": "4" "body_raw_ref": "artifact--80b04ad8-db52-464b-a85a-a44a5f3a60c5"
}, },
{ {
"content_type": "application/zip", "content_type": "application/zip",
"content_disposition": "attachment; filename=\\"tabby_pics.zip\\"", "content_disposition": "attachment; filename=\\"tabby_pics.zip\\"",
"body_raw_ref": "5" "body_raw_ref": "file--e63474fc-b386-5630-a003-1b555e22f99b"
} }
] ]
} }
@ -420,15 +424,7 @@ def test_parse_email_address(data):
], ],
) )
def test_parse_email_message(data): def test_parse_email_message(data):
valid_refs = { odata = stix2.parse_observable(data, version='2.1')
"0": "email-message",
"1": "email-addr",
"2": "email-addr",
"3": "email-addr",
"4": "artifact",
"5": "file",
}
odata = stix2.parse_observable(data, valid_refs, version='2.1')
assert odata.type == "email-message" assert odata.type == "email-message"
assert odata.body_multipart[0].content_disposition == "inline" assert odata.body_multipart[0].content_disposition == "inline"
@ -438,8 +434,8 @@ def test_parse_email_message(data):
""" """
{ {
"type": "email-message", "type": "email-message",
"from_ref": "0", "from_ref": "email-addr--d4ef7e1f-086d-5ff4-bce4-312ddc3eae76",
"to_refs": ["1"], "to_refs": ["email-addr--8b0eb924-208c-5efd-80e5-84e2d610e54b"],
"is_multipart": true, "is_multipart": true,
"date": "1997-11-21T15:55:06.000Z", "date": "1997-11-21T15:55:06.000Z",
"subject": "Saying Hello", "subject": "Saying Hello",
@ -449,12 +445,8 @@ def test_parse_email_message(data):
], ],
) )
def test_parse_email_message_not_multipart(data): def test_parse_email_message_not_multipart(data):
valid_refs = {
"0": "email-addr",
"1": "email-addr",
}
with pytest.raises(stix2.exceptions.DependentPropertiesError) as excinfo: with pytest.raises(stix2.exceptions.DependentPropertiesError) as excinfo:
stix2.parse_observable(data, valid_refs, version='2.1') stix2.parse_observable(data, version='2.1')
assert excinfo.value.cls == stix2.v21.EmailMessage assert excinfo.value.cls == stix2.v21.EmailMessage
assert excinfo.value.dependencies == [("is_multipart", "body")] assert excinfo.value.dependencies == [("is_multipart", "body")]
@ -464,18 +456,21 @@ def test_parse_email_message_not_multipart(data):
"data", [ "data", [
""""0": { """"0": {
"type": "file", "type": "file",
"id": "file--ecd47d73-15e4-5250-afda-ef8897b22340",
"hashes": { "hashes": {
"SHA-256": "ceafbfd424be2ca4a5f0402cae090dda2fb0526cf521b60b60077c0f622b285a" "SHA-256": "ceafbfd424be2ca4a5f0402cae090dda2fb0526cf521b60b60077c0f622b285a"
} }
}, },
"1": { "1": {
"type": "file", "type": "file",
"id": "file--65f2873d-38c2-56b4-bfa5-e3ef21e8a3c3",
"hashes": { "hashes": {
"SHA-256": "19c549ec2628b989382f6b280cbd7bb836a0b461332c0fe53511ce7d584b89d3" "SHA-256": "19c549ec2628b989382f6b280cbd7bb836a0b461332c0fe53511ce7d584b89d3"
} }
}, },
"2": { "2": {
"type": "file", "type": "file",
"id": "file--ef2d6dca-ec7d-5ab7-8dd9-ec9c0dee0eac",
"hashes": { "hashes": {
"SHA-256": "0969de02ecf8a5f003e3f6d063d848c8a193aada092623f8ce408c15bcb5f038" "SHA-256": "0969de02ecf8a5f003e3f6d063d848c8a193aada092623f8ce408c15bcb5f038"
} }
@ -490,9 +485,9 @@ def test_parse_email_message_not_multipart(data):
"extensions": { "extensions": {
"archive-ext": { "archive-ext": {
"contains_refs": [ "contains_refs": [
"0", "file--ecd47d73-15e4-5250-afda-ef8897b22340",
"1", "file--65f2873d-38c2-56b4-bfa5-e3ef21e8a3c3",
"2" "file--ef2d6dca-ec7d-5ab7-8dd9-ec9c0dee0eac"
] ]
} }
} }
@ -503,7 +498,11 @@ def test_parse_file_archive(data):
odata_str = OBJECTS_REGEX.sub('"objects": { %s }' % data, EXPECTED) odata_str = OBJECTS_REGEX.sub('"objects": { %s }' % data, EXPECTED)
odata = stix2.parse(odata_str, version="2.1") odata = stix2.parse(odata_str, version="2.1")
assert all(x in odata.objects["3"].extensions['archive-ext'].contains_refs assert all(x in odata.objects["3"].extensions['archive-ext'].contains_refs
for x in ["0", "1", "2"]) for x in [
"file--ecd47d73-15e4-5250-afda-ef8897b22340",
"file--65f2873d-38c2-56b4-bfa5-e3ef21e8a3c3",
"file--ef2d6dca-ec7d-5ab7-8dd9-ec9c0dee0eac",
])
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -514,12 +513,12 @@ def test_parse_file_archive(data):
"is_multipart": true, "is_multipart": true,
"content_type": "multipart/mixed", "content_type": "multipart/mixed",
"date": "2016-06-19T14:20:40.000Z", "date": "2016-06-19T14:20:40.000Z",
"from_ref": "1", "from_ref": "email-addr--d4ef7e1f-086d-5ff4-bce4-312ddc3eae76",
"to_refs": [ "to_refs": [
"2" "email-addr--8b0eb924-208c-5efd-80e5-84e2d610e54b"
], ],
"cc_refs": [ "cc_refs": [
"3" "email-addr--1766f860-5cf3-5697-8789-35f1242663d5"
], ],
"subject": "Check out this picture of a cat!", "subject": "Check out this picture of a cat!",
"additional_header_fields": { "additional_header_fields": {
@ -540,7 +539,7 @@ def test_parse_file_archive(data):
{ {
"content_type": "application/zip", "content_type": "application/zip",
"content_disposition": "attachment; filename=\\"tabby_pics.zip\\"", "content_disposition": "attachment; filename=\\"tabby_pics.zip\\"",
"body_raw_ref": "5" "body_raw_ref": "file--e63474fc-b386-5630-a003-1b555e22f99b"
} }
] ]
} }
@ -548,16 +547,8 @@ def test_parse_file_archive(data):
], ],
) )
def test_parse_email_message_with_at_least_one_error(data): def test_parse_email_message_with_at_least_one_error(data):
valid_refs = {
"0": "email-message",
"1": "email-addr",
"2": "email-addr",
"3": "email-addr",
"4": "artifact",
"5": "file",
}
with pytest.raises(stix2.exceptions.InvalidValueError) as excinfo: with pytest.raises(stix2.exceptions.InvalidValueError) as excinfo:
stix2.parse_observable(data, valid_refs, version='2.1') stix2.parse_observable(data, version='2.1')
assert excinfo.value.cls == stix2.v21.EmailMessage assert excinfo.value.cls == stix2.v21.EmailMessage
assert "At least one of the" in str(excinfo.value) assert "At least one of the" in str(excinfo.value)
@ -569,8 +560,8 @@ def test_parse_email_message_with_at_least_one_error(data):
""" """
{ {
"type": "network-traffic", "type": "network-traffic",
"src_ref": "0", "src_ref": "ipv4-addr--e535b017-cc1c-566b-a3e2-f69f92ed9c4c",
"dst_ref": "1", "dst_ref": "ipv4-addr--78327430-9ad9-5632-ae3d-8e2fce8f5483",
"protocols": [ "protocols": [
"tcp" "tcp"
] ]
@ -580,13 +571,12 @@ def test_parse_email_message_with_at_least_one_error(data):
) )
def test_parse_basic_tcp_traffic(data): def test_parse_basic_tcp_traffic(data):
odata = stix2.parse_observable( odata = stix2.parse_observable(
data, {"0": "ipv4-addr", "1": "ipv4-addr"}, data, version='2.1',
version='2.1',
) )
assert odata.type == "network-traffic" assert odata.type == "network-traffic"
assert odata.src_ref == "0" assert odata.src_ref == "ipv4-addr--e535b017-cc1c-566b-a3e2-f69f92ed9c4c"
assert odata.dst_ref == "1" assert odata.dst_ref == "ipv4-addr--78327430-9ad9-5632-ae3d-8e2fce8f5483"
assert odata.protocols == ["tcp"] assert odata.protocols == ["tcp"]
@ -604,7 +594,7 @@ def test_parse_basic_tcp_traffic(data):
"src_byte_count": 35779, "src_byte_count": 35779,
"dst_byte_count": 935750, "dst_byte_count": 935750,
"encapsulates_refs": [ "encapsulates_refs": [
"4" "network-traffic--016914c3-b680-5df2-81c4-bb9ccf8dc8b0"
] ]
} }
""", """,
@ -612,7 +602,7 @@ def test_parse_basic_tcp_traffic(data):
) )
def test_parse_basic_tcp_traffic_with_error(data): def test_parse_basic_tcp_traffic_with_error(data):
with pytest.raises(stix2.exceptions.AtLeastOnePropertyError) as excinfo: with pytest.raises(stix2.exceptions.AtLeastOnePropertyError) as excinfo:
stix2.parse_observable(data, {"4": "network-traffic"}, version='2.1') stix2.parse_observable(data, version='2.1')
assert excinfo.value.cls == stix2.v21.NetworkTraffic assert excinfo.value.cls == stix2.v21.NetworkTraffic
assert excinfo.value.properties == ["dst_ref", "src_ref"] assert excinfo.value.properties == ["dst_ref", "src_ref"]