156 lines
4.6 KiB
Python
156 lines
4.6 KiB
Python
from stix2 import Bundle, ThreatActor, TAXIICollectionSource, TAXIICollectionSink
|
|
from stix2.datastore.filters import Filter
|
|
|
|
import json
|
|
import pytest
|
|
|
|
from taxii2client import Collection, _filter_kwargs_to_query_params
|
|
from medallion.filters.basic_filter import BasicFilter
|
|
|
|
COLLECTION_URL = 'https://example.com/api1/collections/91a7b528-80eb-42ed-a74d-c6fbd5a26116/'
|
|
|
|
|
|
class MockTAXIICollectionEndpoint(Collection):
|
|
"""Mock for taxii2_client.TAXIIClient"""
|
|
|
|
def __init__(self, url, **kwargs):
|
|
super(MockTAXIICollectionEndpoint, self).__init__(url, **kwargs)
|
|
self.objects = []
|
|
|
|
def add_objects(self, bundle):
|
|
self._verify_can_write()
|
|
if isinstance(bundle, str):
|
|
bundle = json.loads(bundle)
|
|
for object in bundle.get("objects", []):
|
|
self.objects.append(object)
|
|
|
|
def get_objects(self, **filter_kwargs):
|
|
self._verify_can_read()
|
|
query_params = _filter_kwargs_to_query_params(filter_kwargs)
|
|
query_params = json.loads(query_params)
|
|
full_filter = BasicFilter(query_params or {})
|
|
objs = full_filter.process_filter(
|
|
self.objects,
|
|
("id", "type", "version"),
|
|
[]
|
|
)
|
|
return Bundle(objects=objs)
|
|
|
|
def get_object(self, id, version=None):
|
|
self._verify_can_read()
|
|
query_params = None
|
|
if version:
|
|
query_params = _filter_kwargs_to_query_params({"version": version})
|
|
if query_params:
|
|
query_params = json.loads(query_params)
|
|
full_filter = BasicFilter(query_params or {})
|
|
objs = full_filter.process_filter(
|
|
self.objects,
|
|
("version",),
|
|
[]
|
|
)
|
|
return Bundle(objects=objs)
|
|
|
|
|
|
@pytest.fixture
|
|
def collection(stix_objs1):
|
|
mock = MockTAXIICollectionEndpoint(COLLECTION_URL, **{
|
|
"id": "91a7b528-80eb-42ed-a74d-c6fbd5a26116",
|
|
"title": "Writable Collection",
|
|
"description": "This collection is a dropbox for submitting indicators",
|
|
"can_read": True,
|
|
"can_write": True,
|
|
"media_types": [
|
|
"application/vnd.oasis.stix+json; version=2.0"
|
|
]
|
|
})
|
|
|
|
mock.objects.extend(stix_objs1)
|
|
return mock
|
|
|
|
|
|
def test_ds_taxii(collection):
|
|
ds = TAXIICollectionSource(collection)
|
|
assert ds.collection is not None
|
|
|
|
|
|
def test_add_stix2_object(collection):
|
|
tc_sink = TAXIICollectionSink(collection)
|
|
|
|
# create new STIX threat-actor
|
|
ta = ThreatActor(name="Teddy Bear",
|
|
labels=["nation-state"],
|
|
sophistication="innovator",
|
|
resource_level="government",
|
|
goals=[
|
|
"compromising environment NGOs",
|
|
"water-hole attacks geared towards energy sector",
|
|
])
|
|
|
|
tc_sink.add(ta)
|
|
|
|
|
|
def test_get_stix2_object(collection):
|
|
tc_sink = TAXIICollectionSource(collection)
|
|
|
|
objects = tc_sink.get("indicator--d81f86b9-975b-bc0b-775e-810c5ad45a4f")
|
|
|
|
assert objects
|
|
|
|
|
|
def test_parse_taxii_filters():
|
|
query = [
|
|
Filter("added_after", "=", "2016-02-01T00:00:01.000Z"),
|
|
Filter("id", "=", "taxii stix object ID"),
|
|
Filter("type", "=", "taxii stix object ID"),
|
|
Filter("version", "=", "first"),
|
|
Filter("created_by_ref", "=", "Bane"),
|
|
]
|
|
|
|
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 = TAXIICollectionSource(collection)
|
|
|
|
taxii_filters = ds._parse_taxii_filters(query)
|
|
|
|
assert taxii_filters == taxii_filters_expected
|
|
|
|
|
|
def test_add_get_remove_filter():
|
|
ds = TAXIICollectionSource(collection)
|
|
|
|
# First 3 filters are valid, remaining properties are erroneous in some way
|
|
valid_filters = [
|
|
Filter('type', '=', 'malware'),
|
|
Filter('id', '!=', 'stix object id'),
|
|
Filter('labels', 'in', ["heartbleed", "malicious-activity"]),
|
|
]
|
|
|
|
assert len(ds.filters) == 0
|
|
|
|
ds.filters.add(valid_filters[0])
|
|
assert len(ds.filters) == 1
|
|
|
|
# Addin the same filter again will have no effect since `filters` uses a set
|
|
ds.filters.add(valid_filters[0])
|
|
assert len(ds.filters) == 1
|
|
|
|
ds.filters.add(valid_filters[1])
|
|
assert len(ds.filters) == 2
|
|
ds.filters.add(valid_filters[2])
|
|
assert len(ds.filters) == 3
|
|
|
|
assert set(valid_filters) == ds.filters
|
|
|
|
# remove
|
|
ds.filters.remove(valid_filters[0])
|
|
|
|
assert len(ds.filters) == 2
|
|
|
|
ds.filters.update(valid_filters)
|