From 605b962c75b671243ee4b080efef12d8b4944cb5 Mon Sep 17 00:00:00 2001 From: Lou Stella Date: Mon, 28 Mar 2022 14:26:35 -0500 Subject: [PATCH] Updated indicator_collect CF --- .../custom_functions/indicator_collect.json | 43 ++++- .../custom_functions/indicator_collect.py | 169 +++++++++++++----- 2 files changed, 161 insertions(+), 51 deletions(-) diff --git a/playbooks/custom_functions/indicator_collect.json b/playbooks/custom_functions/indicator_collect.json index 68e8af34ca..a7a2cc7cf4 100644 --- a/playbooks/custom_functions/indicator_collect.json +++ b/playbooks/custom_functions/indicator_collect.json @@ -1,7 +1,7 @@ { - "create_time": "2021-11-30T14:20:41.840902+00:00", + "create_time": "2022-03-24T16:00:50.097247+00:00", "custom_function_id": "5febf154c78c6815119c08f9dfaba9a661a992d6", - "description": "Collect all indicators in a container and separate them by data type. Additional output data paths are created for each data type. Artifact scope is ignored. ", + "description": "Collect all indicators in a container and separate them by data type. Additional output data paths are created for each data type. Artifact scope is ignored.", "draft_mode": false, "inputs": [ { @@ -12,6 +12,43 @@ "input_type": "item", "name": "container", "placeholder": "container:id" + }, + { + "contains_type": [ + "phantom artifact id" + ], + "description": "Optional parameter to only look for indicator values that occur in the artifacts with these IDs. Must be one of: json serializable list, comma separated integers, or a single integer.", + "input_type": "list", + "name": "artifact_ids_include", + "placeholder": "artifact:*.id" + }, + { + "contains_type": [], + "description": "Optional parameter to only include indicators with at least one of the provided types in the output. If left empty, all indicator types will be included except those that are explicitly excluded. Accepts a comma-separated list.", + "input_type": "list", + "name": "indicator_types_include", + "placeholder": "ip, domain" + }, + { + "contains_type": [], + "description": "Optional parameter to exclude indicators with any of the provided types from the output. Accepts a comma-separated list.", + "input_type": "list", + "name": "indicator_types_exclude", + "placeholder": "ip, domain" + }, + { + "contains_type": [], + "description": "Optional parameter to only include indicators with at least one of the provided tags in the output. If left empty, tags will be ignored except when they are excluded. Accepts a comma-separated list.", + "input_type": "list", + "name": "indicator_tags_include", + "placeholder": "not_contained, malware" + }, + { + "contains_type": [], + "description": "Optional parameter to exclude indicators with any of the provided tags from the output. Accepts a comma-separated list.", + "input_type": "list", + "name": "indicator_tags_exclude", + "placeholder": "contained, not_malware" } ], "outputs": [ @@ -72,6 +109,6 @@ "description": "" } ], - "platform_version": "5.1.0.70187", + "platform_version": "5.2.1.78411", "python_version": "3" } \ No newline at end of file diff --git a/playbooks/custom_functions/indicator_collect.py b/playbooks/custom_functions/indicator_collect.py index 16be6f993c..247f40f996 100644 --- a/playbooks/custom_functions/indicator_collect.py +++ b/playbooks/custom_functions/indicator_collect.py @@ -1,9 +1,14 @@ -def indicator_collect(container=None, **kwargs): +def indicator_collect(container=None, artifact_ids_include=None, indicator_types_include=None, indicator_types_exclude=None, indicator_tags_include=None, indicator_tags_exclude=None, **kwargs): """ - Collect all indicators in a container and separate them by data type. Additional output data paths are created for each data type. Artifact scope is ignored. + Collect all indicators in a container and separate them by data type. Additional output data paths are created for each data type. Artifact scope is ignored. Args: container (CEF type: phantom container id): The current container + artifact_ids_include (CEF type: phantom artifact id): Optional parameter to only look for indicator values that occur in the artifacts with these IDs. Must be one of: json serializable list, comma separated integers, or a single integer. + indicator_types_include: Optional parameter to only include indicators with at least one of the provided types in the output. If left empty, all indicator types will be included except those that are explicitly excluded. Accepts a comma-separated list. + indicator_types_exclude: Optional parameter to exclude indicators with any of the provided types from the output. Accepts a comma-separated list. + indicator_tags_include: Optional parameter to only include indicators with at least one of the provided tags in the output. If left empty, tags will be ignored except when they are excluded. Accepts a comma-separated list. + indicator_tags_exclude: Optional parameter to exclude indicators with any of the provided tags from the output. Accepts a comma-separated list. Returns a JSON-serializable object that implements the configured data paths: all_indicators.*.cef_key @@ -20,30 +25,39 @@ def indicator_collect(container=None, **kwargs): ############################ Custom Code Goes Below This Line ################################# import json import phantom.rules as phantom + from hashlib import sha256 outputs = {'all_indicators': []} - data_types = [ - "domain", - "file name", - "file path", - "hash", - "host name", - "ip", - "mac address", - "md5", - "port", - "process name", - "sha1", - "sha256", - "sha512", - "url", - "user name", - "vault id" - ] - for data_type in data_types: - data_type_escaped = data_type.replace(' ', '_') - outputs[data_type_escaped] = [] + def grouper(seq, size): + return (seq[pos:pos + size] for pos in range(0, len(seq), size)) + + def get_indicator_json(value_set): + value_list = list(value_set) + indicator_url = phantom.build_phantom_rest_url('indicator') + '?page_size=0&timerange=all' + hashed_list = [sha256(item.encode('utf-8')).hexdigest() for item in value_list] + indicator_dictionary = {} + for group in grouper(hashed_list, 100): + query_url = indicator_url + f'&_filter_value_hash__in={group}' + indicator_response = phantom.requests.get(query_url, verify=False) + indicator_json = indicator_response.json() if indicator_response.status_code == 200 else {} + for data in indicator_json.get('data', []): + indicator_dictionary[data['value_hash']] = data + return indicator_dictionary + + def check_numeric_list(input_list): + return (all(isinstance(x, int) for x in input_list) or all(x.isnumeric() for x in input_list)) + + def is_valid_indicator(list_1=None, list_2=None, check_type="include"): + list_1 = [] if not list_1 else list_1 + list_2 = [] if not list_2 else list_2 + if check_type == 'exclude': + if list_1 and any(item in list_1 for item in list_2): + return False + elif check_type == 'include': + if list_1 and not any(item in list_1 for item in list_2): + return False + return True # validate container and get ID if isinstance(container, dict) and container['id']: @@ -52,40 +66,99 @@ def indicator_collect(container=None, **kwargs): elif isinstance(container, int): rest_container = phantom.requests.get(uri=phantom.build_phantom_rest_url('container', container), verify=False).json() if 'id' not in rest_container: - raise ValueError('Failed to find container with id {container}') + raise RuntimeError('Failed to find container with id {container}') container_dict = rest_container container_id = container else: - raise TypeError("The input 'container' is neither a container dictionary nor an int, so it cannot be used") - + raise TypeError("The input 'container' is neither a container dictionary nor a valid container id, so it cannot be used") + + if indicator_types_include: + indicator_types_include = [item.strip(' ') for item in indicator_types_include.split(',')] + if indicator_types_exclude: + indicator_types_exclude = [item.strip(' ') for item in indicator_types_exclude.split(',')] + if indicator_tags_include: + indicator_tags_include = [item.strip(' ').replace(' ', '_') for item in indicator_tags_include.split(',')] + if indicator_tags_exclude: + indicator_tags_exclude = [item.strip(' ').replace(' ', '_') for item in indicator_tags_exclude.split(',')] + + if artifact_ids_include: + # Try to convert to a valid list + if isinstance(artifact_ids_include, str) and artifact_ids_include.startswith('[') and artifact_ids_include.endswith(']'): + artifact_ids_include = json.loads(artifact_ids_include) + elif isinstance(artifact_ids_include, str): + artifact_ids_include = artifact_ids_include.replace(' ','').split(',') + elif isinstance(artifact_ids_include, int): + artifact_ids_include = [artifact_ids_include] + + # Check validity of list + if isinstance(artifact_ids_include, list) and not check_numeric_list(artifact_ids_include): + raise ValueError( + f"Invalid artifact_ids_include entered: '{artifact_ids_include}'. Must be a list of integers." + ) + + artifact_ids_include = [int(art_id) for art_id in artifact_ids_include] + + indicator_set = set() # fetch all artifacts in the container - artifacts = phantom.requests.get(uri=phantom.build_phantom_rest_url('container', container_id, 'artifacts'), params={'page_size': 0}, verify=False).json()['data'] + container_artifact_url = phantom.build_phantom_rest_url('artifact') + container_artifact_url += f'?_filter_container={container_id}&page_size=0&include_all_cef_types' + artifacts = phantom.requests.get(container_artifact_url, verify=False).json()['data'] for artifact in artifacts: artifact_id = artifact['id'] - for cef_key in artifact['cef']: - cef_value = artifact['cef'][cef_key] - params = {'indicator_value': cef_value, "_special_contains": True, 'page_size': 1} - indicator_data = phantom.requests.get(uri=phantom.build_phantom_rest_url('indicator_by_value'), params=params, verify=False) - if indicator_data.status_code == 200: - indicator_json = indicator_data.json() - data_types = [] - if indicator_json.get('id'): - data_types = indicator_json['_special_contains'] - # drop none - data_types = [item for item in data_types if item] + if (artifact_ids_include and artifact_id in artifact_ids_include) or not artifact_ids_include: + + for cef_key in artifact['cef']: + cef_value = artifact['cef'][cef_key] + data_types = artifact['cef_types'].get(cef_key, []) - # store the value in all_indicators and a list of values for each data type - outputs['all_indicators'].append({'cef_key': cef_key, 'cef_value': cef_value, 'artifact_id': artifact_id, 'data_types': data_types}) - for data_type in data_types: - # outputs will have underscores instead of spaces - data_type_escaped = data_type.replace(' ', '_') - if data_type_escaped not in outputs: - outputs[data_type_escaped] = [] - outputs[data_type_escaped].append({'cef_key': cef_key, 'cef_value': cef_value, 'artifact_id': artifact_id}) + # get indicator details if valid type + if ( + ( + is_valid_indicator(indicator_types_exclude, data_types, check_type='exclude') + and is_valid_indicator(indicator_types_include, data_types, check_type='include') + ) + and + ( + isinstance(cef_value, str) or isinstance(cef_value, bool) or isinstance(cef_value, int) or isinstance(cef_value, float) + ) + ): + indicator_set.add(str(cef_value)) + + indicator_dictionary = get_indicator_json(indicator_set) + for artifact in artifacts: + artifact_id = artifact['id'] + if (artifact_ids_include and artifact_id in artifact_ids_include) or not artifact_ids_include: + for cef_key in artifact['cef']: - # sort the all_indicators outputs to make them more consistent - outputs['all_indicators'].sort(key=lambda indicator: str(indicator['cef_value'])) + cef_value = artifact['cef'][cef_key] + cef_value_hash = sha256(str(cef_value).encode('utf-8')).hexdigest() + data_types = artifact['cef_types'].get(cef_key, []) + if indicator_dictionary.get(cef_value_hash): + + tags = indicator_dictionary[cef_value_hash]['tags'] + if ( + is_valid_indicator(indicator_tags_exclude, tags, check_type='exclude') + and is_valid_indicator(indicator_tags_include, tags, check_type='include') + ): + outputs['all_indicators'].append({ + 'cef_key': cef_key, + 'cef_value': cef_value, + 'artifact_id': artifact_id, + 'data_types': data_types, + 'tags': tags + }) + for data_type in data_types: + # outputs will have underscores instead of spaces + data_type_escaped = data_type.replace(' ', '_') + if data_type_escaped not in outputs: + outputs[data_type_escaped] = [] + outputs[data_type_escaped].append( + {'cef_key': cef_key, 'cef_value': cef_value, 'artifact_id': artifact_id, 'tags': tags} + ) + if outputs.get('all_indicators'): + # sort the all_indicators outputs to make them more consistent + outputs['all_indicators'].sort(key=lambda indicator: str(indicator['cef_value'])) # Return a JSON-serializable object assert json.dumps(outputs) # Will raise an exception if the :outputs: object is not JSON-serializable