adds support for provider lists and negations (#495)

adds support for negation (via prepended exclamation mark) on both providers and tags (e.g.: 'tags: !botnet' to exclude botnet tags or 'provider: !csirtg.io, !otherprovid.xyz' to exclude listed providers)
attempt at ES search tuning by favoring filters over queries

Co-authored-by: wes <[email protected]>

Author: mdavis332

cif/store/zelasticsearch/filters.py

@@ -65,17 +65,36 @@ def filter_reporttime(s, filter):
     if not filter.get('reporttime'):
         return s
 
-    c = filter.pop('reporttime')
-    if PYVERSION == 2:
-        if type(c) == unicode:
-            c = str(c)
+    high = 'now/m'
+    # if passed 'days' or 'hours', preferentially use that for ES filtering/caching
+    if filter.get('days') or filter.get('hours'):
+        if filter.get('hours'):
+            lookback_amount = filter.pop('hours')
+            lookback_unit = 'h'
+        elif filter.get('days'):
+            lookback_amount = filter.pop('days')
+            lookback_unit = 'd'
+
+        try:
+            lookback_amount = int(lookback_amount)
+        except Exception as e:
+            raise InvalidSearch('Lookback time filter {}{} is not a valid time'.format(lookback_amount, lookback_unit))
+
+        # don't put spaces in relative date math operator query to make it easier to read. ES hates that and will error.
+        low = 'now/m-{}{}'.format(lookback_amount, lookback_unit)
+    # no relative 'days' or 'hours' params, so fallback to 'reporttime'
+    else:
+        c = filter.pop('reporttime')
+        if PYVERSION == 2:
+            if type(c) == unicode:
+                c = str(c)
 
-    low, high = c, arrow.utcnow()
-    if isinstance(c, basestring) and ',' in c:
-        low, high = c.split(',')
+        if isinstance(c, basestring) and ',' in c:
+            low, high = c.split(',')
+        else:
+            low = c
 
-    low = arrow.get(low).datetime
-    high = arrow.get(high).datetime
+        low = arrow.get(low).datetime
 
     s = s.filter('range', reporttime={'gte': low, 'lte': high})
     return s
@@ -114,7 +133,7 @@ def filter_indicator(s, q_filters):
 
 def filter_terms(s, q_filters):
     for f in q_filters:
-        if f in ['nolog', 'days', 'hours', 'groups', 'limit', 'tags']:
+        if f in ['nolog', 'days', 'hours', 'groups', 'limit', 'provider', 'reporttime', 'tags']:
             continue
 
         kwargs = {f: q_filters[f]}
@@ -127,16 +146,70 @@ def filter_terms(s, q_filters):
 
 
 def filter_tags(s, q_filters):
-    tags = q_filters['tags']
+    if not q_filters.get('tags'):
+        return s
+
+    tags = q_filters.pop('tags')
 
     if isinstance(tags, basestring):
-        tags = tags.split(',')
+        tags = [x.strip() for x in tags.split(',')]
 
+    # each array element is implicitly ORed (aka, 'should') using a terms filter
+    #s = s.filter('terms', tags=tags)
     tt = []
+    not_tt = []
     for t in tags:
-        tt.append(Q('term', tags=t))
+        # used for tags exclusion/negation
+        if t.startswith('!'):
+            t = t[1:]
+            not_tt.append(t)
+        else:
+            tt.append(t)
+
+    if len(not_tt) > 0:
+        if len(not_tt) == 1:
+            s = s.exclude('term', tags=not_tt[0])
+        else:
+            s = s.exclude('terms', tags=not_tt)
+
+    if len(tt) > 0:
+        if len(tt) == 1:
+            s = s.filter('term', tags=tt[0])
+        else:
+            s = s.filter('terms', tags=tt)
+
+    return s
+
+def filter_provider(s, q_filters):
+    if not q_filters.get('provider'):
+        return s
 
-    s.query = Q('bool', must=s.query, should=tt, minimum_should_match=1)
+    provider = q_filters.pop('provider')
+
+    if isinstance(provider, basestring):
+        provider = [x.strip() for x in provider.split(',')]
+
+    pp = []
+    not_pp = []
+    for p in provider:
+        # used for provider exclusion/negation
+        if p.startswith('!'):
+            p = p[1:]
+            not_pp.append(p)
+        else:
+            pp.append(p)
+
+    if len(not_pp) > 0:
+        if len(not_pp) == 1:
+            s = s.exclude('term', provider=not_pp[0])
+        else:
+            s = s.exclude('terms', provider=not_pp)
+
+    if len(pp) > 0:
+        if len(pp) == 1:
+            s = s.filter('term', provider=pp[0])
+        else:
+            s = s.filter('terms', provider=pp)
 
     return s
 
@@ -145,16 +218,13 @@ def filter_groups(s, q_filters, token=None):
     if token:
         groups = token.get('groups', 'everyone')
     else:
-        groups = q_filters['groups']
+        groups = q_filters.pop('groups')
 
     if isinstance(groups, basestring):
         groups = [groups]
 
-    gg = []
-    for g in groups:
-        gg.append(Q('term', group=g))
-
-    s.query = Q('bool', must=s.query, should=gg, minimum_should_match=1)
+    # each array element is implicitly ORed (aka, 'should') using a terms filter
+    s = s.filter('terms', group=groups)
 
     return s
 
@@ -171,32 +241,34 @@ def filter_id(s, q_filters):
 def filter_build(s, filters, token=None):
     limit = filters.get('limit')
     if limit and int(limit) > WINDOW_LIMIT:
-        raise InvalidSearch('request limit should be <= server threshold of {} but was set to {}'.format(WINDOW_LIMIT, limit))
-        
+        raise InvalidSearch('Request limit should be <= server threshold of {} but was set to {}'.format(WINDOW_LIMIT, limit))
+
     q_filters = {}
     for f in VALID_FILTERS:
         if filters.get(f):
             q_filters[f] = filters[f]
 
-    # treat indicator as special, transform into Search
-    s = filter_indicator(s, q_filters)
+    s = filter_provider(s, q_filters)
+
+    s = filter_confidence(s, q_filters)
 
     s = filter_id(s, q_filters)
 
-    s = filter_confidence(s, q_filters)
+    # treat indicator as special, transform into Search
+    s = filter_indicator(s, q_filters)
 
     s = filter_reporttime(s, q_filters)
 
     # transform all other filters into term=
     s = filter_terms(s, q_filters)
 
-    if filters.get('tags'):
-        s = filter_tags(s, filters)
-
-    if filters.get('groups'):
-        s = filter_groups(s, filters)
+    if q_filters.get('groups'):
+        s = filter_groups(s, q_filters)
     else:
         if token:
             s = filter_groups(s, {}, token=token)
 
+    if q_filters.get('tags'):
+        s = filter_tags(s, q_filters)
+
     return s

test/zelasticsearch/test_store_elasticsearch_indicators.py

@@ -58,6 +58,18 @@ def indicator():
     )
 
 
