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

Merge pull request #147 from oasis-open/119-taxii-filters 

119 taxii filters
stix2.0
Greg Back 2018-03-28 15:24:28 -05:00 committed by GitHub
parent 369eb604ef 51dcd6c184
commit e92db2417a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 79 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
stix2/datastore/filters.py
View File

@ -35,7 +35,11 @@ def _check_filter_components(prop, op, value):
if type(value) not in FILTER_VALUE_TYPES:
# check filter value type is supported
raise TypeError("Filter value type '%s' is not supported. The type must be a Python immutable type or dictionary" % type(value))
raise TypeError("Filter value of '%s' is not supported. The type must be a Python immutable type or dictionary" % type(value))
if prop == "type" and "_" in value:
# check filter where the property is type, value (type name) cannot have underscores
raise ValueError("Filter for property 'type' cannot have its value '%s' include underscores" % value)
return True

46
stix2/datastore/taxii.py
View File

@ -224,18 +224,21 @@ class TAXIICollectionSource(DataSource):
if _composite_filters:
query.update(_composite_filters)
# separate taxii query terms (can be done remotely)
# parse taxii query params (that can be applied remotely)
taxii_filters = self._parse_taxii_filters(query)
# taxii2client requires query params as keywords
taxii_filters_dict = dict((f.property, f.value) for f in taxii_filters)
# query TAXII collection
try:
all_data = self.collection.get_objects(filters=taxii_filters)["objects"]
all_data = self.collection.get_objects(**taxii_filters_dict)["objects"]
# deduplicate data (before filtering as reduces wasted filtering)
all_data = deduplicate(all_data)
# apply local (CompositeDataSource, TAXIICollectionSource and query filters)
all_data = list(apply_common_filters(all_data, query))
# apply local (CompositeDataSource, TAXIICollectionSource and query) filters
all_data = list(apply_common_filters(all_data, (query - taxii_filters)))
except HTTPError:
# if resources not found or access is denied from TAXII server, return empty list
@ -247,30 +250,35 @@ class TAXIICollectionSource(DataSource):
return stix_objs
def _parse_taxii_filters(self, query):
"""Parse out TAXII filters that the TAXII server can filter on.
"""Parse out TAXII filters that the TAXII server can filter on
Note:
For instance - "?match[type]=indicator,sighting" should be in a
query dict as follows:
Does not put in TAXII spec format as the TAXII2Client (that we use)
does this for us.
NOTE:
Currently, the TAXII2Client can handle TAXII filters where the
filter value is list, as both a comma-seperated string or python list
For instance - "?match[type]=indicator,sighting" can be in a
filter in any of these formats:
Filter("type", "<any op>", "indicator,sighting")
Filter("type", "<any op>", ["indicator", "sighting"])
Filter("type", "=", "indicator,sighting")
Args:
query (list): list of filters to extract which ones are TAXII
query (set): set of filters to extract which ones are TAXII
specific.
Returns:
params (dict): dict of the TAXII filters but in format required
for 'requests.get()'.
taxii_filters (set): set of the TAXII filters
"""
params = {}
taxii_filters = set()
for filter_ in query:
if filter_.property in TAXII_FILTERS:
if filter_.property == "added_after":
params[filter_.property] = filter_.value
else:
taxii_field = "match[%s]" % filter_.property
params[taxii_field] = filter_.value
return params
taxii_filters.add(filter_)
return taxii_filters

70
stix2/test/test_datastore.py
View File

@ -148,18 +148,18 @@ def test_parse_taxii_filters():
Filter("created_by_ref", "=", "Bane"),
]
expected_params = {
"added_after": "2016-02-01T00:00:01.000Z",
"match[id]": "taxii stix object ID",
"match[type]": "taxii stix object ID",
"match[version]": "first"
}
taxii_filters_expected = set([
Filter("added_after", "=", "2016-02-01T00:00:01.000Z"),
Filter("id", "=", "taxii stix object ID"),
Filter("type", "=", "taxii stix object ID"),
Filter("version", "=", "first")
])
ds = taxii.TAXIICollectionSource(collection)
taxii_filters = ds._parse_taxii_filters(query)
assert taxii_filters == expected_params
assert taxii_filters == taxii_filters_expected
def test_add_get_remove_filter():
@ -172,22 +172,6 @@ def test_add_get_remove_filter():
Filter('labels', 'in', ["heartbleed", "malicious-activity"]),
]
# Invalid filters - wont pass creation
# these filters will not be allowed to be created
# check proper errors are raised when trying to create them
with pytest.raises(ValueError) as excinfo:
# create Filter that has an operator that is not allowed
Filter('modified', '*', 'not supported operator - just place holder')
assert str(excinfo.value) == "Filter operator '*' not supported for specified property: 'modified'"
with pytest.raises(TypeError) as excinfo:
# create Filter that has a value type that is not allowed
Filter('created', '=', object())
# On Python 2, the type of object() is `<type 'object'>` On Python 3, it's `<class 'object'>`.
assert str(excinfo.value).startswith("Filter value type")
assert str(excinfo.value).endswith("is not supported. The type must be a Python immutable type or dictionary")
assert len(ds.filters) == 0
ds.filters.add(valid_filters[0])
@ -212,6 +196,46 @@ def test_add_get_remove_filter():
ds.filters.update(valid_filters)
def test_filter_ops_check():
# invalid filters - non supported operators
with pytest.raises(ValueError) as excinfo:
# create Filter that has an operator that is not allowed
Filter('modified', '*', 'not supported operator')
assert str(excinfo.value) == "Filter operator '*' not supported for specified property: 'modified'"
with pytest.raises(ValueError) as excinfo:
Filter("type", "%", "4")
assert "Filter operator '%' not supported for specified property" in str(excinfo.value)
def test_filter_value_type_check():
# invalid filters - non supported value types
with pytest.raises(TypeError) as excinfo:
Filter('created', '=', object())
# On Python 2, the type of object() is `<type 'object'>` On Python 3, it's `<class 'object'>`.
assert any([s in str(excinfo.value) for s in ["<type 'object'>", "'<class 'object'>'"]])
assert "is not supported. The type must be a Python immutable type or dictionary" in str(excinfo.value)
with pytest.raises(TypeError) as excinfo:
Filter("type", "=", complex(2, -1))
assert any([s in str(excinfo.value) for s in ["<type 'complex'>", "'<class 'complex'>'"]])
assert "is not supported. The type must be a Python immutable type or dictionary" in str(excinfo.value)
with pytest.raises(TypeError) as excinfo:
Filter("type", "=", set([16, 23]))
assert any([s in str(excinfo.value) for s in ["<type 'set'>", "'<class 'set'>'"]])
assert "is not supported. The type must be a Python immutable type or dictionary" in str(excinfo.value)
def test_filter_type_underscore_check():
# check that Filters where property="type", value (name) doesnt have underscores
with pytest.raises(ValueError) as excinfo:
Filter("type", "=", "oh_underscore")
assert "Filter for property 'type' cannot have its value 'oh_underscore'" in str(excinfo.value)
def test_apply_common_filters():
stix_objs = [
{