diff --git a/misp_modules/modules/export_mod/threat_connect_export.py b/misp_modules/modules/export_mod/threat_connect_export.py new file mode 100644 index 0000000..0b51fb7 --- /dev/null +++ b/misp_modules/modules/export_mod/threat_connect_export.py @@ -0,0 +1,118 @@ +""" +Export module for converting MISP events into ThreatConnect Structured Import files. This export data is meant to be used with the "Structured Import" ability of ThreatConnect. + +Source: http://kb.threatconnect.com/customer/en/portal/articles/1912599-using-structured-import/ +Source: http://kb.threatconnect.com/customer/en/portal/articles/2092925-the-threatconnect-data-model/ +""" +import base64 +import csv +import io +import json +import logging + +misperrors = {"error": "Error"} + +moduleinfo = { + "version": "0.1", + "author": "CenturyLink CIRT", + "description": "Export a structured CSV file for uploading to ThreatConnect", + "module-type": ["export"] +} + +# config fields expected from the MISP administrator +# Default_Source: The source of the data. Typically this won't be changed from the default +moduleconfig = ["Default_Source"] + +# Map of MISP fields => ThreatConnect fields +fieldmap = { + "domain": "Host", + "domain|ip": "Host|Address", + "hostname": "Host", + "ip-src": "Address", + "ip-dst": "Address", + "ip-src|port": "Address", + "ip-dst|port": "Address", + "whois-registrant-email": "EmailAddress", + "email-src": "EmailAddress", + "email-dst": "EmailAddress", + "url": "URL", + "md5": "File", + "filename|md5": "File" +} + +# combine all the MISP fields from fieldmap into one big list +mispattributes = { + "input": list(fieldmap.keys()) +} + + +def handler(q=False): + """ + Convert a MISP query into a CSV file matching the ThreatConnect Structured Import file format. + Input + q: Query dictionary + """ + if q is False or not q: + return False + + # Check if we were given a configuration + request = json.loads(q) + config = request.get("config", {"Default_Source": ""}) + logging.info("Setting config to: %s", config) + + response = io.StringIO() + writer = csv.DictWriter(response, fieldnames=["Type", "Value", "Source", "Description"]) + writer.writeheader() + + # start parsing MISP data + for event in request["data"]: + for attribute in event["Attribute"]: + if attribute["type"] in mispattributes["input"]: + logging.debug("Adding %s to structured CSV export of ThreatConnectExport", attribute["value"]) + if "|" in attribute["type"]: + # if the attribute type has multiple values, line it up with the corresponding ThreatConnect values in fieldmap + indicators = tuple(attribute["value"].split("|")) + tc_types = tuple(fieldmap[attribute["type"]].split("|")) + for i, indicator in enumerate(indicators): + writer.writerow({ + "Type": tc_types[i], + "Value": indicator, + "Source": config["Default_Source"], + "Description": attribute["comment"] + }) + else: + writer.writerow({ + "Type": fieldmap[attribute["type"]], + "Value": attribute["value"], + "Source": config["Default_Source"], + "Description": attribute["comment"] + }) + + return {"response": [], "data": str(base64.b64encode(bytes(response.getvalue(), 'utf-8')), 'utf-8')} + + +def introspection(): + """ + Relay the supported attributes to MISP. + No Input + Output + Dictionary of supported MISP attributes + """ + modulesetup = { + "responseType": "application/txt", + "outputFileExtension": "csv", + "userConfig": {}, + "inputSource": [] + } + return modulesetup + + +def version(): + """ + Relay module version and associated metadata to MISP. + No Input + Output + moduleinfo: metadata output containing all potential configuration values + """ + moduleinfo["config"] = moduleconfig + return moduleinfo