+@pytest.fixture
+def indicator_alt_provider():
+    return Indicator(
+        indicator='example2.com',
+        tags='phishing',
+        provider='notcsirtg.io',
+        group='everyone',
+        lasttime=arrow.utcnow().datetime,
+        reporttime=arrow.utcnow().datetime
+    )
+
+
 @pytest.fixture
 def indicator_email():
     return Indicator(
@@ -232,3 +244,89 @@ def test_store_elasticsearch_indicators_malware(store, token, indicator_malware)
     assert (x[0]['reporttime'])
 
     assert x[0]['indicator'] == indicator_malware.indicator
+
+
+## test returning a list of providers
+@pytest.mark.skipif(DISABLE_TESTS, reason='need to set CIF_ELASTICSEARCH_TEST=1 to run')
+def test_store_elasticsearch_indicators_provider_list(store, token, indicator, indicator_alt_provider):
+    x = store.handle_indicators_create(token, indicator.__dict__(), flush=True)
+    assert x == 1
+    
+    y = store.handle_indicators_create(token, indicator_alt_provider.__dict__(), flush=True)
+    assert y == 1
+
+    x = store.handle_indicators_search(token, {
+        'provider': '{}, {}'.format(indicator.provider, indicator_alt_provider.provider)
+    })
+
+    x = json.loads(x)
+    pprint(x)
+    
+    x = [i['_source'] for i in x['hits']['hits']]
+    
+    assert len(x) == 2
+
+
+## test provider negation
+@pytest.mark.skipif(DISABLE_TESTS, reason='need to set CIF_ELASTICSEARCH_TEST=1 to run')
+def test_store_elasticsearch_indicators_provider_negation(store, token, indicator, indicator_alt_provider):
+    x = store.handle_indicators_create(token, indicator.__dict__(), flush=True)
+    assert x == 1
+    
+    y = store.handle_indicators_create(token, indicator_alt_provider.__dict__(), flush=True)
+    assert y == 1
+
+    x = store.handle_indicators_search(token, {
+        'provider': '!{}'.format(indicator.provider)
+    })
+
+    x = json.loads(x)
+    pprint(x)
+    
+    x = [i['_source'] for i in x['hits']['hits']]
+    
+    assert len(x) == 1
+    
+    assert x[0]['provider'] == indicator_alt_provider.provider
+    
+    
+## test tags negation
+@pytest.mark.skipif(DISABLE_TESTS, reason='need to set CIF_ELASTICSEARCH_TEST=1 to run')
+def test_store_elasticsearch_indicators_tags_negation(store, token, indicator, indicator_alt_provider):
+    x = store.handle_indicators_create(token, indicator.__dict__(), flush=True)
+    assert x == 1
+    
+    y = store.handle_indicators_create(token, indicator_alt_provider.__dict__(), flush=True)
+    assert y == 1
+
+    x = store.handle_indicators_search(token, {
+        'tags': '!{}'.format(indicator_alt_provider.tags[0])
+    })
+
+    x = json.loads(x)
+    pprint(x)
+    
+    x = [i['_source'] for i in x['hits']['hits']]
+    
+    assert len(x) == 1
+    
+    assert x[0]['tags'] == indicator.tags
+
+
+## test multi-tags negation
+@pytest.mark.skipif(DISABLE_TESTS, reason='need to set CIF_ELASTICSEARCH_TEST=1 to run')
+def test_store_elasticsearch_indicators_multi_tags_negation(store, token, indicator, indicator_alt_provider):
+    x = store.handle_indicators_create(token, indicator.__dict__(), flush=True)
+    assert x == 1
+    
+    y = store.handle_indicators_create(token, indicator_alt_provider.__dict__(), flush=True)
+    assert y == 1
+
+    x = store.handle_indicators_search(token, {
+        'tags': '!{},!{}'.format(indicator.tags[0], indicator_alt_provider.tags[0])
+    })
+
+    x = json.loads(x)
+    pprint(x)
+    
+    assert len(x) == 0