Merge pull request #376 from khdesai/fix_issue_363

Fix existing tests and add new tests. Fixes #363

.travis.yml

@@ -1,4 +1,4 @@
-sudo: false
+os: linux
 language: python
 cache: pip
 dist: xenial

stix2/core.py

@@ -8,7 +8,7 @@ import re
 import stix2
 
 from .base import _Observable, _STIXBase
-from .exceptions import ParseError
+from .exceptions import DuplicateRegistrationError, ParseError
 from .markings import _MarkingsMixin
 from .utils import SCO21_EXT_REGEX, TYPE_REGEX, _get_dict
 
@@ -217,6 +217,8 @@ def _register_object(new_type, version=None):
         v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
 
     OBJ_MAP = STIX2_OBJ_MAPS[v]['objects']
+    if new_type._type in OBJ_MAP.keys():
+        raise DuplicateRegistrationError("STIX Object", new_type._type)
     OBJ_MAP[new_type._type] = new_type
 
 
@@ -236,6 +238,8 @@ def _register_marking(new_marking, version=None):
         v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
 
     OBJ_MAP_MARKING = STIX2_OBJ_MAPS[v]['markings']
+    if new_marking._type in OBJ_MAP_MARKING.keys():
+        raise DuplicateRegistrationError("STIX Marking", new_marking._type)
     OBJ_MAP_MARKING[new_marking._type] = new_marking
 
 
@@ -255,6 +259,8 @@ def _register_observable(new_observable, version=None):
         v = 'v' + stix2.DEFAULT_VERSION.replace('.', '')
 
     OBJ_MAP_OBSERVABLE = STIX2_OBJ_MAPS[v]['observables']
+    if new_observable._type in OBJ_MAP_OBSERVABLE.keys():
+        raise DuplicateRegistrationError("Cyber Observable", new_observable._type)
     OBJ_MAP_OBSERVABLE[new_observable._type] = new_observable
 
 
