def parse_unique_field_results()

in detection_rules/eswrap.py [0:0]

• 20 lines of code
• 13 McCabe index (conditional complexity)

def parse_unique_field_results(rule_type: str, unique_fields: List[str], search_results: dict):
    parsed_results = defaultdict(lambda: defaultdict(int))
    hits = search_results['hits']
    hits = hits['hits'] if rule_type != 'eql' else hits.get('events') or hits.get('sequences', [])
    for hit in hits:
        for field in unique_fields:
            if 'events' in hit:
                match = []
                for event in hit['events']:
                    matched = nested_get(event['_source'], field)
                    match.extend([matched] if not isinstance(matched, list) else matched)
                    if not match:
                        continue
            else:
                match = nested_get(hit['_source'], field)
                if not match:
                    continue

            match = ','.join(sorted(match)) if isinstance(match, list) else match
            parsed_results[field][match] += 1
    # if rule.type == eql, structure is different
    return {'results': parsed_results} if parsed_results else {}