Modules for expansion services, import and export in MISP http://misp.github.io/misp-modules
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

118 lines
3.8 KiB

"""
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