diff --git a/bin/generate.py b/bin/generate.py index aed5cc18f5..81cb23e938 100644 --- a/bin/generate.py +++ b/bin/generate.py @@ -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..")