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

Update README and refactor code to make 2.0 default. Update some tests

master
Emmanuelle Vargas-Gonzalez 2018-11-01 09:21:02 -04:00
parent 5e5d10e7aa
commit 493bd65ead
7 changed files with 44 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
README.rst
View File

@ -63,12 +63,15 @@ For more in-depth documentation, please see `https://stix2.readthedocs.io/ <http
STIX 2.X Technical Specification Support STIX 2.X Technical Specification Support
---------------------------------------- ----------------------------------------
This version of python-stix2 supports STIX 2.1 by default. Although, the This version of python-stix2 brings initial support to STIX 2.1 currently at the
`stix2` Python library is built to support multiple versions of the STIX CSD level. The intention is to help debug components of the library and also
check for problems that should be fixed in the specification.
The `stix2` Python library is built to support multiple versions of the STIX
Technical Specification. With every major release of stix2 the ``import stix2`` Technical Specification. With every major release of stix2 the ``import stix2``
statement will automatically load the SDO/SROs equivalent to the most recent statement will automatically load the SDO/SROs equivalent to the most recent
supported 2.X Technical Specification. Please see the library documentation supported 2.X Committee Specification. Please see the library documentation for
for more details. more details.
Governance Governance
---------- ----------

4
stix2/__init__.py
View File

@ -35,7 +35,7 @@
# flake8: noqa # flake8: noqa
from .core import _collect_stix2_mappings, parse, parse_observable from .core import _collect_stix2_mappings, parse, parse_observable
from .v21 import * # This import will always be the latest STIX 2.X version from .v20 import * # This import will always be the latest STIX 2.X version
from .datastore import CompositeDataSource from .datastore import CompositeDataSource
from .datastore.filesystem import ( from .datastore.filesystem import (
FileSystemSink, FileSystemSource, FileSystemSink, FileSystemSource,
@ -78,4 +78,4 @@ from .version import __version__
_collect_stix2_mappings() _collect_stix2_mappings()
DEFAULT_VERSION = '2.1' # Default version will always be the latest STIX 2.X version DEFAULT_VERSION = '2.0' # Default version will always be the latest STIX 2.X version

5
stix2/core.py
View File

@ -97,6 +97,11 @@ def dict_to_stix2(stix_dict, allow_custom=False, version=None):
# bundles without spec_version are ambiguous. # bundles without spec_version are ambiguous.
if version: if version:
v = 'v' + version.replace('.', '') v = 'v' + version.replace('.', '')
elif 'spec_version' in stix_dict:
v = 'v' + stix_dict['spec_version'].replace('.', '')
elif any('spec_version' in x for x in stix_dict['objects']):
# Only on 2.1 we are allowed multiple version of SDOs/SROs.
v = 'v21'
else: else:
v = 'v' + stix2.DEFAULT_VERSION.replace('.', '') v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
else: else:

15
stix2/test/v20/test_workbench.py
View File

@ -1,7 +1,5 @@
import os import os
import pytest
import stix2 import stix2
from stix2.workbench import ( from stix2.workbench import (
AttackPattern, Campaign, CourseOfAction, ExternalReference, AttackPattern, Campaign, CourseOfAction, ExternalReference,
@ -24,7 +22,6 @@ from .constants import (
) )
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_environment(): def test_workbench_environment():
# Create a STIX object # Create a STIX object
@ -79,7 +76,6 @@ def test_workbench_get_all_identities():
assert resp[0].id == IDENTITY_ID assert resp[0].id == IDENTITY_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_get_all_indicators(): def test_workbench_get_all_indicators():
resp = indicators() resp = indicators()
assert len(resp) == 1 assert len(resp) == 1
@ -95,7 +91,6 @@ def test_workbench_get_all_intrusion_sets():
assert resp[0].id == INTRUSION_SET_ID assert resp[0].id == INTRUSION_SET_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_get_all_malware(): def test_workbench_get_all_malware():
mal = Malware(id=MALWARE_ID, **MALWARE_KWARGS) mal = Malware(id=MALWARE_ID, **MALWARE_KWARGS)
save(mal) save(mal)
@ -114,7 +109,6 @@ def test_workbench_get_all_observed_data():
assert resp[0].id == OBSERVED_DATA_ID assert resp[0].id == OBSERVED_DATA_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_get_all_reports(): def test_workbench_get_all_reports():
rep = Report(id=REPORT_ID, **REPORT_KWARGS) rep = Report(id=REPORT_ID, **REPORT_KWARGS)
save(rep) save(rep)
@ -124,7 +118,6 @@ def test_workbench_get_all_reports():
assert resp[0].id == REPORT_ID assert resp[0].id == REPORT_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_get_all_threat_actors(): def test_workbench_get_all_threat_actors():
thr = ThreatActor(id=THREAT_ACTOR_ID, **THREAT_ACTOR_KWARGS) thr = ThreatActor(id=THREAT_ACTOR_ID, **THREAT_ACTOR_KWARGS)
save(thr) save(thr)
@ -134,7 +127,6 @@ def test_workbench_get_all_threat_actors():
assert resp[0].id == THREAT_ACTOR_ID assert resp[0].id == THREAT_ACTOR_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_get_all_tools(): def test_workbench_get_all_tools():
tool = Tool(id=TOOL_ID, **TOOL_KWARGS) tool = Tool(id=TOOL_ID, **TOOL_KWARGS)
save(tool) save(tool)
@ -153,14 +145,12 @@ def test_workbench_get_all_vulnerabilities():
assert resp[0].id == VULNERABILITY_ID assert resp[0].id == VULNERABILITY_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_add_to_bundle(): def test_workbench_add_to_bundle():
vuln = Vulnerability(**VULNERABILITY_KWARGS) vuln = Vulnerability(**VULNERABILITY_KWARGS)
bundle = stix2.v20.Bundle(vuln) bundle = stix2.v20.Bundle(vuln)
assert bundle.objects[0].name == 'Heartbleed' assert bundle.objects[0].name == 'Heartbleed'
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_relationships(): def test_workbench_relationships():
rel = Relationship(INDICATOR_ID, 'indicates', MALWARE_ID) rel = Relationship(INDICATOR_ID, 'indicates', MALWARE_ID)
save(rel) save(rel)
@ -180,7 +170,6 @@ def test_workbench_created_by():
assert creator.id == IDENTITY_ID assert creator.id == IDENTITY_ID
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_related(): def test_workbench_related():
rel1 = Relationship(MALWARE_ID, 'targets', IDENTITY_ID) rel1 = Relationship(MALWARE_ID, 'targets', IDENTITY_ID)
rel2 = Relationship(CAMPAIGN_ID, 'uses', MALWARE_ID) rel2 = Relationship(CAMPAIGN_ID, 'uses', MALWARE_ID)
@ -196,7 +185,6 @@ def test_workbench_related():
assert len(resp) == 1 assert len(resp) == 1
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_workbench_related_with_filters(): def test_workbench_related_with_filters():
malware = Malware(labels=["ransomware"], name="CryptorBit", created_by_ref=IDENTITY_ID) malware = Malware(labels=["ransomware"], name="CryptorBit", created_by_ref=IDENTITY_ID)
rel = Relationship(malware.id, 'variant-of', MALWARE_ID) rel = Relationship(malware.id, 'variant-of', MALWARE_ID)
@ -214,7 +202,6 @@ def test_workbench_related_with_filters():
assert len(resp) == 1 assert len(resp) == 1
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_add_data_source(): def test_add_data_source():
fs_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stix2_data") fs_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stix2_data")
fs = FileSystemSource(fs_path) fs = FileSystemSource(fs_path)
@ -228,13 +215,11 @@ def test_add_data_source():
assert 'tool--242f3da3-4425-4d11-8f5c-b842886da966' in resp_ids assert 'tool--242f3da3-4425-4d11-8f5c-b842886da966' in resp_ids
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_additional_filter(): def test_additional_filter():
resp = tools(Filter('created_by_ref', '=', 'identity--c78cb6e5-0c4b-4611-8297-d1b8b55e40b5')) resp = tools(Filter('created_by_ref', '=', 'identity--c78cb6e5-0c4b-4611-8297-d1b8b55e40b5'))
assert len(resp) == 2 assert len(resp) == 2
@pytest.mark.skip(reason='The workbench is not working correctly for 2.0')
def test_additional_filters_list(): def test_additional_filters_list():
resp = tools([ resp = tools([
Filter('created_by_ref', '=', 'identity--c78cb6e5-0c4b-4611-8297-d1b8b55e40b5'), Filter('created_by_ref', '=', 'identity--c78cb6e5-0c4b-4611-8297-d1b8b55e40b5'),

34
stix2/test/v21/test_core.py
View File

@ -1,7 +1,7 @@
import pytest import pytest
import stix2 import stix2
from stix2 import DEFAULT_VERSION, core, exceptions from stix2 import core, exceptions
BUNDLE = { BUNDLE = {
"type": "bundle", "type": "bundle",
@ -52,29 +52,29 @@ def test_dict_to_stix2_bundle_with_version():
assert str(excinfo.value) == msg assert str(excinfo.value) == msg
def test_parse_observable_with_default_version(): def test_parse_observable_with_version():
observable = {"type": "file", "name": "foo.exe"} observable = {"type": "file", "name": "foo.exe"}
obs_obj = core.parse_observable(observable) obs_obj = core.parse_observable(observable, version='2.1')
v = 'v21'
assert 'v' + DEFAULT_VERSION.replace('.', '') in str(obs_obj.__class__) assert v in str(obs_obj.__class__)
def test_register_object_with_version(): def test_register_object_with_version():
bundle = core.dict_to_stix2(BUNDLE, version='2.1') bundle = core.dict_to_stix2(BUNDLE, version='2.1')
core._register_object(bundle.objects[0].__class__) core._register_object(bundle.objects[0].__class__)
v = 'v21'
v = 'v' + DEFAULT_VERSION.replace('.', '')
assert bundle.objects[0].type in core.STIX2_OBJ_MAPS[v]['objects'] assert bundle.objects[0].type in core.STIX2_OBJ_MAPS[v]['objects']
assert 'v' + DEFAULT_VERSION.replace('.', '') in str(bundle.objects[0].__class__) assert v in str(bundle.objects[0].__class__)
def test_register_marking_with_default_version(): def test_register_marking_with_version():
core._register_marking(stix2.TLP_WHITE.__class__) core._register_marking(stix2.v21.TLP_WHITE.__class__, version='2.1')
v = 'v' + DEFAULT_VERSION.replace('.', '') v = 'v21'
assert stix2.TLP_WHITE.type in core.STIX2_OBJ_MAPS[v]['markings'] assert stix2.v21.TLP_WHITE.definition._type in core.STIX2_OBJ_MAPS[v]['markings']
assert 'v' + DEFAULT_VERSION.replace('.', '') in str(stix2.TLP_WHITE.__class__) assert v in str(stix2.v21.TLP_WHITE.__class__)
def test_register_observable_with_default_version(): def test_register_observable_with_default_version():
@ -109,10 +109,10 @@ def test_register_observable_with_default_version():
}, },
) )
core._register_observable(observed_data.objects['0'].__class__) core._register_observable(observed_data.objects['0'].__class__)
v = 'v' + DEFAULT_VERSION.replace('.', '') v = 'v21'
assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables'] assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables']
assert 'v' + DEFAULT_VERSION.replace('.', '') in str(observed_data.objects['0'].__class__) assert v in str(observed_data.objects['0'].__class__)
def test_register_observable_extension_with_default_version(): def test_register_observable_extension_with_default_version():
@ -147,10 +147,10 @@ def test_register_observable_extension_with_default_version():
}, },
) )
core._register_observable_extension(observed_data.objects['0'], observed_data.objects['0'].extensions['ntfs-ext'].__class__) core._register_observable_extension(observed_data.objects['0'], observed_data.objects['0'].extensions['ntfs-ext'].__class__)
v = 'v' + DEFAULT_VERSION.replace('.', '') v = 'v21'
assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables'] assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables']
assert 'v' + DEFAULT_VERSION.replace('.', '') in str(observed_data.objects['0'].__class__) assert v in str(observed_data.objects['0'].__class__)
assert observed_data.objects['0'].extensions['ntfs-ext']._type in core.STIX2_OBJ_MAPS[v]['observable-extensions']['file'] assert observed_data.objects['0'].extensions['ntfs-ext']._type in core.STIX2_OBJ_MAPS[v]['observable-extensions']['file']
assert 'v' + DEFAULT_VERSION.replace('.', '') in str(observed_data.objects['0'].extensions['ntfs-ext'].__class__) assert v in str(observed_data.objects['0'].extensions['ntfs-ext'].__class__)

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

@ -734,7 +734,7 @@ def test_file_example_with_empty_NTFSExt():
}, },
) )
assert excinfo.value.cls == stix2.NTFSExt assert excinfo.value.cls == stix2.v21.NTFSExt
assert excinfo.value.properties == sorted(list(stix2.NTFSExt._properties.keys())) assert excinfo.value.properties == sorted(list(stix2.NTFSExt._properties.keys()))

12
stix2/test/v21/test_workbench.py
View File

@ -1,5 +1,7 @@
import os import os
import pytest
import stix2 import stix2
from stix2.workbench import ( from stix2.workbench import (
AttackPattern, Campaign, CourseOfAction, ExternalReference, AttackPattern, Campaign, CourseOfAction, ExternalReference,
@ -22,6 +24,7 @@ from .constants import (
) )
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_environment(): def test_workbench_environment():
# Create a STIX object # Create a STIX object
@ -76,6 +79,7 @@ def test_workbench_get_all_identities():
assert resp[0].id == IDENTITY_ID assert resp[0].id == IDENTITY_ID
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_get_all_indicators(): def test_workbench_get_all_indicators():
resp = indicators() resp = indicators()
assert len(resp) == 1 assert len(resp) == 1
@ -91,6 +95,7 @@ def test_workbench_get_all_intrusion_sets():
assert resp[0].id == INTRUSION_SET_ID assert resp[0].id == INTRUSION_SET_ID
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_get_all_malware(): def test_workbench_get_all_malware():
mal = Malware(id=MALWARE_ID, **MALWARE_KWARGS) mal = Malware(id=MALWARE_ID, **MALWARE_KWARGS)
save(mal) save(mal)
@ -109,6 +114,7 @@ def test_workbench_get_all_observed_data():
assert resp[0].id == OBSERVED_DATA_ID assert resp[0].id == OBSERVED_DATA_ID
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_get_all_reports(): def test_workbench_get_all_reports():
rep = Report(id=REPORT_ID, **REPORT_KWARGS) rep = Report(id=REPORT_ID, **REPORT_KWARGS)
save(rep) save(rep)
@ -118,6 +124,7 @@ def test_workbench_get_all_reports():
assert resp[0].id == REPORT_ID assert resp[0].id == REPORT_ID
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_get_all_threat_actors(): def test_workbench_get_all_threat_actors():
thr = ThreatActor(id=THREAT_ACTOR_ID, **THREAT_ACTOR_KWARGS) thr = ThreatActor(id=THREAT_ACTOR_ID, **THREAT_ACTOR_KWARGS)
save(thr) save(thr)
@ -127,6 +134,7 @@ def test_workbench_get_all_threat_actors():
assert resp[0].id == THREAT_ACTOR_ID assert resp[0].id == THREAT_ACTOR_ID
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_get_all_tools(): def test_workbench_get_all_tools():
tool = Tool(id=TOOL_ID, **TOOL_KWARGS) tool = Tool(id=TOOL_ID, **TOOL_KWARGS)
save(tool) save(tool)
@ -151,6 +159,7 @@ def test_workbench_add_to_bundle():
assert bundle.objects[0].name == 'Heartbleed' assert bundle.objects[0].name == 'Heartbleed'
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_relationships(): def test_workbench_relationships():
rel = Relationship(INDICATOR_ID, 'indicates', MALWARE_ID) rel = Relationship(INDICATOR_ID, 'indicates', MALWARE_ID)
save(rel) save(rel)
@ -170,6 +179,7 @@ def test_workbench_created_by():
assert creator.id == IDENTITY_ID assert creator.id == IDENTITY_ID
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_related(): def test_workbench_related():
rel1 = Relationship(MALWARE_ID, 'targets', IDENTITY_ID) rel1 = Relationship(MALWARE_ID, 'targets', IDENTITY_ID)
rel2 = Relationship(CAMPAIGN_ID, 'uses', MALWARE_ID) rel2 = Relationship(CAMPAIGN_ID, 'uses', MALWARE_ID)
@ -185,6 +195,7 @@ def test_workbench_related():
assert len(resp) == 1 assert len(resp) == 1
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_workbench_related_with_filters(): def test_workbench_related_with_filters():
malware = Malware( malware = Malware(
malware_types=["ransomware"], name="CryptorBit", malware_types=["ransomware"], name="CryptorBit",
@ -205,6 +216,7 @@ def test_workbench_related_with_filters():
assert len(resp) == 1 assert len(resp) == 1
@pytest.mark.xfail(reason='The workbench is not working correctly for 2.1')
def test_add_data_source(): def test_add_data_source():
fs_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stix2_data") fs_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stix2_data")
fs = FileSystemSource(fs_path) fs = FileSystemSource(fs_path)