Repository: oasis-open/cti-python-stix2
Pull request #203 (2018-07-02 22:29:20)

Add "contains" filter
This allows filtering on, for example, observed-data objects that contain a specific cyber observable.

Related: #200.


Patch:
diff --git a/docs/guide/datastore.ipynb b/docs/guide/datastore.ipynb
index 2fabf88..1c88cf1 100644
--- a/docs/guide/datastore.ipynb
+++ b/docs/guide/datastore.ipynb
@@ -369,6 +369,7 @@
     "* < \n",
     "* ```>=```\n",
     "* <=\n",
+    "* contains\n",
     "\n",
     "Value types of the property values must be one of these (Python) types:\n",
     "\n",
diff --git a/stix2/base.py b/stix2/base.py
index c26f60a..7c62db4 100644
--- a/stix2/base.py
+++ b/stix2/base.py
@@ -56,7 +56,7 @@ class STIXJSONIncludeOptionalDefaultsEncoder(json.JSONEncoder):
         elif isinstance(obj, _STIXBase):
             return dict(obj)
         else:
-            return super(STIXJSONIncludeOptionalDefaultsEncoder, self).default(obj)
+            return super(STIXJSONEncoder, self).default(obj)
 
 
 def get_required_properties(properties):
@@ -270,8 +270,10 @@ class _STIXBase(collections.Mapping):
             overridden: indent=4, separators=(",", ": "), item_sort_key=sort_by.
         """
         if pretty:
+            properties = self.object_properties()
+
             def sort_by(element):
-                return find_property_index(self, *element)
+                return find_property_index(self, properties, element)
 
             kwargs.update({'indent': 4, 'separators': (",", ": "), 'item_sort_key': sort_by})
 
diff --git a/stix2/datastore/filters.py b/stix2/datastore/filters.py
index a32b14a..03585ad 100644
--- a/stix2/datastore/filters.py
+++ b/stix2/datastore/filters.py
@@ -9,7 +9,7 @@ from datetime import datetime
 from stix2.utils import format_datetime
 
 """Supported filter operations"""
-FILTER_OPS = ['=', '!=', 'in', '>', '<', '>=', '<=']
+FILTER_OPS = ['=', '!=', 'in', '>', '<', '>=', '<=', 'contains']
 
 """Supported filter value types"""
 FILTER_VALUE_TYPES = [bool, dict, float, int, list, str, tuple]
@@ -100,6 +100,11 @@ class Filter(collections.namedtuple("Filter", ['property', 'op', 'value'])):
             return stix_obj_property != self.value
         elif self.op == "in":
             return stix_obj_property in self.value
+        elif self.op == "contains":
+            if isinstance(self.value, dict):
+                return self.value in stix_obj_property.values()
+            else:
+                return self.value in stix_obj_property
         elif self.op == ">":
             return stix_obj_property > self.value
         elif self.op == "<":
diff --git a/stix2/utils.py b/stix2/utils.py
index 64c4a32..137ea01 100644
--- a/stix2/utils.py
+++ b/stix2/utils.py
@@ -165,74 +165,87 @@ def _get_dict(data):
             raise ValueError("Cannot convert '%s' to dictionary." % str(data))
 
 
-def _find(seq, val):
-    """
-    Search sequence 'seq' for val.  This behaves like str.find(): if not found,
-    -1 is returned instead of throwing an exception.
+def _iterate_over_values(dict_values, tuple_to_find):
+    """Loop recursively over dictionary values"""
+    from .base import _STIXBase
+    for pv in dict_values:
+        if isinstance(pv, list):
+            for item in pv:
+                if isinstance(item, _STIXBase):
+                    index = find_property_index(
+                        item,
+                        item.object_properties(),
+                        tuple_to_find
+                    )
+                    if index is not None:
+                        return index
+                elif isinstance(item, dict):
+                    for idx, val in enumerate(sorted(item)):
+                        if (tuple_to_find[0] == val and
+                                item.get(val) == tuple_to_find[1]):
+                            return idx
+        elif isinstance(pv, dict):
+            if pv.get(tuple_to_find[0]) is not None:
+                for idx, item in enumerate(sorted(pv.keys())):
+                    if ((item == tuple_to_find[0] and str.isdigit(item)) and
+                            (pv[item] == tuple_to_find[1])):
+                        return int(tuple_to_find[0])
+                    elif pv[item] == tuple_to_find[1]:
+                        return idx
+            for item in pv.values():
+                if isinstance(item, _STIXBase):
+                    index = find_property_index(
+                        item,
+                        item.object_properties(),
+                        tuple_to_find
+                    )
+                    if index is not None:
+                        return index
+                elif isinstance(item, dict):
+                    index = find_property_index(
+                        item,
+                        item.keys(),
+                        tuple_to_find
+                    )
+                    if index is not None:
+                        return index
+
+
+def find_property_index(obj, properties, tuple_to_find):
+    """Recursively find the property in the object model, return the index
+    according to the ``properties`` OrderedDict when working with `stix2`
+    objects. If it's a list look for individual objects. Returns and integer
+    indicating its location.
+
+    Notes:
+        This method is intended to pretty print `stix2` properties for better
+        visual feedback when working with the library.
+
+    Warnings:
+        This method may not be able to produce the same output if called
+        multiple times and makes a best effort attempt to print the properties
+        according to the STIX technical specification.
+
+    See Also:
+        py:meth:`stix2.base._STIXBase.serialize` for more information.
 
-    :param seq: The sequence to search
-    :param val: The value to search for
-    :return: The index of the value if found, or -1 if not found
     """
