mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
Branch was auto-updated.
This commit is contained in:
github-actions[bot]
+301
-364
@@ -67,7 +67,6 @@ def generate_collections_conf(lookups, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
def generate_savedsearches_conf(detections, response_tasks, baselines, deployments, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
'''
|
||||
@param detections: input list of individual YAML detections in detections/ directory
|
||||
@@ -77,6 +76,229 @@ def generate_savedsearches_conf(detections, response_tasks, baselines, deploymen
|
||||
@return: the savedsearches.conf file located in package/default/
|
||||
'''
|
||||
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
j2_env.filters['custom_jinja2_enrichment_filter'] = custom_jinja2_enrichment_filter
|
||||
template = j2_env.get_template('savedsearches.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/savedsearches.conf')
|
||||
output = template.render(detections=detections, baselines=baselines, response_tasks=response_tasks, time=utc_time)
|
||||
with open(output_path, 'w') as f:
|
||||
output = output.encode('ascii', 'ignore').decode('ascii')
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
def generate_analytic_story_conf(stories, detections, response_tasks, baselines, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('analytic_stories.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/analytic_stories.conf')
|
||||
output = template.render(stories=stories, time=utc_time)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
def generate_use_case_library_conf(stories, detections, response_tasks, baselines, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('use_case_library.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/use_case_library.conf')
|
||||
output = template.render(stories=stories, detections=detections,
|
||||
response_tasks=response_tasks,
|
||||
baselines=baselines, time=utc_time)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
def generate_macros_conf(macros, detections, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
filter_macros = []
|
||||
for detection in detections:
|
||||
new_dict = {}
|
||||
new_dict['definition'] = 'search *'
|
||||
new_dict['description'] = 'Update this macro to limit the output results to filter out false positives. '
|
||||
new_dict['name'] = detection['name']. \
|
||||
replace(' ', '_').replace('-', '_').replace('.', '_').replace('/', '_').lower() + '_filter'
|
||||
filter_macros.append(new_dict)
|
||||
|
||||
all_macros = macros + filter_macros
|
||||
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('macros.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/macros.conf')
|
||||
output = template.render(macros=all_macros, time=utc_time)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
def generate_workbench_panels(response_tasks, stories, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
workbench_panel_objects = []
|
||||
for response_task in response_tasks:
|
||||
if 'search' in response_task:
|
||||
if 'inputs' in response_task:
|
||||
response_file_name = response_task['name'].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
|
||||
response_file_name_xml = response_file_name + "___response_task.xml"
|
||||
response_task['lowercase_name'] = response_file_name
|
||||
workbench_panel_objects.append(response_task)
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('panel.j2')
|
||||
file_path = "default/data/ui/panels/workbench_panel_" + response_file_name_xml
|
||||
output_path = path.join(OUTPUT_PATH, file_path)
|
||||
response_task['search']= response_task['search'].replace(">",">")
|
||||
response_task['search']= response_task['search'].replace("<","<")
|
||||
|
||||
output = template.render(search=response_task['search'])
|
||||
with open(output_path, 'w') as f:
|
||||
f.write(output)
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('es_investigations.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/es_investigations.conf')
|
||||
output = template.render(response_tasks=workbench_panel_objects, stories=stories)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('workflow_actions.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/workflow_actions.conf')
|
||||
output = template.render(response_tasks=workbench_panel_objects)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return workbench_panel_objects
|
||||
|
||||
|
||||
def parse_data_models_from_search(search):
|
||||
match = re.search(r'from\sdatamodel\s?=\s?([^\s.]*)', search)
|
||||
if match is not None:
|
||||
return match.group(1)
|
||||
return False
|
||||
|
||||
def parse_author_company(story):
|
||||
match_author = re.search(r'^([^,]+)', story['author'])
|
||||
if match_author is None:
|
||||
match_author = 'no'
|
||||
else:
|
||||
match_author = match_author.group(1)
|
||||
|
||||
match_company = re.search(r',\s?(.*)$', story['author'])
|
||||
if match_company is None:
|
||||
match_company = 'no'
|
||||
else:
|
||||
match_company = match_company.group(1)
|
||||
|
||||
return match_author, match_company
|
||||
|
||||
|
||||
def get_deployments(object, deployments):
|
||||
matched_deployments = []
|
||||
|
||||
for deployment in deployments:
|
||||
|
||||
for tag in object['tags'].keys():
|
||||
if tag in deployment['tags'].keys():
|
||||
if type(object['tags'][tag]) is str:
|
||||
tag_array = [object['tags'][tag]]
|
||||
else:
|
||||
tag_array = object['tags'][tag]
|
||||
|
||||
for tag_value in tag_array:
|
||||
if type(deployment['tags'][tag]) is str:
|
||||
tag_array_deployment = [deployment['tags'][tag]]
|
||||
else:
|
||||
tag_array_deployment = deployment['tags'][tag]
|
||||
|
||||
for tag_value_deployment in tag_array_deployment:
|
||||
if tag_value == tag_value_deployment:
|
||||
# print("tag value: {}, matched deployment tag: {} on deployment: {}".format(tag_value,tag_value_deployment, deployment))
|
||||
matched_deployments.append(deployment)
|
||||
continue
|
||||
|
||||
# grab default for all stories if deployment not set
|
||||
if len(matched_deployments) == 0:
|
||||
for deployment in deployments:
|
||||
if 'analytic_story' in deployment['tags']:
|
||||
if deployment['tags']['analytic_story'] == 'all':
|
||||
last_deployment = deployment
|
||||
else:
|
||||
last_deployment = matched_deployments[-1]
|
||||
# last_deployment = replace_vars_in_deployment(last_deployment, object) # Not needed because of custom_jinja2_enrichment_filter
|
||||
|
||||
# print(last_deployment)
|
||||
return last_deployment
|
||||
|
||||
def get_nes_fields(search, deployment):
|
||||
nes_fields_matches = []
|
||||
if 'alert_action' in deployment:
|
||||
if 'notable' in deployment['alert_action']:
|
||||
if 'nes_fields' in deployment['alert_action']['notable']:
|
||||
for field in deployment['alert_action']['notable']['nes_fields']:
|
||||
if (search.find(field + ' ') != -1):
|
||||
nes_fields_matches.append(field)
|
||||
|
||||
return nes_fields_matches
|
||||
|
||||
|
||||
def map_response_tasks_to_stories(response_tasks):
|
||||
sto_res = {}
|
||||
for response_task in response_tasks:
|
||||
if 'tags' in response_task:
|
||||
if 'analytic_story' in response_task['tags']:
|
||||
for story in response_task['tags']['analytic_story']:
|
||||
if 'type' in response_task.keys():
|
||||
if response_task['type'] == 'response':
|
||||
task_name = str('ESCU - ' + response_task['name'] + ' - Response Task')
|
||||
else:
|
||||
task_name = str('ESCU - ' + response_task['name'] + ' - Response Task')
|
||||
if not (story in sto_res):
|
||||
sto_res[story] = {task_name}
|
||||
else:
|
||||
sto_res[story].add(task_name)
|
||||
return sto_res
|
||||
|
||||
def map_baselines_to_stories(baselines):
|
||||
sto_bas = {}
|
||||
for baseline in baselines:
|
||||
if 'tags' in baseline:
|
||||
if 'analytic_story' in baseline['tags']:
|
||||
for story in baseline['tags']['analytic_story']:
|
||||
if 'type' in baseline.keys():
|
||||
if baseline['type'] == 'batch':
|
||||
baseline_name = str('ESCU - ' + baseline['name'])
|
||||
else:
|
||||
baseline_name = str('ESCU - ' + baseline['name'])
|
||||
if not (story in sto_bas):
|
||||
sto_bas[story] = {baseline_name}
|
||||
else:
|
||||
sto_bas[story].add(baseline_name)
|
||||
return sto_bas
|
||||
|
||||
def custom_jinja2_enrichment_filter(string, object):
|
||||
customized_string = string
|
||||
for key in object.keys():
|
||||
[key.encode('utf-8') for key in object]
|
||||
customized_string = customized_string.replace("%" + key + "%", str(object[key]))
|
||||
|
||||
for key in object['tags'].keys():
|
||||
customized_string = customized_string.replace("%" + key + "%", str(object['tags'][key]))
|
||||
|
||||
return customized_string
|
||||
|
||||
|
||||
def prepare_detections(detections, deployments, OUTPUT_PATH):
|
||||
for detection in detections:
|
||||
# parse out data_models
|
||||
data_model = parse_data_models_from_search(detection['search'])
|
||||
@@ -127,8 +349,9 @@ def generate_savedsearches_conf(detections, response_tasks, baselines, deploymen
|
||||
if (OUTPUT_PATH) == 'dist/saaws':
|
||||
detection['disabled'] = 'false'
|
||||
|
||||
return detections
|
||||
|
||||
|
||||
def prepare_baselines(baselines, deployments, OUTPUT_PATH):
|
||||
for baseline in baselines:
|
||||
data_model = parse_data_models_from_search(baseline['search'])
|
||||
if data_model:
|
||||
@@ -139,332 +362,18 @@ def generate_savedsearches_conf(detections, response_tasks, baselines, deploymen
|
||||
matched_deployment = get_deployments(baseline, deployments)
|
||||
baseline['deployment'] = matched_deployment
|
||||
|
||||
return baselines
|
||||
|
||||
def prepare_response_tasks(response_tasks):
|
||||
for response_task in response_tasks:
|
||||
if 'search' in response_task:
|
||||
data_model = parse_data_models_from_search(response_task['search'])
|
||||
if data_model:
|
||||
response_task['data_model'] = data_model
|
||||
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
j2_env.filters['custom_jinja2_enrichment_filter'] = custom_jinja2_enrichment_filter
|
||||
template = j2_env.get_template('savedsearches.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/savedsearches.conf')
|
||||
output = template.render(detections=detections, baselines=baselines, response_tasks=response_tasks, time=utc_time)
|
||||
with open(output_path, 'w') as f:
|
||||
output = output.encode('ascii', 'ignore').decode('ascii')
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
def generate_analytic_story_conf(stories, detections, response_tasks, baselines, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
|
||||
sto_det = map_detection_to_stories(detections)
|
||||
|
||||
sto_res = map_response_tasks_to_stories(response_tasks)
|
||||
|
||||
sto_bas = map_baselines_to_stories(baselines)
|
||||
|
||||
for story in stories:
|
||||
if story['name'] in sto_det:
|
||||
story['detections'] = list(sto_det[story['name']])
|
||||
if story['name'] in sto_res:
|
||||
story['response_tasks'] = list(sto_res[story['name']])
|
||||
if story['name'] in sto_bas:
|
||||
story['baselines'] = list(sto_bas[story['name']])
|
||||
|
||||
stories = prepare_stories(stories, detections)
|
||||
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('analytic_stories.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/analytic_stories.conf')
|
||||
output = template.render(stories=stories, time=utc_time)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
def generate_use_case_library_conf(stories, detections, response_tasks, baselines, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
|
||||
sto_det = map_detection_to_stories(detections)
|
||||
|
||||
sto_res = map_response_tasks_to_stories(response_tasks)
|
||||
|
||||
for story in stories:
|
||||
story['author_name'], story['author_company'] = parse_author_company(story)
|
||||
if story['name'] in sto_det:
|
||||
story['detections'] = list(sto_det[story['name']])
|
||||
|
||||
if story['name'] in sto_res:
|
||||
|
||||
story['response_tasks'] = list(sto_res[story['name']])
|
||||
story['searches'] = story['detections'] + story['response_tasks']
|
||||
else:
|
||||
story['searches'] = story['detections']
|
||||
|
||||
for detection in detections:
|
||||
|
||||
keys = ['mitre_attack', 'kill_chain_phases', 'cis20', 'nist']
|
||||
mappings = {}
|
||||
for key in keys:
|
||||
if key == 'mitre_attack':
|
||||
if 'mitre_attack_id' in detection['tags']:
|
||||
mappings[key] = detection['tags']['mitre_attack_id']
|
||||
else:
|
||||
if key in detection['tags']:
|
||||
mappings[key] = detection['tags'][key]
|
||||
detection['mappings'] = mappings
|
||||
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('use_case_library.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/use_case_library.conf')
|
||||
output = template.render(stories=stories, detections=detections,
|
||||
response_tasks=response_tasks,
|
||||
baselines=baselines, time=utc_time)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
def generate_macros_conf(macros, detections, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
filter_macros = []
|
||||
for detection in detections:
|
||||
new_dict = {}
|
||||
new_dict['definition'] = 'search *'
|
||||
new_dict['description'] = 'Update this macro to limit the output results to filter out false positives. '
|
||||
new_dict['name'] = detection['name']. \
|
||||
replace(' ', '_').replace('-', '_').replace('.', '_').replace('/', '_').lower() + '_filter'
|
||||
filter_macros.append(new_dict)
|
||||
|
||||
all_macros = macros + filter_macros
|
||||
|
||||
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('macros.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/macros.conf')
|
||||
output = template.render(macros=all_macros, time=utc_time)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
def generate_workbench_panels(response_tasks, stories, TEMPLATE_PATH, OUTPUT_PATH):
|
||||
|
||||
sto_res = map_response_tasks_to_stories(response_tasks)
|
||||
|
||||
for story in stories:
|
||||
if story['name'] in sto_res:
|
||||
response_task_names = list(sto_res[story['name']])
|
||||
story['workbench_panels'] = []
|
||||
for response_task_name in response_task_names:
|
||||
str = 'panel://workbench_panel_' + response_task_name[7:].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
|
||||
story['workbench_panels'].append(str)
|
||||
story['lowercase_name'] = story['name'].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
|
||||
|
||||
workbench_panel_objects = []
|
||||
for response_task in response_tasks:
|
||||
if 'search' in response_task:
|
||||
if 'inputs' in response_task:
|
||||
response_file_name = response_task['name'].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
|
||||
response_file_name_xml = response_file_name + "___response_task.xml"
|
||||
response_task['lowercase_name'] = response_file_name
|
||||
workbench_panel_objects.append(response_task)
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('panel.j2')
|
||||
file_path = "default/data/ui/panels/workbench_panel_" + response_file_name_xml
|
||||
output_path = path.join(OUTPUT_PATH, file_path)
|
||||
response_task['search']= response_task['search'].replace(">",">")
|
||||
response_task['search']= response_task['search'].replace("<","<")
|
||||
|
||||
output = template.render(search=response_task['search'])
|
||||
with open(output_path, 'w') as f:
|
||||
f.write(output)
|
||||
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('es_investigations.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/es_investigations.conf')
|
||||
output = template.render(response_tasks=workbench_panel_objects, stories=stories)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
|
||||
trim_blocks=True)
|
||||
template = j2_env.get_template('workflow_actions.j2')
|
||||
output_path = path.join(OUTPUT_PATH, 'default/workflow_actions.conf')
|
||||
output = template.render(response_tasks=workbench_panel_objects)
|
||||
with open(output_path, 'w', encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
|
||||
return workbench_panel_objects
|
||||
|
||||
|
||||
def parse_data_models_from_search(search):
|
||||
match = re.search(r'from\sdatamodel\s?=\s?([^\s.]*)', search)
|
||||
if match is not None:
|
||||
return match.group(1)
|
||||
return False
|
||||
|
||||
|
||||
def parse_author_company(story):
|
||||
match_author = re.search(r'^([^,]+)', story['author'])
|
||||
if match_author is None:
|
||||
match_author = 'no'
|
||||
else:
|
||||
match_author = match_author.group(1)
|
||||
|
||||
match_company = re.search(r',\s?(.*)$', story['author'])
|
||||
if match_company is None:
|
||||
match_company = 'no'
|
||||
else:
|
||||
match_company = match_company.group(1)
|
||||
|
||||
return match_author, match_company
|
||||
|
||||
|
||||
def get_deployments(object, deployments):
|
||||
matched_deployments = []
|
||||
|
||||
for deployment in deployments:
|
||||
|
||||
for tag in object['tags'].keys():
|
||||
if tag in deployment['tags'].keys():
|
||||
if type(object['tags'][tag]) is str:
|
||||
tag_array = [object['tags'][tag]]
|
||||
else:
|
||||
tag_array = object['tags'][tag]
|
||||
|
||||
for tag_value in tag_array:
|
||||
if type(deployment['tags'][tag]) is str:
|
||||
tag_array_deployment = [deployment['tags'][tag]]
|
||||
else:
|
||||
tag_array_deployment = deployment['tags'][tag]
|
||||
|
||||
for tag_value_deployment in tag_array_deployment:
|
||||
if tag_value == tag_value_deployment:
|
||||
# print("tag value: {}, matched deployment tag: {} on deployment: {}".format(tag_value,tag_value_deployment, deployment))
|
||||
matched_deployments.append(deployment)
|
||||
continue
|
||||
|
||||
# grab default for all stories if deployment not set
|
||||
if len(matched_deployments) == 0:
|
||||
for deployment in deployments:
|
||||
if 'analytic_story' in deployment['tags']:
|
||||
if deployment['tags']['analytic_story'] == 'all':
|
||||
last_deployment = deployment
|
||||
else:
|
||||
last_deployment = matched_deployments[-1]
|
||||
last_deployment = replace_vars_in_deployment(last_deployment, object)
|
||||
|
||||
# print(last_deployment)
|
||||
return last_deployment
|
||||
|
||||
|
||||
def replace_vars_in_deployment(deployment, object):
|
||||
if 'alert_action' in deployment:
|
||||
if 'email' in deployment['alert_action']:
|
||||
deployment['alert_action']['email']['message']=re.sub(r'%([a-z_]+)%]', lambda x: object[x.group(1)], str(v))
|
||||
deployment['alert_action']['email']['subject']=re.sub(r'%([a-z_]+)%]', lambda x: object[x.group(1)], str(v))
|
||||
|
||||
if 'notable' in deployment:
|
||||
deployment['alert_action']['notable']['rule_description']=re.sub(r'%([a-z_]+)%]', lambda x: object[x.group(1)], str(v))
|
||||
deployment['alert_action']['notable']['rule_title']=re.sub(r'%([a-z_]+)%]', lambda x: object[x.group(1)], str(v))
|
||||
|
||||
return deployment
|
||||
|
||||
|
||||
def get_nes_fields(search, deployment):
|
||||
nes_fields_matches = []
|
||||
if 'alert_action' in deployment:
|
||||
if 'notable' in deployment['alert_action']:
|
||||
if 'nes_fields' in deployment['alert_action']['notable']:
|
||||
for field in deployment['alert_action']['notable']['nes_fields']:
|
||||
if (search.find(field + ' ') != -1):
|
||||
nes_fields_matches.append(field)
|
||||
|
||||
return nes_fields_matches
|
||||
|
||||
|
||||
def map_detection_to_stories(detections):
|
||||
sto_det = {}
|
||||
for detection in detections:
|
||||
if 'analytic_story' in detection['tags']:
|
||||
for story in detection['tags']['analytic_story']:
|
||||
if 'type' in detection.keys():
|
||||
if detection['type'] == 'batch':
|
||||
rule_name = str('ESCU - ' + detection['name'] + ' - Rule')
|
||||
else:
|
||||
rule_name = str('ESCU - ' + detection['name'] + ' - Rule')
|
||||
if not (story in sto_det):
|
||||
sto_det[story] = {rule_name}
|
||||
else:
|
||||
sto_det[story].add(rule_name)
|
||||
return sto_det
|
||||
|
||||
|
||||
def map_response_tasks_to_stories(response_tasks):
|
||||
sto_res = {}
|
||||
for response_task in response_tasks:
|
||||
if 'tags' in response_task:
|
||||
if 'analytic_story' in response_task['tags']:
|
||||
for story in response_task['tags']['analytic_story']:
|
||||
if 'type' in response_task.keys():
|
||||
if response_task['type'] == 'response':
|
||||
task_name = str('ESCU - ' + response_task['name'] + ' - Response Task')
|
||||
else:
|
||||
task_name = str('ESCU - ' + response_task['name'] + ' - Response Task')
|
||||
if not (story in sto_res):
|
||||
sto_res[story] = {task_name}
|
||||
else:
|
||||
sto_res[story].add(task_name)
|
||||
return sto_res
|
||||
|
||||
|
||||
def map_baselines_to_stories(baselines):
|
||||
sto_bas = {}
|
||||
for baseline in baselines:
|
||||
if 'tags' in baseline:
|
||||
if 'analytic_story' in baseline['tags']:
|
||||
for story in baseline['tags']['analytic_story']:
|
||||
if 'type' in baseline.keys():
|
||||
if baseline['type'] == 'batch':
|
||||
baseline_name = str('ESCU - ' + baseline['name'])
|
||||
else:
|
||||
baseline_name = str('ESCU - ' + baseline['name'])
|
||||
if not (story in sto_bas):
|
||||
sto_bas[story] = {baseline_name}
|
||||
else:
|
||||
sto_bas[story].add(baseline_name)
|
||||
return sto_bas
|
||||
|
||||
def custom_jinja2_enrichment_filter(string, object):
|
||||
customized_string = string
|
||||
for key in object.keys():
|
||||
[key.encode('utf-8') for key in object]
|
||||
customized_string = customized_string.replace("%" + key + "%", str(object[key]))
|
||||
|
||||
for key in object['tags'].keys():
|
||||
customized_string = customized_string.replace("%" + key + "%", str(object['tags'][key]))
|
||||
|
||||
return customized_string
|
||||
|
||||
|
||||
def prepare_stories(stories, detections):
|
||||
return response_tasks
|
||||
|
||||
def prepare_stories(stories, detections, response_tasks, baselines):
|
||||
# enrich stories with information from detections: data_models, mitre_ids, kill_chain_phases, nists
|
||||
sto_to_data_models = {}
|
||||
sto_to_mitre_attack_ids = {}
|
||||
@@ -521,8 +430,14 @@ def prepare_stories(stories, detections):
|
||||
else:
|
||||
sto_to_nists[story] = set(detection['tags']['nist'])
|
||||
|
||||
sto_res = map_response_tasks_to_stories(response_tasks)
|
||||
sto_bas = map_baselines_to_stories(baselines)
|
||||
|
||||
for story in stories:
|
||||
story['author_name'], story['author_company'] = parse_author_company(story)
|
||||
story['lowercase_name'] = story['name'].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
|
||||
story['detections'] = sorted(sto_to_det[story['name']])
|
||||
story['searches'] = story['detections']
|
||||
if story['name'] in sto_to_data_models:
|
||||
story['data_models'] = sorted(sto_to_data_models[story['name']])
|
||||
if story['name'] in sto_to_mitre_attack_ids:
|
||||
@@ -533,6 +448,16 @@ def prepare_stories(stories, detections):
|
||||
story['cis20'] = sorted(sto_to_ciss[story['name']])
|
||||
if story['name'] in sto_to_nists:
|
||||
story['nist'] = sorted(sto_to_nists[story['name']])
|
||||
if story['name'] in sto_res:
|
||||
story['response_tasks'] = sorted(list(sto_res[story['name']]))
|
||||
story['searches'] = story['searches'] + story['response_tasks']
|
||||
story['workbench_panels'] = []
|
||||
for response_task_name in story['response_tasks']:
|
||||
s = 'panel://workbench_panel_' + response_task_name[7:].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
|
||||
story['workbench_panels'].append(s)
|
||||
if story['name'] in sto_bas:
|
||||
story['baselines'] = sorted(list(sto_bas[story['name']]))
|
||||
|
||||
|
||||
keys = ['mitre_attack', 'kill_chain_phases', 'cis20', 'nist']
|
||||
mappings = {}
|
||||
@@ -572,29 +497,54 @@ def generate_mitre_lookup(OUTPUT_PATH):
|
||||
writer.writerows(csv_mitre_rows)
|
||||
|
||||
|
||||
def import_objects(VERBOSE, REPO_PATH):
|
||||
objects = {
|
||||
"stories": load_objects("stories/*.yml", VERBOSE, REPO_PATH),
|
||||
"macros": load_objects("macros/*.yml", VERBOSE, REPO_PATH),
|
||||
"lookups": load_objects("lookups/*.yml", VERBOSE, REPO_PATH),
|
||||
"baselines": load_objects("baselines/*.yml", VERBOSE, REPO_PATH),
|
||||
"responses": load_objects("responses/*.yml", VERBOSE, REPO_PATH),
|
||||
"response_tasks": load_objects("response_tasks/*.yml", VERBOSE, REPO_PATH),
|
||||
"deployments": load_objects("deployments/*.yml", VERBOSE, REPO_PATH),
|
||||
"detections": load_objects("detections/*/*.yml", VERBOSE, REPO_PATH),
|
||||
}
|
||||
objects["detections"].extend(load_objects("detections/*/*/*.yml", VERBOSE, REPO_PATH))
|
||||
|
||||
return objects
|
||||
|
||||
def compute_objects(objects, PRODUCT, OUTPUT_PATH):
|
||||
if PRODUCT == "SAAWS":
|
||||
objects["detections"] = [object for object in objects["detections"] if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
objects["stories"] = [object for object in objects["stories"] if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
objects["baselines"] = [object for object in objects["baselines"] if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
objects["response_tasks"] = [object for object in objects["response_tasks"] if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
|
||||
# only use ESCU detections to the configurations
|
||||
objects["detections"] = sorted(filter(lambda d: d['type'].lower() == 'batch', objects["detections"]), key=lambda d: d['name'])
|
||||
# only use ESCU stories to the configuration
|
||||
objects["stories"] = sorted(filter(lambda s: s['type'].lower() == 'batch', objects["stories"]), key=lambda s: s['name'])
|
||||
|
||||
objects["response_tasks"] = sorted(objects["response_tasks"], key=lambda i: i['name'])
|
||||
objects["baselines"] = sorted(objects["baselines"], key=lambda b: b['name'])
|
||||
objects["macros"] = sorted(objects["macros"], key=lambda m: m['name'])
|
||||
|
||||
objects["detections"] = prepare_detections(objects["detections"], objects["deployments"], OUTPUT_PATH)
|
||||
objects["baselines"] = prepare_baselines(objects["baselines"], objects["deployments"], OUTPUT_PATH)
|
||||
objects["response_tasks"] = prepare_response_tasks(objects["response_tasks"])
|
||||
objects["stories"] = prepare_stories(objects["stories"], objects["detections"], objects["response_tasks"], objects["baselines"])
|
||||
|
||||
return objects
|
||||
|
||||
def get_objects(REPO_PATH, OUTPUT_PATH, PRODUCT, VERBOSE):
|
||||
objects = import_objects(VERBOSE, REPO_PATH)
|
||||
objects = compute_objects(objects, PRODUCT, OUTPUT_PATH)
|
||||
return objects
|
||||
|
||||
def main(REPO_PATH, OUTPUT_PATH, PRODUCT, VERBOSE):
|
||||
|
||||
TEMPLATE_PATH = path.join(REPO_PATH, 'bin/jinja2_templates')
|
||||
|
||||
stories = load_objects("stories/*.yml", VERBOSE, REPO_PATH)
|
||||
macros = load_objects("macros/*.yml", VERBOSE, REPO_PATH)
|
||||
lookups = load_objects("lookups/*.yml", VERBOSE, REPO_PATH)
|
||||
baselines = load_objects("baselines/*.yml", VERBOSE, REPO_PATH)
|
||||
responses = load_objects("responses/*.yml", VERBOSE, REPO_PATH)
|
||||
response_tasks = load_objects("response_tasks/*.yml", VERBOSE, REPO_PATH)
|
||||
deployments = load_objects("deployments/*.yml", VERBOSE, REPO_PATH)
|
||||
|
||||
# process all detections
|
||||
detections = []
|
||||
detections = load_objects("detections/*/*.yml", VERBOSE, REPO_PATH)
|
||||
detections.extend(load_objects("detections/*/*/*.yml", VERBOSE, REPO_PATH))
|
||||
|
||||
if PRODUCT == "SAAWS":
|
||||
detections = [object for object in detections if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
stories = [object for object in stories if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
baselines = [object for object in baselines if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
response_tasks = [object for object in response_tasks if 'Splunk Security Analytics for AWS' in object['tags']['product']]
|
||||
objects = get_objects(REPO_PATH, OUTPUT_PATH, PRODUCT, VERBOSE)
|
||||
|
||||
try:
|
||||
if VERBOSE:
|
||||
@@ -604,39 +554,26 @@ def main(REPO_PATH, OUTPUT_PATH, PRODUCT, VERBOSE):
|
||||
print('Error: ' + str(e))
|
||||
print("WARNING: Generation of Mitre lookup failed.")
|
||||
|
||||
lookups_path = generate_transforms_conf(lookups, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
lookups_path = generate_collections_conf(lookups, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
lookups_path = generate_transforms_conf(objects["lookups"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
lookups_path = generate_collections_conf(objects["lookups"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
detections = sorted(detections, key=lambda d: d['name'])
|
||||
detection_path = generate_savedsearches_conf(objects["detections"], objects["response_tasks"], objects["baselines"], objects["deployments"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
# only use ESCU detections to the configurations
|
||||
detections = [object for object in detections if object["type"].lower() == "batch"]
|
||||
story_path = generate_analytic_story_conf(objects["stories"], objects["detections"], objects["response_tasks"], objects["baselines"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
response_tasks = sorted(response_tasks, key=lambda i: i['name'])
|
||||
baselines = sorted(baselines, key=lambda b: b['name'])
|
||||
detection_path = generate_savedsearches_conf(detections, response_tasks, baselines, deployments, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
use_case_lib_path = generate_use_case_library_conf(objects["stories"], objects["detections"], objects["response_tasks"], objects["baselines"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
# only use ESCU stories to the configuration
|
||||
stories = sorted(filter(lambda s: s['type'].lower() == 'batch', stories), key=lambda s: s['name'])
|
||||
macros_path = generate_macros_conf(objects["macros"], objects["detections"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
|
||||
|
||||
story_path = generate_analytic_story_conf(stories, detections, response_tasks, baselines, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
use_case_lib_path = generate_use_case_library_conf(stories, detections, response_tasks, baselines, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
macros = sorted(macros, key=lambda m: m['name'])
|
||||
macros_path = generate_macros_conf(macros, detections, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
workbench_panels_objects = generate_workbench_panels(response_tasks, stories, TEMPLATE_PATH, OUTPUT_PATH)
|
||||
workbench_panels_objects = generate_workbench_panels(objects["response_tasks"], objects["stories"], TEMPLATE_PATH, OUTPUT_PATH)
|
||||
|
||||
|
||||
if VERBOSE:
|
||||
print("{0} stories have been successfully written to {1}".format(len(stories), story_path))
|
||||
print("{0} detections have been successfully written to {1}".format(len(detections), detection_path))
|
||||
print("{0} response tasks have been successfully written to {1}".format(len(response_tasks), detection_path))
|
||||
print("{0} baselines have been successfully written to {1}".format(len(baselines), detection_path))
|
||||
print("{0} macros have been successfully written to {1}".format(len(macros), macros_path))
|
||||
print("{0} stories have been successfully written to {1}".format(len(objects["stories"]), story_path))
|
||||
print("{0} detections have been successfully written to {1}".format(len(objects["detections"]), detection_path))
|
||||
print("{0} response tasks have been successfully written to {1}".format(len(objects["response_tasks"]), detection_path))
|
||||
print("{0} baselines have been successfully written to {1}".format(len(objects["baselines"]), detection_path))
|
||||
print("{0} macros have been successfully written to {1}".format(len(objects["macros"]), macros_path))
|
||||
print("{0} workbench panels have been successfully written to {1}, {2} and {3}".format(len(workbench_panels_objects), OUTPUT_PATH + "/default/es_investigations.conf", OUTPUT_PATH + "/default/workflow_actions.conf", OUTPUT_PATH + "/default/data/ui/panels/*"))
|
||||
print("security content generation completed..")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user