MISP
/
misp-modules
mirror of https://github.com/MISP/misp-modules
2
0
Fork 0
Code Issues Releases Wiki Activity

Merge pull request #133 from CenturyLinkCIRT/master 

ThreatConnect export module
pull/135/head
Alexandre Dulaunoy 2017-08-06 20:30:35 +02:00 committed by GitHub
parent 4c2cda9903 72c52da7ed
commit 7695dd1b62
4 changed files with 476 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
misp_modules/modules/export_mod/__init__.py
View File

@ -1 +1 @@
__all__ = ['testexport','cef_export','liteexport']
__all__ = ['testexport','cef_export','liteexport','threat_connect_export']

118
misp_modules/modules/export_mod/threat_connect_export.py Normal file
View File

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

297
tests/test_files/misp_event.json Normal file
View File

@ -0,0 +1,297 @@
{
"Event": {
"id": "625",
"orgc_id": "2",
"org_id": "1",
"date": "2017-05-24",
"threat_level_id": "3",
"info": "M2M - Fwd: IMG_3428.pdf",
"published": false,
"uuid": "59259036-fcd0-4749-8a6c-4d88950d210f",
"attribute_count": "2",
"analysis": "1",
"timestamp": "1500496265",
"distribution": "3",
"proposal_email_lock": false,
"user_id": "1",
"locked": false,
"publish_timestamp": "0",
"sharing_group_id": "0",
"disable_correlation": false
},
"User": {
"email": "admin@misp.training",
"id": "1"
},
"ThreatLevel": {
"name": "Low",
"id": "3"
},
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "2",
"name": "CIRCL",
"uuid": "55f6ea5e-2c60-40e5-964f-47a8950d210f"
},
"Attribute": [{
"id": "157835",
"type": "attachment",
"category": "Artifacts dropped",
"to_ids": false,
"uuid": "59259037-1014-4669-96b1-46af950d210f",
"event_id": "625",
"distribution": "5",
"timestamp": "1495633975",
"comment": "IMG_3428.pdf",
"sharing_group_id": "0",
"deleted": false,
"disable_correlation": false,
"value": "tmpzuni0skf",
"AttributeTag": [],
"ShadowAttribute": []
}, {
"id": "164191",
"type": "domain|ip",
"category": "Network activity",
"to_ids": false,
"uuid": "59430251-e6a4-4900-b78b-060dc0a83832",
"event_id": "625",
"distribution": "5",
"timestamp": "1497563729",
"comment": "Test data",
"sharing_group_id": "0",
"deleted": false,
"disable_correlation": false,
"value": "google.com|127.0.0.1",
"AttributeTag": [],
"ShadowAttribute": []
}],
"ShadowAttribute": [],
"EventTag": [{
"id": "1482",
"event_id": "625",
"tag_id": "2",
"Tag": {
"id": "2",
"name": "tlp:white",
"colour": "#ffffff",
"exportable": true,
"org_id": "0",
"hide_tag": false
}
}],
"Galaxy": [],
"RelatedEvent": [{
"Event": {
"id": "226",
"date": "2015-11-05",
"threat_level_id": "4",
"info": "OSINT Expansion on Systematic cyber attacks against Israeli and Palestinian targets going on for a year by Norman",
"published": true,
"uuid": "563b3ea6-b26c-401f-a68b-4d84950d210b",
"analysis": "2",
"timestamp": "1487757679",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}, {
"Event": {
"id": "207",
"date": "2015-04-03",
"threat_level_id": "4",
"info": "OSINT The Dyre Wolf report from IBM",
"published": true,
"uuid": "551e8745-ace0-461c-b9eb-ce36950d210b",
"analysis": "2",
"timestamp": "1428070986",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}, {
"Event": {
"id": "209",
"date": "2015-01-26",
"threat_level_id": "2",
"info": "OSINT I Know You Want Me - Unplugging PlugX from Takahiro Haruyama & Hiroshi Suzuki Black Hat Asia 2014 presentation",
"published": true,
"uuid": "54c60f43-b084-453a-a162-4e08950d210b",
"analysis": "2",
"timestamp": "1422356942",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}, {
"Event": {
"id": "214",
"date": "2014-12-18",
"threat_level_id": "4",
"info": "Expansion on two IPs listed in OSINT IOCs from various campaigns listed in Detecting Bleeding Edge Malware presentation at hack.lu 2014",
"published": true,
"uuid": "54932a3e-7284-4753-b95c-4e08950d210b",
"analysis": "2",
"timestamp": "1442489489",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}, {
"Event": {
"id": "208",
"date": "2014-11-20",
"threat_level_id": "4",
"info": "Import of CitizenLab public DB of malware indicators",
"published": true,
"uuid": "546e08ce-3134-4892-997b-73ff950d210b",
"analysis": "2",
"timestamp": "1487758220",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}, {
"Event": {
"id": "373",
"date": "2014-11-18",
"threat_level_id": "4",
"info": "OSINT Expansion on Additional indicators relating to Sofacy (APT28) phishing blog post by PWC",
"published": true,
"uuid": "546bc3e8-d498-4e0c-b169-f2ea950d210b",
"analysis": "2",
"timestamp": "1487758281",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}, {
"Event": {
"id": "230",
"date": "2014-10-02",
"threat_level_id": "3",
"info": "OSINT ShellShock scanning IPs from OpenDNS",
"published": true,
"uuid": "542e4c9c-cadc-4f8f-bb11-6d13950d210b",
"analysis": "2",
"timestamp": "1442489604",
"distribution": "3",
"org_id": "1",
"orgc_id": "3",
"Org": {
"id": "1",
"name": "MISP",
"uuid": "56ef3277-1ad4-42f6-b90b-04e5c0a83832"
},
"Orgc": {
"id": "3",
"name": "CthulhuSPRL.be",
"uuid": "55f6ea5f-fd34-43b8-ac1d-40cb950d210f"
}
}
}],
"RelatedAttribute": {
"164191": [{
"id": "207",
"org_id": "1",
"info": "OSINT The Dyre Wolf report from IBM",
"value": "google.com"
}, {
"id": "208",
"org_id": "1",
"info": "Import of CitizenLab public DB of malware indicators",
"value": "127.0.0.1"
}, {
"id": "209",
"org_id": "1",
"info": "OSINT I Know You Want Me - Unplugging PlugX from Takahiro Haruyama & Hiroshi Suzuki Black Hat Asia 2014 presentation",
"value": "127.0.0.1"
}, {
"id": "214",
"org_id": "1",
"info": "Expansion on two IPs listed in OSINT IOCs from various campaigns listed in Detecting Bleeding Edge Malware presentation at hack.lu 2014",
"value": "127.0.0.1"
}, {
"id": "226",
"org_id": "1",
"info": "OSINT Expansion on Systematic cyber attacks against Israeli and Palestinian targets going on for a year by Norman",
"value": "127.0.0.1"
}, {
"id": "230",
"org_id": "1",
"info": "OSINT ShellShock scanning IPs from OpenDNS",
"value": "127.0.0.1"
}, {
"id": "373",
"org_id": "1",
"info": "OSINT Expansion on Additional indicators relating to Sofacy (APT28) phishing blog post by PWC",
"value": "127.0.0.1"
}]
},
"RelatedShadowAttribute": [],
"Sighting": []
}

60
tests/threatconnect_export_test.py Normal file
View File

@ -0,0 +1,60 @@
"""Test module for the ThreatConnect Export module"""
import base64
import csv
import io
import json
import os
import unittest
import requests
class TestModules(unittest.TestCase):
"""Unittest module for threat_connect_export.py"""
def setUp(self):
self.headers = {'Content-Type': 'application/json'}
self.url = "http://127.0.0.1:6666/"
self.module = "threat_connect_export"
input_event_path = "%s/test_files/misp_event.json" % os.path.dirname(os.path.realpath(__file__))
with open(input_event_path, "r") as ifile:
self.event = json.load(ifile)
def test_01_introspection(self):
"""Taken from test.py"""
try:
response = requests.get(self.url + "modules")
modules = [module["name"] for module in response.json()]
assert self.module in modules
finally:
response.connection.close()
def test_02_export(self):
"""Test an event export"""
test_source = "Test Export"
query = {
"module": self.module,
"data": [self.event],
"config": {
"Default_Source": test_source
}
}
try:
response = requests.post(self.url + "query", headers=self.headers, data=json.dumps(query))
data = base64.b64decode(response.json()["data"]).decode("utf-8")
csvfile = io.StringIO(data)
reader = csv.DictReader(csvfile)
values = [field["Value"] for field in reader]
assert "google.com" in values
assert "127.0.0.1" in values
# resetting file pointer to read through again and extract sources
csvfile.seek(0)
# use a set comprehension to deduplicate sources
sources = {field["Source"] for field in reader}
assert test_source in sources
finally:
response.connection.close()
if __name__ == "__main__":
unittest.main()