+    from .base import _STIXBase
     try:
-        return seq.index(val)
+        if isinstance(obj, _STIXBase):
+            if tuple_to_find[1] in obj._inner.values():
+                return properties.index(tuple_to_find[0])
+        elif isinstance(obj, dict):
+            for idx, val in enumerate(sorted(obj)):
+                if (tuple_to_find[0] == val and
+                        obj.get(val) == tuple_to_find[1]):
+                    return idx
+        raise ValueError
     except ValueError:
-        return -1
-
-
-def _find_property_in_seq(seq, search_key, search_value):
-    """
-    Helper for find_property_index(): search for the property in all elements
-    of the given sequence.
-
-    :param seq: The sequence
-    :param search_key: Property name to find
-    :param search_value: Property value to find
-    :return: A property index, or -1 if the property was not found
-    """
-    idx = -1
-    for elem in seq:
-        idx = find_property_index(elem, search_key, search_value)
-        if idx >= 0:
-            break
-
-    return idx
-
-
-def find_property_index(obj, search_key, search_value):
-    """
-    Search (recursively) for the given key and value in the given object.
-    Return an index for the key, relative to whatever object it's found in.
-
-    :param obj: The object to search (list, dict, or stix object)
-    :param search_key: A search key
-    :param search_value: A search value
-    :return: An index; -1 if the key and value aren't found
-    """
-    from .base import _STIXBase
-
-    # Special-case keys which are numbers-as-strings, e.g. for cyber-observable
-    # mappings.  Use the int value of the key as the index.
-    if search_key.isdigit():
-        return int(search_key)
-
-    if isinstance(obj, _STIXBase):
-        if search_key in obj and obj[search_key] == search_value:
-            idx = _find(obj.object_properties(), search_key)
-        else:
-            idx = _find_property_in_seq(obj.values(), search_key, search_value)
-    elif isinstance(obj, dict):
-        if search_key in obj and obj[search_key] == search_value:
-            idx = _find(sorted(obj), search_key)
-        else:
-            idx = _find_property_in_seq(obj.values(), search_key, search_value)
-    elif isinstance(obj, list):
-        idx = _find_property_in_seq(obj, search_key, search_value)
-    else:
-        # Don't know how to search this type
-        idx = -1
-
-    return idx
+        if isinstance(obj, _STIXBase):
+            return _iterate_over_values(obj._inner.values(), tuple_to_find)
+        elif isinstance(obj, dict):
+            return _iterate_over_values(obj.values(), tuple_to_find)
 
 
 def new_version(data, **kwargs):