@@ -319,6 +325,8 @@ def _register_observable_extension(
     EXT_MAP = STIX2_OBJ_MAPS[v]['observable-extensions']
 
     try:
+        if ext_type in EXT_MAP[observable_type].keys():
+            raise DuplicateRegistrationError("Observable Extension", ext_type)
         EXT_MAP[observable_type][ext_type] = new_extension
     except KeyError:
         if observable_type not in OBJ_MAP_OBSERVABLE:

stix2/exceptions.py

@@ -233,3 +233,16 @@ class STIXDeprecationWarning(DeprecationWarning):
     Represents usage of a deprecated component of a STIX specification.
     """
     pass
+
+
+class DuplicateRegistrationError(STIXError):
+    """A STIX object with the same type as an existing object is being registered"""
+
+    def __init__(self, obj_type, reg_obj_type):
+        super(DuplicateRegistrationError, self).__init__()
+        self.obj_type = obj_type
+        self.reg_obj_type = reg_obj_type
+
+    def __str__(self):
+        msg = "A(n) {0} with type '{1}' already exists and cannot be registered again"
+        return msg.format(self.obj_type, self.reg_obj_type)

stix2/test/v20/test_core.py

@@ -3,8 +3,6 @@ import pytest
 import stix2
 from stix2 import core, exceptions
 
-from .constants import IDENTITY_ID
-
 BUNDLE = {
     "type": "bundle",
     "spec_version": "2.0",
@@ -68,109 +66,9 @@ def test_parse_observable_with_no_version():
     assert v in str(obs_obj.__class__)
 
 
-def test_register_object_with_version():
-    bundle = core.dict_to_stix2(BUNDLE, version='2.0')
-    core._register_object(bundle.objects[0].__class__, version='2.0')
-    v = 'v20'
-
-    assert bundle.objects[0].type in core.STIX2_OBJ_MAPS[v]['objects']
-    # spec_version is not in STIX 2.0, and is required in 2.1, so this
-    # suffices as a test for a STIX 2.0 object.
-    assert "spec_version" not in bundle.objects[0]
-
-
 def test_register_marking_with_version():
     core._register_marking(stix2.v20.TLP_WHITE.__class__, version='2.0')
     v = 'v20'
 
     assert stix2.v20.TLP_WHITE.definition._type in core.STIX2_OBJ_MAPS[v]['markings']
     assert v in str(stix2.v20.TLP_WHITE.__class__)
-
-
-@pytest.mark.xfail(reason="The default version is no longer 2.0", condition=stix2.DEFAULT_VERSION != "2.0")
-def test_register_marking_with_no_version():
-    # Uses default version (2.0 in this case)
-    core._register_marking(stix2.v20.TLP_WHITE.__class__)
-    v = 'v20'
-
-    assert stix2.v20.TLP_WHITE.definition._type in core.STIX2_OBJ_MAPS[v]['markings']
-    assert v in str(stix2.v20.TLP_WHITE.__class__)
-
-
-def test_register_observable_with_version():
-    observed_data = stix2.v20.ObservedData(
-        id="observed-data--b67d30ff-02ac-498a-92f9-32f845f448cf",
-        created_by_ref=IDENTITY_ID,
-        created="2016-04-06T19:58:16.000Z",
-        modified="2016-04-06T19:58:16.000Z",
-        first_observed="2015-12-21T19:00:00Z",
-        last_observed="2015-12-21T19:00:00Z",
-        number_observed=50,
-        objects={
-            "0": {
-                "name": "foo.exe",
-                "type": "file",
-                "extensions": {
-                    "ntfs-ext": {
-                        "alternate_data_streams": [
-                            {
-                                "name": "second.stream",
-                                "size": 25536,
-                            },
-                        ],
-                    },
-                },
-            },
-            "1": {
-                "type": "directory",
-                "path": "/usr/home",
-                "contains_refs": ["0"],
-            },
-        },
-    )
-    core._register_observable(observed_data.objects['0'].__class__, version='2.0')
-    v = 'v20'
-
-    assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables']
-    assert v in str(observed_data.objects['0'].__class__)
-
-
-def test_register_observable_extension_with_version():
-    observed_data = stix2.v20.ObservedData(
-        id="observed-data--b67d30ff-02ac-498a-92f9-32f845f448cf",
-        created_by_ref=IDENTITY_ID,
-        created="2016-04-06T19:58:16.000Z",
-        modified="2016-04-06T19:58:16.000Z",
-        first_observed="2015-12-21T19:00:00Z",
-        last_observed="2015-12-21T19:00:00Z",
-        number_observed=50,
-        objects={
-            "0": {
-                "name": "foo.exe",
-                "type": "file",
-                "extensions": {
-                    "ntfs-ext": {
-                        "alternate_data_streams": [
-                            {
-                                "name": "second.stream",
-                                "size": 25536,
-                            },
-                        ],
-                    },
-                },
-            },
-            "1": {
-                "type": "directory",
-                "path": "/usr/home",
-                "contains_refs": ["0"],
-            },
-        },
-    )
-    core._register_observable_extension(observed_data.objects['0'], observed_data.objects['0'].extensions['ntfs-ext'].__class__, version='2.0')
-    v = 'v20'
-
-    assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables']
-    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 v in str(observed_data.objects['0'].extensions['ntfs-ext'].__class__)

stix2/test/v20/test_custom.py

@@ -1,9 +1,10 @@
 import pytest
 
 import stix2
+from stix2 import core
 import stix2.v20
 
-from ...exceptions import InvalidValueError
+from ...exceptions import DuplicateRegistrationError, InvalidValueError
 from .constants import FAKE_TIME, IDENTITY_ID, MARKING_DEFINITION_ID
 
 IDENTITY_CUSTOM_PROP = stix2.v20.Identity(
@@ -449,7 +450,7 @@ def test_custom_observable_raises_exception():
 
 def test_custom_observable_object_no_init_1():
     @stix2.v20.CustomObservable(
-        'x-new-observable', [
+        'x-new-observable-1', [
             ('property1', stix2.properties.StringProperty()),
         ],
     )
@@ -1011,3 +1012,123 @@ def test_custom_object_nested_dictionary(data):
     )
 
     assert data == str(example)
+
+
+@stix2.v20.CustomObject(
+    'x-new-type-2', [
+        ('property1', stix2.properties.StringProperty()),
+        ('property2', stix2.properties.IntegerProperty()),
+    ],
+)
+class NewType2(object):
+    pass
+
+
+def test_register_custom_object_with_version():
+    custom_obj_1 = {
+        "type": "x-new-type-2",
+        "id": "x-new-type-2--00000000-0000-4000-8000-000000000007",
+    }
+
+    cust_obj_1 = core.dict_to_stix2(custom_obj_1, version='2.0')
+    v = 'v20'
+
+    assert cust_obj_1.type in core.STIX2_OBJ_MAPS[v]['objects']
+    # spec_version is not in STIX 2.0, and is required in 2.1, so this
+    # suffices as a test for a STIX 2.0 object.
+    assert "spec_version" not in cust_obj_1
+
+
+def test_register_duplicate_object_with_version():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v20.CustomObject(
+            'x-new-type-2', [
+                ('property1', stix2.properties.StringProperty()),
+                ('property2', stix2.properties.IntegerProperty()),
+            ],
+        )
+        class NewType2(object):
+            pass
+    assert "cannot be registered again" in str(excinfo.value)
+
+
+@stix2.v20.CustomObservable(
+    'x-new-observable-2', [
+        ('property1', stix2.properties.StringProperty()),
+    ],
+)
+class NewObservable2(object):
+    pass
+
+
+def test_register_observable_with_version():
+    custom_obs = NewObservable2(property1="Test Observable")
+    v = 'v20'
+
+    assert custom_obs.type in core.STIX2_OBJ_MAPS[v]['observables']
+
+
+def test_register_duplicate_observable_with_version():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v20.CustomObservable(
+            'x-new-observable-2', [
+                ('property1', stix2.properties.StringProperty()),
+            ],
+        )
+        class NewObservable2(object):
+            pass
+    assert "cannot be registered again" in str(excinfo.value)
+
+
+def test_register_marking_with_version():
+    @stix2.v20.CustomMarking(
+        'x-new-obj-2', [
+            ('property1', stix2.properties.StringProperty(required=True)),
+        ],
+    )
+    class NewObj2():
+        pass
+    v = 'v20'
+
+    no = NewObj2(property1='something')
+    assert no._type in core.STIX2_OBJ_MAPS[v]['markings']
+
+
+def test_register_observable_extension_with_version():
+    @stix2.v20.CustomExtension(
+        stix2.v20.UserAccount, 'some-extension-2', [
+            ('keys', stix2.properties.StringProperty(required=True)),
+        ],
+    )
+    class SomeCustomExtension2:
+        pass
+
+    v = 'v20'
+    example = SomeCustomExtension2(keys='test123')
+
+    assert example._type in core.STIX2_OBJ_MAPS[v]['observable-extensions']['user-account']
+
+
+def test_register_duplicate_observable_extension():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v20.CustomExtension(
+            stix2.v20.UserAccount, 'some-extension-2', [
+                ('property1', stix2.properties.StringProperty(required=True)),
+                ('property2', stix2.properties.IntegerProperty()),
+            ],
+        )
+        class NewExtension2():
+            pass
+    assert "cannot be registered again" in str(excinfo.value)
+
+
+def test_register_duplicate_marking():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v20.CustomMarking(
+            'x-new-obj-2', [
+                ('property1', stix2.properties.StringProperty(required=True)),
+            ],
+        )
+        class NewObj2():
+            pass
+    assert "cannot be registered again" in str(excinfo.value)

stix2/test/v20/test_datastore_filesystem.py

@@ -635,7 +635,7 @@ def test_filesystem_object_with_custom_property_in_bundle(fs_store):
 
 def test_filesystem_custom_object(fs_store):
     @stix2.v20.CustomObject(
-        'x-new-obj', [
+        'x-new-obj-2', [
             ('property1', stix2.properties.StringProperty(required=True)),
         ],
     )
@@ -650,7 +650,7 @@ def test_filesystem_custom_object(fs_store):
     assert newobj_r["property1"] == 'something'
 
     # remove dir
-    shutil.rmtree(os.path.join(FS_PATH, "x-new-obj"), True)
+    shutil.rmtree(os.path.join(FS_PATH, "x-new-obj-2"), True)
 
 
 def test_relationships(rel_fs_store):

stix2/test/v20/test_datastore_memory.py

@@ -329,7 +329,7 @@ def test_memory_store_object_with_custom_property_in_bundle(mem_store):
 
 def test_memory_store_custom_object(mem_store):
     @CustomObject(
-        'x-new-obj', [
+        'x-new-obj-3', [
             ('property1', properties.StringProperty(required=True)),
         ],
     )

stix2/test/v20/test_properties.py

@@ -392,7 +392,7 @@ def test_dictionary_property_invalid(d):
 
 def test_property_list_of_dictionary():
     @stix2.v20.CustomObject(
-        'x-new-obj', [
+        'x-new-obj-4', [
             ('property1', ListProperty(DictionaryProperty(spec_version="2.0"), required=True)),
         ],
     )

stix2/test/v21/test_core.py

@@ -3,8 +3,6 @@ import pytest
 import stix2
 from stix2 import core, exceptions
 
-from .constants import IDENTITY_ID, OBSERVED_DATA_ID
-
 BUNDLE = {
     "type": "bundle",
     "id": "bundle--00000000-0000-4000-8000-000000000007",
@@ -73,15 +71,6 @@ def test_parse_observable_with_no_version():
     assert v in str(obs_obj.__class__)
 
 
-def test_register_object_with_version():
-    bundle = core.dict_to_stix2(BUNDLE, version='2.1')
-    core._register_object(bundle.objects[0].__class__)
-    v = 'v21'
-
-    assert bundle.objects[0].type in core.STIX2_OBJ_MAPS[v]['objects']
-    assert bundle.objects[0].spec_version == "2.1"
-
-
 def test_register_marking_with_version():
     core._register_marking(stix2.v21.TLP_WHITE.__class__, version='2.1')
     v = 'v21'
@@ -98,82 +87,3 @@ def test_register_marking_with_no_version():
 
     assert stix2.v21.TLP_WHITE.definition._type in core.STIX2_OBJ_MAPS[v]['markings']
     assert v in str(stix2.v21.TLP_WHITE.__class__)
-
-
-def test_register_observable_with_default_version():
-    observed_data = stix2.v21.ObservedData(
-        id=OBSERVED_DATA_ID,
-        created_by_ref=IDENTITY_ID,
-        created="2016-04-06T19:58:16.000Z",
-        modified="2016-04-06T19:58:16.000Z",
-        first_observed="2015-12-21T19:00:00Z",
-        last_observed="2015-12-21T19:00:00Z",
-        number_observed=50,
-        objects={
-            "0": {
-                "name": "foo.exe",
-                "type": "file",
-                "extensions": {
-                    "ntfs-ext": {
-                        "alternate_data_streams": [
-                            {
-                                "name": "second.stream",
-                                "size": 25536,
-                            },
-                        ],
-                    },
-                },
-            },
-            "1": {
-                "type": "directory",
-                "path": "/usr/home",
-                "contains_refs": ["file--420bc087-8b53-5ae9-8210-20d27d5e96c8"],
-            },
-        },
-    )
-    core._register_observable(observed_data.objects['0'].__class__)
-    v = 'v21'
-
-    assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables']
-    assert v in str(observed_data.objects['0'].__class__)
-
-
-def test_register_observable_extension_with_default_version():
-    observed_data = stix2.v21.ObservedData(
-        id=OBSERVED_DATA_ID,
-        created_by_ref=IDENTITY_ID,
-        created="2016-04-06T19:58:16.000Z",
-        modified="2016-04-06T19:58:16.000Z",
-        first_observed="2015-12-21T19:00:00Z",
-        last_observed="2015-12-21T19:00:00Z",
-        number_observed=50,
-        objects={
-            "0": {
-                "name": "foo.exe",
-                "type": "file",
-                "extensions": {
-                    "ntfs-ext": {
-                        "alternate_data_streams": [
-                            {
-                                "name": "second.stream",
-                                "size": 25536,
-                            },
-                        ],
-                    },
-                },
-            },
-            "1": {
-                "type": "directory",
-                "path": "/usr/home",
-                "contains_refs": ["file--420bc087-8b53-5ae9-8210-20d27d5e96c8"],
-            },
-        },
-    )
-    core._register_observable_extension(observed_data.objects['0'], observed_data.objects['0'].extensions['ntfs-ext'].__class__)
-    v = 'v21'
-
-    assert observed_data.objects['0'].type in core.STIX2_OBJ_MAPS[v]['observables']
-    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 v in str(observed_data.objects['0'].extensions['ntfs-ext'].__class__)

stix2/test/v21/test_custom.py

@@ -6,7 +6,7 @@ import stix2
 import stix2.base
 import stix2.v21
 
-from ...exceptions import InvalidValueError
+from ...exceptions import DuplicateRegistrationError, InvalidValueError
 from .constants import FAKE_TIME, IDENTITY_ID, MARKING_DEFINITION_ID
 
 IDENTITY_CUSTOM_PROP = stix2.v21.Identity(
@@ -455,7 +455,7 @@ def test_custom_observable_raises_exception():
 
 def test_custom_observable_object_no_init_1():
     @stix2.v21.CustomObservable(
-        'x-new-observable', [
+        'x-new-observable-2', [
             ('property1', stix2.properties.StringProperty()),
         ],
     )
@@ -1065,3 +1065,110 @@ def test_custom_object_nested_dictionary(data):
     )
 
     assert data == str(example)
+
+
+@stix2.v21.CustomObject(
+    'x-new-type-2', [
+        ('property1', stix2.properties.StringProperty()),
+        ('property2', stix2.properties.IntegerProperty()),
+    ],
+)
+class NewType3(object):
+    pass
+
+
+def test_register_custom_object_with_version():
+    custom_obj_1 = {
+        "type": "x-new-type-2",
+        "id": "x-new-type-2--00000000-0000-4000-8000-000000000007",
+        "spec_version": "2.1",
+    }
+
+    cust_obj_1 = stix2.core.dict_to_stix2(custom_obj_1, version='2.1')
+    v = 'v21'
+
+    assert cust_obj_1.type in stix2.core.STIX2_OBJ_MAPS[v]['objects']
+    assert cust_obj_1.spec_version == "2.1"
+
+
+def test_register_duplicate_object_with_version():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v21.CustomObject(
+            'x-new-type-2', [
+                ('property1', stix2.properties.StringProperty()),
+                ('property2', stix2.properties.IntegerProperty()),
+            ],
+        )
+        class NewType2(object):
+            pass
+    assert "cannot be registered again" in str(excinfo.value)
+
+
+@stix2.v21.CustomObservable(
+    'x-new-observable-3', [
+        ('property1', stix2.properties.StringProperty()),
+    ],
+)
+class NewObservable3(object):
+    pass
+
+
+def test_register_observable():
+    custom_obs = NewObservable3(property1="Test Observable")
+    v = 'v21'
+
+    assert custom_obs.type in stix2.core.STIX2_OBJ_MAPS[v]['observables']
+
+
+def test_register_duplicate_observable():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v21.CustomObservable(
+            'x-new-observable-2', [
+                ('property1', stix2.properties.StringProperty()),
+            ],
+        )
+        class NewObservable2(object):
+            pass
+    assert "cannot be registered again" in str(excinfo.value)
+
+
+def test_register_observable_custom_extension():
+    @stix2.v21.CustomExtension(
+        stix2.v21.DomainName, 'x-new-2-ext', [
+            ('property1', stix2.properties.StringProperty(required=True)),
+            ('property2', stix2.properties.IntegerProperty()),
+        ],
+    )
+    class NewExtension2():
+        pass
+
+    example = NewExtension2(property1="Hi there")
+    v = 'v21'
+
+    assert 'domain-name' in stix2.core.STIX2_OBJ_MAPS[v]['observables']
+    assert example._type in stix2.core.STIX2_OBJ_MAPS[v]['observable-extensions']['domain-name']
+
+
+def test_register_duplicate_observable_extension():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v21.CustomExtension(
+            stix2.v21.DomainName, 'x-new-2-ext', [
+                ('property1', stix2.properties.StringProperty(required=True)),
+                ('property2', stix2.properties.IntegerProperty()),
+            ],
+        )
+        class NewExtension2():
+            pass
+    assert "cannot be registered again" in str(excinfo.value)
+
+
+def test_register_duplicate_marking():
+    with pytest.raises(DuplicateRegistrationError) as excinfo:
+        @stix2.v21.CustomMarking(
+            'x-new-obj', [
+                ('property1', stix2.properties.StringProperty(required=True)),
+            ],
+        )
+        class NewObj2():
+            pass
+    assert "cannot be registered again" in str(excinfo.value)

stix2/test/v21/test_datastore_filesystem.py

@@ -656,7 +656,7 @@ def test_filesystem_object_with_custom_property_in_bundle(fs_store):
 
 def test_filesystem_custom_object(fs_store):
     @stix2.v21.CustomObject(
-        'x-new-obj', [
+        'x-new-obj-2', [
             ('property1', stix2.properties.StringProperty(required=True)),
         ],
     )
@@ -671,7 +671,7 @@ def test_filesystem_custom_object(fs_store):
     assert newobj_r["property1"] == 'something'
 
     # remove dir
-    shutil.rmtree(os.path.join(FS_PATH, "x-new-obj"), True)
+    shutil.rmtree(os.path.join(FS_PATH, "x-new-obj-2"), True)
 
 
 def test_relationships(rel_fs_store):

stix2/test/v21/test_datastore_memory.py

@@ -344,7 +344,7 @@ def test_memory_store_object_with_custom_property_in_bundle(mem_store):
 
 def test_memory_store_custom_object(mem_store):
     @CustomObject(
-        'x-new-obj', [
+        'x-new-obj-3', [
             ('property1', properties.StringProperty(required=True)),
         ],
     )

stix2/test/v21/test_properties.py

@@ -404,7 +404,7 @@ def test_dictionary_property_invalid(d):
 
 def test_property_list_of_dictionary():
     @stix2.v21.CustomObject(
-        'x-new-obj', [
+        'x-new-obj-4', [
             ('property1', ListProperty(DictionaryProperty(spec_version='2.1'), required=True)),
         ],
     )

tox.ini

@@ -8,11 +8,11 @@ deps =
   pytest
   pytest-cov
   coverage
-  taxii2-client
+  taxii2-client<1.0.0
   fuzzywuzzy
   haversine
   python-Levenshtein
-  medallion
+  medallion<2.0.0
 commands =
   python -m pytest --cov=stix2 stix2/test/ --cov-report term-missing -W ignore::stix2.exceptions.STIXDeprecationWarning