#!/usr/bin/python import sys import argparse import json import glob import yaml import os import csv from os import path from stix2 import FileSystemSource from stix2 import Filter VERSION = "3.0" NAME = "Detection Priority by Threat Actors" DESCRIPTION = "security-content detection priorty by common techniques used from threat actors" DOMAIN = "mitre-enterprise" def main(argv): # parse input variables parser = argparse.ArgumentParser(description='Detection Priority based on APT groups') parser.add_argument('-p','--projects_path', default='.', action='store', metavar='N', help='folder containing the projects Mitre Cyber Threat Intelligence Repository, Security Content and Sigma') parser.add_argument('-o','--output', default='output', action='store', help='result output directory, defaults to output') cmdargs = parser.parse_args() print("get all techniques for group") techniques, all_techniques = get_all_techniques_for_groups(cmdargs.projects_path) print("count techniques") counted_techniques, max_count = count_techniques(techniques, all_techniques) print("load detections techniques") detections = load_objects(path.join(cmdargs.projects_path),'detections/*.yml') print("get matched techniques") matched_techniques = get_matched_techniques(counted_techniques, detections) print("generate navigator layer") generate_navigator_layer(matched_techniques, max_count, cmdargs.output) print("generate csv file") generate_csv_file(matched_techniques, cmdargs.output) def count_techniques(techniques, all_techniques): counted_techniques = [] max_count = 0 actors = [] for all_technique in all_techniques: count_technique = sum(t['name'] == all_technique['name'] for t in techniques) if count_technique > 0: counted_techniques.append({'name': all_technique['name'], 'object': all_technique, 'count': count_technique}) max_count = count_technique if count_technique > max_count else max_count sum_techniques = [] for all_technique in all_techniques: for t in counted_techniques: if t['name'] == all_technique['name']: print(t["object"]["external_references"][0]["external_id"]) counted_techniques = sorted(counted_techniques, key = lambda i: i['count'], reverse=True) return counted_techniques, max_count def get_all_techniques_for_groups(projects_path): path_cti = path.join(projects_path,'cti/enterprise-attack') fs = FileSystemSource(path_cti) all_techniques = get_all_techniques(fs) techniques = [] groups = get_all_groups(fs) for group_obj in groups: techniques.extend(get_technique_by_group(fs, group_obj)) # ONLY FOR TESTING if len(techniques) > 10: return techniques, all_techniques return techniques, all_techniques def get_all_techniques(src): filt = [Filter('type', '=', 'attack-pattern')] return src.query(filt) def get_all_groups(src): filt = [Filter('type', '=', 'intrusion-set')] return src.query(filt) def get_technique_by_group(src, stix_id): relations = src.relationships(stix_id, 'uses', source_only=True) return src.query([ Filter('type', '=', 'attack-pattern'), Filter('id', 'in', [r.target_ref for r in relations]) ]) def get_matched_techniques(counted_techniques, detections): matched_techniques = [] for technique in counted_techniques: matched_splunk_detections = [] # find detections from Splunks security content # https://github.com/splunk/security-content for detection in detections: if 'mitre_attack_id' in detection['object']['tags']: for mitreid in detection['object']['tags']['mitre_attack_id']: if mitreid == technique["object"]["external_references"][0]["external_id"]: matched_splunk_detections.append(detection) matched_techniques.append({ "name": technique["name"], "ID": technique["object"]["external_references"][0]["external_id"], # substract the amount of detections we have from the score "score": technique["count"] - len(matched_splunk_detections), "splunk_rules": matched_splunk_detections, }) return matched_techniques def generate_navigator_layer(matched_techniques, max_count, output): # Base ATT&CK Navigator layer layer_json = { "version": VERSION, "name": NAME, "description": DESCRIPTION, "domain": DOMAIN, "techniques": [] } for technique in matched_techniques: comments = [] if len(technique["splunk_rules"]) > 0: for splunk_rule in technique["splunk_rules"]: comments.append("https://github.com/splunk/security-content/blob/develop/detections/" + splunk_rule['filename']) layer_technique = { "techniqueID": technique["ID"], "score" : technique["score"] } if len(comments) > 0: layer_technique["comment"] = "\n\n".join(comments) layer_json["techniques"].append(layer_technique) # add a color gradient (white -> red) to layer # ranging from zero (white) to the maximum score in the file (red) layer_json["gradient"] = { "colors": [ "#66b1ff", "#ff66f4", "#ff6666" ], "minValue": 0, "maxValue": max_count } layer_json["filters"] = { "platforms": ["Windows", "Linux", "macOS", "AWS", "GCP", "Azure", "Office 365", "SaaS" ] } layer_json["legendItems"] = [ { "label": "Low Priority", "color": "#66b1ff" }, { "label": "Medium Priority", "color": "#ff66f4" }, { "label": "High Priority", "color": "#ff6666" } ] layer_json['showTacticRowBackground'] = True layer_json['tacticRowBackground'] = "#dddddd" # output JSON with open(output + '/detections.json', 'w') as f: json.dump(layer_json, f, indent=4) # print("Mitre ATT&CK Navigator overlay was successfully written to output/detections.json") def generate_csv_file(matched_techniques, output): security_content_url = 'https://github.com/splunk/security-content/blob/develop/detections/' with open(output + '/detections.csv', 'w') as f: writer = csv.writer(f) writer.writerow(['Name', 'Technique ID', 'Detection Available', 'Link', 'score']) for technique in matched_techniques: if len(technique['splunk_rules']) > 0: for splunk_rule in technique["splunk_rules"]: writer.writerow([technique["name"],technique["ID"], "Yes", \ security_content_url + splunk_rule["filename"], technique['score']]) else: writer.writerow([technique["name"],technique["ID"], "No", \ "-", technique['score']]) # print("Recommended detections were successfully written to output/detections.csv") def load_objects(security_content_path, file_path): files = [] detection_files = path.join(path.expanduser(security_content_path), file_path) for file in glob.glob(detection_files): files.append({ "filename": os.path.basename(file), "object": load_file(file) }) return files def load_file(file_path): with open(file_path, 'r') as stream: try: file = list(yaml.safe_load_all(stream))[0] except yaml.YAMLError as exc: sys.exit("ERROR: reading {0}".format(file_path)) return file if __name__ == "__main__": main(sys.argv)