diff --git a/stix2/v20/observables.py b/stix2/v20/observables.py
index 659929b..098c7dd 100644
--- a/stix2/v20/observables.py
+++ b/stix2/v20/observables.py
@@ -302,7 +302,7 @@ class RasterImageExt(_Extension):
     _properties = OrderedDict()
     _properties.update([
         ('image_height', IntegerProperty()),
-        ('image_width', IntegerProperty()),
+        ('image_weight', IntegerProperty()),
         ('bits_per_pixel', IntegerProperty()),
         ('image_compression_algorithm', StringProperty()),
         ('exif_tags', DictionaryProperty()),


Test patch:
diff --git a/stix2/test/test_datastore_filters.py b/stix2/test/test_datastore_filters.py
index 5ffd051..4ec6e7e 100644
--- a/stix2/test/test_datastore_filters.py
+++ b/stix2/test/test_datastore_filters.py
@@ -100,7 +100,9 @@ filters = [
     Filter("object_marking_refs", "=", "marking-definition--613f2e26-0000-0000-0000-b8e91df99dc9"),
     Filter("granular_markings.selectors", "in", "description"),
     Filter("external_references.source_name", "=", "CVE"),
-    Filter("objects", "=", {"0": {"type": "file", "name": "HAL 9000.exe"}})
+    Filter("objects", "=", {"0": {"type": "file", "name": "HAL 9000.exe"}}),
+    Filter("objects", "contains", {"type": "file", "name": "HAL 9000.exe"}),
+    Filter("labels", "contains", "heartbleed"),
 ]
 
 # same as above objects but converted to real Python STIX2 objects
@@ -302,6 +304,28 @@ def test_apply_common_filters13():
     assert len(resp) == 1
 
 
+def test_apply_common_filters14():
+    # Return any object that contains a specific File Cyber Observable Object
+    resp = list(apply_common_filters(stix_objs, [filters[15]]))
+    assert resp[0]['id'] == stix_objs[4]['id']
+    assert len(resp) == 1
+
+    resp = list(apply_common_filters(real_stix_objs, [filters[15]]))
+    assert resp[0].id == real_stix_objs[4].id
+    assert len(resp) == 1
+
+
+def test_apply_common_filters15():
+    # Return any object that contains 'heartbleed' in "labels"
+    resp = list(apply_common_filters(stix_objs, [filters[16]]))
+    assert resp[0]['id'] == stix_objs[3]['id']
+    assert len(resp) == 1
+
+    resp = list(apply_common_filters(real_stix_objs, [filters[16]]))
+    assert resp[0].id == real_stix_objs[3].id
+    assert len(resp) == 1
+
+
 def test_datetime_filter_behavior():
     """if a filter is initialized with its value being a datetime object
     OR the STIX object property being filtered on is a datetime object, all
diff --git a/stix2/test/test_observed_data.py b/stix2/test/test_observed_data.py
index 11c74ca..30c3cab 100644
--- a/stix2/test/test_observed_data.py
+++ b/stix2/test/test_observed_data.py
@@ -765,51 +765,6 @@ def test_file_example_with_RasterImageExt_Object():
     assert f.extensions["raster-image-ext"].exif_tags["XResolution"] == 4928
 
 
-RASTER_IMAGE_EXT = """{
-"type": "observed-data",
-"id": "observed-data--b67d30ff-02ac-498a-92f9-32f845f448cf",
-"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": 1,
-"objects": {
-  "0": {
-    "type": "file",
-    "name": "picture.jpg",
-    "hashes": {
-      "SHA-256": "35a01331e9ad96f751278b891b6ea09699806faedfa237d40513d92ad1b7100f"
-    },
-    "extensions": {
-      "raster-image-ext": {
-        "image_height": 768,
-        "image_width": 1024,
-        "bits_per_pixel": 72,
-        "image_compression_algorithm": "JPEG",
-        "exif_tags": {
-          "Make": "Nikon",
-          "Model": "D7000",
-          "XResolution": 4928,
-          "YResolution": 3264
-        }
-      }
-    }
-  }
-}
-}
-"""
-
-
-def test_raster_image_ext_parse():
-    obj = stix2.parse(RASTER_IMAGE_EXT)
-    assert obj.objects["0"].extensions['raster-image-ext'].image_width == 1024
-
-
-def test_raster_images_ext_create():
-    ext = stix2.RasterImageExt(image_width=1024)
-    assert "image_width" in str(ext)
-
-
 def test_file_example_with_WindowsPEBinaryExt():
     f = stix2.File(name="qwerty.dll",
                    extensions={
diff --git a/stix2/test/test_utils.py b/stix2/test/test_utils.py
index 885c4d9..fb63ff7 100644
--- a/stix2/test/test_utils.py
+++ b/stix2/test/test_utils.py
@@ -178,7 +178,8 @@ def test_deduplicate(stix_objs1):
 def test_find_property_index(object, tuple_to_find, expected_index):
     assert stix2.utils.find_property_index(
         object,
-        *tuple_to_find
+        [],
+        tuple_to_find
     ) == expected_index
 
 
@@ -207,4 +208,4 @@ def test_find_property_index(object, tuple_to_find, expected_index):
     }, ('key_one', 1), 0)
 ])
 def test_iterate_over_values(dict_value, tuple_to_find, expected_index):
-    assert stix2.utils._find_property_in_seq(dict_value.values(), *tuple_to_find) == expected_index
+    assert stix2.utils._iterate_over_values(dict_value.values(), tuple_to_find) == expected_index