diff --git a/stix2/__init__.py b/stix2/__init__.py index f3e02b43..26eed6fc 100644 --- a/stix2/__init__.py +++ b/stix2/__init__.py @@ -53,8 +53,8 @@ RepeatQualifier, StartStopQualifier, StringConstant, TimestampConstant, WithinQualifier, ) -from .utils import new_version, revoke from .v20 import * # This import will always be the latest STIX 2.X version from .version import __version__ +from .versioning import new_version, revoke _collect_stix2_mappings() diff --git a/stix2/base.py b/stix2/base.py index ef3fcb80..3da57968 100644 --- a/stix2/base.py +++ b/stix2/base.py @@ -21,8 +21,8 @@ from .utils import ( NOW, PREFIX_21_REGEX, find_property_index, format_datetime, get_timestamp, ) -from .utils import new_version as _new_version -from .utils import revoke as _revoke +from .versioning import new_version as _new_version +from .versioning import revoke as _revoke try: from collections.abc import Mapping diff --git a/stix2/markings/granular_markings.py b/stix2/markings/granular_markings.py index 5456f831..5021ec80 100644 --- a/stix2/markings/granular_markings.py +++ b/stix2/markings/granular_markings.py @@ -2,7 +2,8 @@ from stix2 import exceptions from stix2.markings import utils -from stix2.utils import is_marking, new_version +from stix2.utils import is_marking +from stix2.versioning import new_version def get_markings(obj, selectors, inherited=False, descendants=False, marking_ref=True, lang=True): diff --git a/stix2/markings/object_markings.py b/stix2/markings/object_markings.py index dc85dfa0..5d4e4f10 100644 --- a/stix2/markings/object_markings.py +++ b/stix2/markings/object_markings.py @@ -2,7 +2,7 @@ from stix2 import exceptions from stix2.markings import utils -from stix2.utils import new_version +from stix2.versioning import new_version def get_markings(obj): diff --git a/stix2/properties.py b/stix2/properties.py index c876c11d..5c6333f4 100644 --- a/stix2/properties.py +++ b/stix2/properties.py @@ -17,16 +17,15 @@ MutuallyExclusivePropertiesError, ) from .parsing import STIX2_OBJ_MAPS, parse, parse_observable -from .utils import ( - TYPE_21_REGEX, TYPE_REGEX, _get_dict, get_class_hierarchy_names, - parse_into_datetime, -) +from .utils import _get_dict, get_class_hierarchy_names, parse_into_datetime try: from collections.abc import Mapping except ImportError: from collections import Mapping +TYPE_REGEX = re.compile(r'^\-?[a-z0-9]+(-[a-z0-9]+)*\-?$') +TYPE_21_REGEX = re.compile(r'^([a-z][a-z0-9]*)+(-[a-z0-9]+)*\-?$') ERROR_INVALID_ID = ( "not a valid STIX identifier, must match --: {}" ) diff --git a/stix2/test/v20/test_object_markings.py b/stix2/test/v20/test_object_markings.py index 191f33a9..6bd2269d 100644 --- a/stix2/test/v20/test_object_markings.py +++ b/stix2/test/v20/test_object_markings.py @@ -14,6 +14,7 @@ MALWARE_KWARGS = MALWARE_KWARGS_CONST.copy() MALWARE_KWARGS.update({ 'id': MALWARE_ID, + 'type': 'malware', 'created': FAKE_TIME, 'modified': FAKE_TIME, }) diff --git a/stix2/test/v20/test_versioning.py b/stix2/test/v20/test_versioning.py index 9974e429..03d43ccb 100644 --- a/stix2/test/v20/test_versioning.py +++ b/stix2/test/v20/test_versioning.py @@ -1,6 +1,10 @@ import pytest import stix2 +import stix2.exceptions +import stix2.utils +import stix2.v20 +import stix2.versioning from .constants import CAMPAIGN_MORE_KWARGS @@ -142,7 +146,7 @@ def test_versioning_error_revoke_of_revoked(): def test_making_new_version_dict(): campaign_v1 = CAMPAIGN_MORE_KWARGS - campaign_v2 = stix2.utils.new_version(CAMPAIGN_MORE_KWARGS, name="fred") + campaign_v2 = stix2.versioning.new_version(CAMPAIGN_MORE_KWARGS, name="fred") assert campaign_v1['id'] == campaign_v2['id'] assert campaign_v1['created_by_ref'] == campaign_v2['created_by_ref'] @@ -155,7 +159,7 @@ def test_making_new_version_dict(): def test_versioning_error_dict_bad_modified_value(): with pytest.raises(stix2.exceptions.InvalidValueError) as excinfo: - stix2.utils.new_version(CAMPAIGN_MORE_KWARGS, modified="2015-04-06T20:03:00.000Z") + stix2.versioning.new_version(CAMPAIGN_MORE_KWARGS, modified="2015-04-06T20:03:00.000Z") assert excinfo.value.cls == dict assert excinfo.value.prop_name == "modified" @@ -171,7 +175,7 @@ def test_versioning_error_dict_no_modified_value(): 'created': "2016-04-06T20:03:00.000Z", 'name': "Green Group Attacks Against Finance", } - campaign_v2 = stix2.utils.new_version(campaign_v1, modified="2017-04-06T20:03:00.000Z") + campaign_v2 = stix2.versioning.new_version(campaign_v1, modified="2017-04-06T20:03:00.000Z") assert str(campaign_v2['modified']) == "2017-04-06T20:03:00.000Z" @@ -179,14 +183,14 @@ def test_versioning_error_dict_no_modified_value(): def test_making_new_version_invalid_cls(): campaign_v1 = "This is a campaign." with pytest.raises(ValueError) as excinfo: - stix2.utils.new_version(campaign_v1, name="fred") + stix2.versioning.new_version(campaign_v1, name="fred") assert 'cannot create new version of object of this type' in str(excinfo.value) def test_revoke_dict(): campaign_v1 = CAMPAIGN_MORE_KWARGS - campaign_v2 = stix2.utils.revoke(campaign_v1) + campaign_v2 = stix2.versioning.revoke(campaign_v1) assert campaign_v1['id'] == campaign_v2['id'] assert campaign_v1['created_by_ref'] == campaign_v2['created_by_ref'] @@ -198,12 +202,18 @@ def test_revoke_dict(): assert campaign_v2['revoked'] +def test_revoke_unversionable(): + sco = stix2.v20.File(name="data.txt") + with pytest.raises(ValueError): + sco.revoke() + + def test_versioning_error_revoke_of_revoked_dict(): campaign_v1 = CAMPAIGN_MORE_KWARGS - campaign_v2 = stix2.utils.revoke(campaign_v1) + campaign_v2 = stix2.versioning.revoke(campaign_v1) with pytest.raises(stix2.exceptions.RevokeError) as excinfo: - stix2.utils.revoke(campaign_v2) + stix2.versioning.revoke(campaign_v2) assert excinfo.value.called_by == "revoke" @@ -211,7 +221,7 @@ def test_versioning_error_revoke_of_revoked_dict(): def test_revoke_invalid_cls(): campaign_v1 = "This is a campaign." with pytest.raises(ValueError) as excinfo: - stix2.utils.revoke(campaign_v1) + stix2.versioning.revoke(campaign_v1) assert 'cannot revoke object of this type' in str(excinfo.value) @@ -224,7 +234,7 @@ def test_remove_custom_stix_property(): allow_custom=True, ) - mal_nc = stix2.utils.remove_custom_stix(mal) + mal_nc = stix2.versioning.remove_custom_stix(mal) assert "x_custom" not in mal_nc assert (stix2.utils.parse_into_datetime(mal["modified"], precision="millisecond") < @@ -243,15 +253,136 @@ class Animal(object): animal = Animal(species="lion", animal_class="mammal") - nc = stix2.utils.remove_custom_stix(animal) + nc = stix2.versioning.remove_custom_stix(animal) assert nc is None def test_remove_custom_stix_no_custom(): campaign_v1 = stix2.v20.Campaign(**CAMPAIGN_MORE_KWARGS) - campaign_v2 = stix2.utils.remove_custom_stix(campaign_v1) + campaign_v2 = stix2.versioning.remove_custom_stix(campaign_v1) assert len(campaign_v1.keys()) == len(campaign_v2.keys()) assert campaign_v1.id == campaign_v2.id assert campaign_v1.description == campaign_v2.description + + +def test_version_unversionable_dict(): + f = { + "type": "file", + "name": "data.txt", + } + + with pytest.raises(ValueError): + stix2.versioning.new_version(f) + + +def test_version_sco_with_modified(): + """ + Ensure new_version() doesn't get tripped up over unversionable objects with + properties not used for versioning, but whose names conflict with + versioning properties. + """ + + file_sco = { + "type": "file", + "name": "data.txt", + "created": "1973-11-23T02:31:37Z", + "modified": "1991-05-13T19:24:57Z", + } + + with pytest.raises(ValueError): + stix2.versioning.new_version(file_sco, name="newname.txt") + + with pytest.raises(ValueError): + stix2.versioning.revoke(file_sco) + + file_sco_obj = stix2.v20.File( + name="data.txt", + created="1973-11-23T02:31:37Z", + modified="1991-05-13T19:24:57Z", + ) + + with pytest.raises(ValueError): + stix2.versioning.new_version(file_sco_obj, name="newname.txt") + + with pytest.raises(ValueError): + stix2.versioning.revoke(file_sco_obj) + + +def test_version_sco_with_custom(): + """ + If we add custom properties named like versioning properties to an object + type which is otherwise unversionable, versioning should start working. + """ + + file_sco_obj = stix2.v20.File( + name="data.txt", + created="1973-11-23T02:31:37Z", + modified="1991-05-13T19:24:57Z", + revoked=False, # the custom property + allow_custom=True, + ) + + new_file_sco_obj = stix2.versioning.new_version( + file_sco_obj, name="newname.txt", + ) + + assert new_file_sco_obj.name == "newname.txt" + + revoked_obj = stix2.versioning.revoke(new_file_sco_obj) + assert revoked_obj.revoked + + +def test_version_disable_custom(): + m = stix2.v20.Malware( + name="foo", labels=["label"], description="Steals your identity!", + x_custom=123, allow_custom=True, + ) + + # Remove the custom property, and disallow custom properties in the + # resulting object. + m2 = stix2.versioning.new_version(m, x_custom=None, allow_custom=False) + assert "x_custom" not in m2 + + # Remove a regular property and leave the custom one, disallow custom + # properties, and make sure we get an error. + with pytest.raises(stix2.exceptions.ExtraPropertiesError): + stix2.versioning.new_version(m, description=None, allow_custom=False) + + +def test_version_enable_custom(): + m = stix2.v20.Malware( + name="foo", labels=["label"], description="Steals your identity!", + ) + + # Add a custom property to an object for which it was previously disallowed + m2 = stix2.versioning.new_version(m, x_custom=123, allow_custom=True) + assert "x_custom" in m2 + + # Add a custom property without enabling it, make sure we get an error + with pytest.raises(stix2.exceptions.ExtraPropertiesError): + stix2.versioning.new_version(m, x_custom=123, allow_custom=False) + + +def test_version_propagate_custom(): + m = stix2.v20.Malware( + name="foo", labels=["label"], + ) + + # Remember custom-not-allowed setting from original; produce error + with pytest.raises(stix2.exceptions.ExtraPropertiesError): + stix2.versioning.new_version(m, x_custom=123) + + m2 = stix2.versioning.new_version(m, description="Steals your identity!") + assert "description" in m2 + assert m2.description == "Steals your identity!" + + m_custom = stix2.v20.Malware( + name="foo", labels=["label"], x_custom=123, allow_custom=True, + ) + + # Remember custom-allowed setting from original; should work + m2_custom = stix2.versioning.new_version(m_custom, x_other_custom="abc") + assert "x_other_custom" in m2_custom + assert m2_custom.x_other_custom == "abc" diff --git a/stix2/test/v21/test_object_markings.py b/stix2/test/v21/test_object_markings.py index a21fbf6a..bb1c4ab0 100644 --- a/stix2/test/v21/test_object_markings.py +++ b/stix2/test/v21/test_object_markings.py @@ -13,6 +13,7 @@ MALWARE_KWARGS = MALWARE_KWARGS_CONST.copy() MALWARE_KWARGS.update({ 'id': MALWARE_ID, + 'type': 'malware', 'created': FAKE_TIME, 'modified': FAKE_TIME, }) diff --git a/stix2/test/v21/test_versioning.py b/stix2/test/v21/test_versioning.py index bee0c07f..adfa7a01 100644 --- a/stix2/test/v21/test_versioning.py +++ b/stix2/test/v21/test_versioning.py @@ -3,7 +3,10 @@ import pytest import stix2 +import stix2.exceptions import stix2.utils +import stix2.v21 +import stix2.versioning from .constants import CAMPAIGN_MORE_KWARGS @@ -151,7 +154,7 @@ def test_versioning_error_revoke_of_revoked(): def test_making_new_version_dict(): campaign_v1 = CAMPAIGN_MORE_KWARGS - campaign_v2 = stix2.utils.new_version(CAMPAIGN_MORE_KWARGS, name="fred") + campaign_v2 = stix2.versioning.new_version(CAMPAIGN_MORE_KWARGS, name="fred") assert campaign_v1['id'] == campaign_v2['id'] assert campaign_v1['spec_version'] == campaign_v2['spec_version'] @@ -165,7 +168,7 @@ def test_making_new_version_dict(): def test_versioning_error_dict_bad_modified_value(): with pytest.raises(stix2.exceptions.InvalidValueError) as excinfo: - stix2.utils.new_version(CAMPAIGN_MORE_KWARGS, modified="2015-04-06T20:03:00.000Z") + stix2.versioning.new_version(CAMPAIGN_MORE_KWARGS, modified="2015-04-06T20:03:00.000Z") assert excinfo.value.cls == dict assert excinfo.value.prop_name == "modified" @@ -181,7 +184,7 @@ def test_versioning_error_dict_no_modified_value(): 'created': "2016-04-06T20:03:00.000Z", 'name': "Green Group Attacks Against Finance", } - campaign_v2 = stix2.utils.new_version(campaign_v1, modified="2017-04-06T20:03:00.000Z") + campaign_v2 = stix2.versioning.new_version(campaign_v1, modified="2017-04-06T20:03:00.000Z") assert str(campaign_v2['modified']) == "2017-04-06T20:03:00.000Z" @@ -189,14 +192,14 @@ def test_versioning_error_dict_no_modified_value(): def test_making_new_version_invalid_cls(): campaign_v1 = "This is a campaign." with pytest.raises(ValueError) as excinfo: - stix2.utils.new_version(campaign_v1, name="fred") + stix2.versioning.new_version(campaign_v1, name="fred") assert 'cannot create new version of object of this type' in str(excinfo.value) def test_revoke_dict(): campaign_v1 = CAMPAIGN_MORE_KWARGS - campaign_v2 = stix2.utils.revoke(campaign_v1) + campaign_v2 = stix2.versioning.revoke(campaign_v1) assert campaign_v1['id'] == campaign_v2['id'] assert campaign_v1['spec_version'] == campaign_v2['spec_version'] @@ -209,12 +212,18 @@ def test_revoke_dict(): assert campaign_v2['revoked'] +def test_revoke_unversionable(): + sco = stix2.v21.File(name="data.txt") + with pytest.raises(ValueError): + sco.revoke() + + def test_versioning_error_revoke_of_revoked_dict(): campaign_v1 = CAMPAIGN_MORE_KWARGS - campaign_v2 = stix2.utils.revoke(campaign_v1) + campaign_v2 = stix2.versioning.revoke(campaign_v1) with pytest.raises(stix2.exceptions.RevokeError) as excinfo: - stix2.utils.revoke(campaign_v2) + stix2.versioning.revoke(campaign_v2) assert excinfo.value.called_by == "revoke" @@ -222,7 +231,7 @@ def test_versioning_error_revoke_of_revoked_dict(): def test_revoke_invalid_cls(): campaign_v1 = "This is a campaign." with pytest.raises(ValueError) as excinfo: - stix2.utils.revoke(campaign_v1) + stix2.versioning.revoke(campaign_v1) assert 'cannot revoke object of this type' in str(excinfo.value) @@ -236,7 +245,7 @@ def test_remove_custom_stix_property(): is_family=False, ) - mal_nc = stix2.utils.remove_custom_stix(mal) + mal_nc = stix2.versioning.remove_custom_stix(mal) assert "x_custom" not in mal_nc assert mal["modified"] < mal_nc["modified"] @@ -254,14 +263,14 @@ class Animal(object): animal = Animal(species="lion", animal_class="mammal") - nc = stix2.utils.remove_custom_stix(animal) + nc = stix2.versioning.remove_custom_stix(animal) assert nc is None def test_remove_custom_stix_no_custom(): campaign_v1 = stix2.v21.Campaign(**CAMPAIGN_MORE_KWARGS) - campaign_v2 = stix2.utils.remove_custom_stix(campaign_v1) + campaign_v2 = stix2.versioning.remove_custom_stix(campaign_v1) assert len(campaign_v1.keys()) == len(campaign_v2.keys()) assert campaign_v1.id == campaign_v2.id @@ -294,5 +303,96 @@ def test_fudge_modified(old, candidate_new, expected_new, use_stix21): expected_new, "%Y-%m-%dT%H:%M:%S.%fZ", ) - fudged = stix2.utils._fudge_modified(old_dt, candidate_new_dt, use_stix21) + fudged = stix2.versioning._fudge_modified( + old_dt, candidate_new_dt, use_stix21, + ) assert fudged == expected_new_dt + + +def test_version_unversionable_dict(): + f = { + "type": "file", + "id": "file--4efb5217-e987-4438-9a1b-c800099401df", + "name": "data.txt", + } + + with pytest.raises(ValueError): + stix2.versioning.new_version(f) + + +def test_version_sco_with_custom(): + """ + If we add custom properties named like versioning properties to an object + type which is otherwise unversionable, versioning should start working. + """ + + file_sco_obj = stix2.v21.File( + name="data.txt", + created="1973-11-23T02:31:37Z", + modified="1991-05-13T19:24:57Z", + revoked=False, + allow_custom=True, + ) + + new_file_sco_obj = stix2.versioning.new_version( + file_sco_obj, size=1234, + ) + + assert new_file_sco_obj.size == 1234 + + revoked_obj = stix2.versioning.revoke(new_file_sco_obj) + assert revoked_obj.revoked + + +def test_version_disable_custom(): + m = stix2.v21.Malware( + name="foo", description="Steals your identity!", is_family=False, + x_custom=123, allow_custom=True, + ) + + # Remove the custom property, and disallow custom properties in the + # resulting object. + m2 = stix2.versioning.new_version(m, x_custom=None, allow_custom=False) + assert "x_custom" not in m2 + + # Remove a regular property and leave the custom one, disallow custom + # properties, and make sure we get an error. + with pytest.raises(stix2.exceptions.ExtraPropertiesError): + stix2.versioning.new_version(m, description=None, allow_custom=False) + + +def test_version_enable_custom(): + m = stix2.v21.Malware( + name="foo", description="Steals your identity!", is_family=False, + ) + + # Add a custom property to an object for which it was previously disallowed + m2 = stix2.versioning.new_version(m, x_custom=123, allow_custom=True) + assert "x_custom" in m2 + + # Add a custom property without enabling it, make sure we get an error + with pytest.raises(stix2.exceptions.ExtraPropertiesError): + stix2.versioning.new_version(m, x_custom=123, allow_custom=False) + + +def test_version_propagate_custom(): + m = stix2.v21.Malware( + name="foo", is_family=False, + ) + + # Remember custom-not-allowed setting from original; produce error + with pytest.raises(stix2.exceptions.ExtraPropertiesError): + stix2.versioning.new_version(m, x_custom=123) + + m2 = stix2.versioning.new_version(m, description="Steals your identity!") + assert "description" in m2 + assert m2.description == "Steals your identity!" + + m_custom = stix2.v21.Malware( + name="foo", is_family=False, x_custom=123, allow_custom=True, + ) + + # Remember custom-allowed setting from original; should work + m2_custom = stix2.versioning.new_version(m_custom, x_other_custom="abc") + assert "x_other_custom" in m2_custom + assert m2_custom.x_other_custom == "abc" diff --git a/stix2/utils.py b/stix2/utils.py index 766fd4b7..7a8d8cb4 100644 --- a/stix2/utils.py +++ b/stix2/utils.py @@ -1,10 +1,5 @@ """Utility functions and classes for the STIX2 library.""" -try: - from collections.abc import Mapping -except ImportError: - from collections import Mapping -import copy import datetime as dt import enum import json @@ -15,20 +10,11 @@ import stix2 -from .exceptions import ( - InvalidValueError, RevokeError, UnmodifiablePropertyError, -) - # Sentinel value for properties that should be set to the current time. # We can't use the standard 'default' approach, since if there are multiple # timestamps in a single object, the timestamps will vary by a few microseconds. NOW = object() -# STIX object properties that cannot be modified -STIX_UNMOD_PROPERTIES = ['created', 'created_by_ref', 'id', 'type'] - -TYPE_REGEX = re.compile(r'^\-?[a-z0-9]+(-[a-z0-9]+)*\-?$') -TYPE_21_REGEX = re.compile(r'^([a-z][a-z0-9]*)+(-[a-z0-9]+)*\-?$') PREFIX_21_REGEX = re.compile(r'^[a-z].*') _TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ" @@ -389,121 +375,6 @@ def find_property_index(obj, search_key, search_value): return idx -def _fudge_modified(old_modified, new_modified, use_stix21): - """ - Ensures a new modified timestamp is newer than the old. When they are - too close together, new_modified must be pushed further ahead to ensure - it is distinct and later, after JSON serialization (which may mean it's - actually being pushed a little ways into the future). JSON serialization - can remove precision, which can cause distinct timestamps to accidentally - become equal, if we're not careful. - - :param old_modified: A previous "modified" timestamp, as a datetime object - :param new_modified: A candidate new "modified" timestamp, as a datetime - object - :param use_stix21: Whether to use STIX 2.1+ versioning timestamp precision - rules (boolean). This is important so that we are aware of how - timestamp precision will be truncated, so we know how close together - the timestamps can be, and how far ahead to potentially push the new - one. - :return: A suitable new "modified" timestamp. This may be different from - what was passed in, if it had to be pushed ahead. - """ - if use_stix21: - # 2.1+: we can use full precision - if new_modified <= old_modified: - new_modified = old_modified + dt.timedelta(microseconds=1) - else: - # 2.0: we must use millisecond precision - one_ms = dt.timedelta(milliseconds=1) - if new_modified - old_modified < one_ms: - new_modified = old_modified + one_ms - - return new_modified - - -def new_version(data, **kwargs): - """Create a new version of a STIX object, by modifying properties and - updating the ``modified`` property. - """ - - if not isinstance(data, Mapping): - raise ValueError( - "cannot create new version of object of this type! " - "Try a dictionary or instance of an SDO or SRO class.", - ) - - unchangable_properties = [] - if data.get('revoked'): - raise RevokeError("new_version") - try: - new_obj_inner = copy.deepcopy(data._inner) - except AttributeError: - new_obj_inner = copy.deepcopy(data) - properties_to_change = kwargs.keys() - - # Make sure certain properties aren't trying to change - for prop in STIX_UNMOD_PROPERTIES: - if prop in properties_to_change: - unchangable_properties.append(prop) - if unchangable_properties: - raise UnmodifiablePropertyError(unchangable_properties) - - # Different versioning precision rules in STIX 2.0 vs 2.1, so we need - # to know which rules to apply. - is_21 = "spec_version" in data - precision_constraint = "min" if is_21 else "exact" - - cls = type(data) - if 'modified' not in kwargs: - old_modified = parse_into_datetime( - data["modified"], precision="millisecond", - precision_constraint=precision_constraint, - ) - - new_modified = get_timestamp() - new_modified = _fudge_modified(old_modified, new_modified, is_21) - - kwargs['modified'] = new_modified - - elif 'modified' in data: - old_modified_property = parse_into_datetime( - data.get('modified'), precision='millisecond', - precision_constraint=precision_constraint, - ) - new_modified_property = parse_into_datetime( - kwargs['modified'], precision='millisecond', - precision_constraint=precision_constraint, - ) - if new_modified_property <= old_modified_property: - raise InvalidValueError( - cls, 'modified', - "The new modified datetime cannot be before than or equal to the current modified datetime." - "It cannot be equal, as according to STIX 2 specification, objects that are different " - "but have the same id and modified timestamp do not have defined consumer behavior.", - ) - new_obj_inner.update(kwargs) - # Exclude properties with a value of 'None' in case data is not an instance of a _STIXBase subclass - return cls(**{k: v for k, v in new_obj_inner.items() if v is not None}) - - -def revoke(data): - """Revoke a STIX object. - - Returns: - A new version of the object with ``revoked`` set to ``True``. - """ - if not isinstance(data, Mapping): - raise ValueError( - "cannot revoke object of this type! Try a dictionary " - "or instance of an SDO or SRO class.", - ) - - if data.get('revoked'): - raise RevokeError("revoke") - return new_version(data, revoked=True, allow_custom=True) - - def get_class_hierarchy_names(obj): """Given an object, return the names of the class hierarchy.""" names = [] @@ -512,64 +383,6 @@ def get_class_hierarchy_names(obj): return names -def remove_custom_stix(stix_obj): - """Remove any custom STIX objects or properties. - - Warnings: - This function is a best effort utility, in that it will remove custom - objects and properties based on the type names; i.e. if "x-" prefixes - object types, and "x\\_" prefixes property types. According to the - STIX2 spec, those naming conventions are a SHOULDs not MUSTs, meaning - that valid custom STIX content may ignore those conventions and in - effect render this utility function invalid when used on that STIX - content. - - Args: - stix_obj (dict OR python-stix obj): a single python-stix object - or dict of a STIX object - - Returns: - A new version of the object with any custom content removed - """ - - if stix_obj['type'].startswith('x-'): - # if entire object is custom, discard - return None - - custom_props = [] - for prop in stix_obj.items(): - if prop[0].startswith('x_'): - # for every custom property, record it and set value to None - # (so we can pass it to new_version() and it will be dropped) - custom_props.append((prop[0], None)) - - if custom_props: - # obtain set of object properties that can be transferred - # to a new object version. This is 1)custom props with their - # values set to None, and 2)any properties left that are not - # unmodifiable STIX properties or the "modified" property - - # set of properties that are not supplied to new_version() - # to be used for updating properties. This includes unmodifiable - # properties (properties that new_version() just re-uses from the - # existing STIX object) and the "modified" property. We dont supply the - # "modified" property so that new_version() creates a new datetime - # value for this property - non_supplied_props = STIX_UNMOD_PROPERTIES + ['modified'] - - props = [(prop, stix_obj[prop]) for prop in stix_obj if prop not in non_supplied_props] - - # add to set the custom properties we want to get rid of (with their value=None) - props.extend(custom_props) - - new_obj = new_version(stix_obj, **(dict(props))) - - return new_obj - - else: - return stix_obj - - def get_type_from_id(stix_id): return stix_id.split('--', 1)[0] diff --git a/stix2/versioning.py b/stix2/versioning.py new file mode 100644 index 00000000..8db8f7aa --- /dev/null +++ b/stix2/versioning.py @@ -0,0 +1,275 @@ +import copy +import datetime as dt +import itertools +import uuid + +import six +from six.moves.collections_abc import Mapping + +import stix2.base +from stix2.utils import get_timestamp, parse_into_datetime +import stix2.v20 + +from .exceptions import ( + InvalidValueError, RevokeError, UnmodifiablePropertyError, +) + +# STIX object properties that cannot be modified +STIX_UNMOD_PROPERTIES = ['created', 'created_by_ref', 'id', 'type'] +_VERSIONING_PROPERTIES = {"created", "modified", "revoked"} + + +def _fudge_modified(old_modified, new_modified, use_stix21): + """ + Ensures a new modified timestamp is newer than the old. When they are + too close together, new_modified must be pushed further ahead to ensure + it is distinct and later, after JSON serialization (which may mean it's + actually being pushed a little ways into the future). JSON serialization + can remove precision, which can cause distinct timestamps to accidentally + become equal, if we're not careful. + + :param old_modified: A previous "modified" timestamp, as a datetime object + :param new_modified: A candidate new "modified" timestamp, as a datetime + object + :param use_stix21: Whether to use STIX 2.1+ versioning timestamp precision + rules (boolean). This is important so that we are aware of how + timestamp precision will be truncated, so we know how close together + the timestamps can be, and how far ahead to potentially push the new + one. + :return: A suitable new "modified" timestamp. This may be different from + what was passed in, if it had to be pushed ahead. + """ + if use_stix21: + # 2.1+: we can use full precision + if new_modified <= old_modified: + new_modified = old_modified + dt.timedelta(microseconds=1) + else: + # 2.0: we must use millisecond precision + one_ms = dt.timedelta(milliseconds=1) + if new_modified - old_modified < one_ms: + new_modified = old_modified + one_ms + + return new_modified + + +def _is_versionable(data): + """ + Determine whether the given object is versionable. This check is done on + the basis of support for three properties for the object type: "created", + "modified", and "revoked". If all three are supported, the object is + versionable; otherwise it is not. Dicts must have a "type" property whose + value is for a registered object type. This is used to determine a + complete set of supported properties for the type. + + Also, detect whether it represents a STIX 2.1 or greater spec version. + + :param data: The object to check. Must be either a stix object, or a dict + with a "type" property. + :return: A 2-tuple of bools: the first is True if the object is versionable + and False if not; the second is True if the object is STIX 2.1+ and + False if not. + """ + + is_versionable = False + is_21 = False + stix_vid = None + + if isinstance(data, Mapping): + + # First, determine spec version. It's easy for our stix2 objects; more + # work for dicts. + is_21 = False + if isinstance(data, stix2.base._STIXBase) and \ + not isinstance(data, stix2.v20._STIXBase20): + # (is_21 means 2.1 or later; try not to be 2.1-specific) + is_21 = True + elif isinstance(data, dict): + stix_vid = stix2.parsing._detect_spec_version(data) + is_21 = stix_vid != "v20" + + # Then, determine versionability. + + if six.PY2: + # dumb python2 compatibility: map.keys() returns a list, not a set! + # six.viewkeys() compatibility function uses dict.viewkeys() on + # python2, which is not a Mapping mixin method, so that doesn't + # work either (for our stix2 objects). + keys = set(data) + else: + keys = data.keys() + + # This should be sufficient for STIX objects; maybe we get lucky with + # dicts here but probably not. + if keys >= _VERSIONING_PROPERTIES: + is_versionable = True + + # Tougher to handle dicts. We need to consider STIX version, map to a + # registered class, and from that get a more complete picture of its + # properties. + elif isinstance(data, dict): + class_maps = stix2.parsing.STIX2_OBJ_MAPS[stix_vid] + obj_type = data["type"] + + if obj_type in class_maps["objects"]: + # Should we bother checking properties for SDOs/SROs? + # They were designed to be versionable. + is_versionable = True + + elif obj_type in class_maps["observables"]: + # but do check SCOs + cls = class_maps["observables"][obj_type] + is_versionable = _VERSIONING_PROPERTIES.issubset( + cls._properties, + ) + + return is_versionable, is_21 + + +def new_version(data, allow_custom=None, **kwargs): + """ + Create a new version of a STIX object, by modifying properties and + updating the ``modified`` property. + + :param data: The object to create a new version of. Maybe a stix2 object + or dict. + :param allow_custom: Whether to allow custom properties on the new object. + If True, allow them (regardless of whether the original had custom + properties); if False disallow them; if None, propagate the preference + from the original object. + :param kwargs: The properties to change. Setting to None requests property + removal. + :return: The new object. + """ + + is_versionable, is_21 = _is_versionable(data) + + if not is_versionable: + raise ValueError( + "cannot create new version of object of this type! " + "Try a dictionary or instance of an SDO or SRO class.", + ) + + if data.get('revoked'): + raise RevokeError("new_version") + try: + new_obj_inner = copy.deepcopy(data._inner) + except AttributeError: + new_obj_inner = copy.deepcopy(data) + + # Make sure certain properties aren't trying to change + # ID contributing properties of 2.1+ SCOs may also not change if a UUIDv5 + # is in use (depending on whether they were used to create it... but they + # probably were). That would imply an ID change, which is not allowed + # across versions. + sco_locked_props = [] + if is_21 and isinstance(data, stix2.base._Observable): + uuid_ = uuid.UUID(data["id"][-36:]) + if uuid_.variant == uuid.RFC_4122 and uuid_.version == 5: + sco_locked_props = data._id_contributing_properties + + unchangable_properties = set() + for prop in itertools.chain(STIX_UNMOD_PROPERTIES, sco_locked_props): + if prop in kwargs: + unchangable_properties.add(prop) + if unchangable_properties: + raise UnmodifiablePropertyError(unchangable_properties) + + # Different versioning precision rules in STIX 2.0 vs 2.1, so we need + # to know which rules to apply. + precision_constraint = "min" if is_21 else "exact" + + cls = type(data) + if 'modified' not in kwargs: + old_modified = parse_into_datetime( + data["modified"], precision="millisecond", + precision_constraint=precision_constraint, + ) + + new_modified = get_timestamp() + new_modified = _fudge_modified(old_modified, new_modified, is_21) + + kwargs['modified'] = new_modified + + elif 'modified' in data: + old_modified_property = parse_into_datetime( + data.get('modified'), precision='millisecond', + precision_constraint=precision_constraint, + ) + new_modified_property = parse_into_datetime( + kwargs['modified'], precision='millisecond', + precision_constraint=precision_constraint, + ) + if new_modified_property <= old_modified_property: + raise InvalidValueError( + cls, 'modified', + "The new modified datetime cannot be before than or equal to the current modified datetime." + "It cannot be equal, as according to STIX 2 specification, objects that are different " + "but have the same id and modified timestamp do not have defined consumer behavior.", + ) + new_obj_inner.update(kwargs) + + # Set allow_custom appropriately if versioning an object. We will ignore + # it for dicts. + if isinstance(data, stix2.base._STIXBase): + if allow_custom is None: + new_obj_inner["allow_custom"] = data._allow_custom + else: + new_obj_inner["allow_custom"] = allow_custom + + # Exclude properties with a value of 'None' in case data is not an instance of a _STIXBase subclass + return cls(**{k: v for k, v in new_obj_inner.items() if v is not None}) + + +def revoke(data): + """Revoke a STIX object. + + Returns: + A new version of the object with ``revoked`` set to ``True``. + """ + if not isinstance(data, Mapping): + raise ValueError( + "cannot revoke object of this type! Try a dictionary " + "or instance of an SDO or SRO class.", + ) + + if data.get('revoked'): + raise RevokeError("revoke") + return new_version(data, revoked=True) + + +def remove_custom_stix(stix_obj): + """Remove any custom STIX objects or properties. + + Warnings: + This function is a best effort utility, in that it will remove custom + objects and properties based on the type names; i.e. if "x-" prefixes + object types, and "x\\_" prefixes property types. According to the + STIX2 spec, those naming conventions are a SHOULDs not MUSTs, meaning + that valid custom STIX content may ignore those conventions and in + effect render this utility function invalid when used on that STIX + content. + + Args: + stix_obj (dict OR python-stix obj): a single python-stix object + or dict of a STIX object + + Returns: + A new version of the object with any custom content removed + """ + + if stix_obj['type'].startswith('x-'): + # if entire object is custom, discard + return None + + custom_props = { + k: None + for k in stix_obj if k.startswith("x_") + } + + if custom_props: + new_obj = new_version(stix_obj, allow_custom=False, **custom_props) + + return new_obj + + else: + return stix_obj