admin/splunk-security_content
1
0
Fork 0
mirror of https://github.com/splunk/security_content synced 2026-06-08 17:32:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2126 from splunk/TR-1819

Browse Source 
Updated indicator_collect CF
This commit is contained in:
Bhavin Patel
2022-03-28 17:27:48 -07:00
committed by GitHub
parent 6fdb694d1c de59745959
commit 46e9b2ce44
2 changed files with 161 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
playbooks/custom_functions/indicator_collect.json
+40 -3
View File
@@ -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"
}
playbooks/custom_functions/indicator_collect.py
+121 -48
View File
@@ -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