mirror of https://github.com/MISP/misp-modules
Christian Studer
GitHub
commit
73302bef1e
|
|
@ -1444,11 +1444,10 @@ Module to export a structured CSV file for uploading to ThreatConnect.
|
|||
Module to import MISP attributes from a csv file.
|
||||
- **features**:
|
||||
>In order to parse data from a csv file, a header is required to let the module know which column is matching with known attribute fields / MISP types.
|
||||
>This header is part of the configuration of the module and should be filled out in MISP plugin settings, each field separated by COMMAS. Fields that do not match with any type known in MISP can be ignored in import, using a space or simply nothing between two separators (example: 'ip-src, , comment, ').
|
||||
>There is also one type that is confused and can be either a MISP attribute type or an attribute field: 'comment'. In this case, using 'attrComment' specifies that the attribute field 'comment' should be considered, otherwise it will be considered as the MISP attribute type.
|
||||
>
|
||||
>For each MISP attribute type, an attribute is created.
|
||||
>Attribute fields that are imported are the following: value, type, category, to-ids, distribution, comment, tag.
|
||||
>This header either comes from the csv file itself or is part of the configuration of the module and should be filled out in MISP plugin settings, each field separated by COMMAS. Fields that do not match with any type known in MISP or are not MISP attribute fields should be ignored in import, using a space or simply nothing between two separators (example: 'ip-src, , comment, ').
|
||||
>
|
||||
>If the csv file already contains a header that does not start by a '#', you should tick the checkbox 'has_header' to avoid importing it and have potential issues. You can also redefine the header even if it is already contained in the file, by following the rules for headers explained earlier. One reason why you would redefine a header is for instance when you want to skip some fields, or some fields are not valid types.
|
||||
- **input**:
|
||||
>CSV format file.
|
||||
- **output**:
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"description": "Module to import MISP attributes from a csv file.",
|
||||
"requirements": ["PyMISP"],
|
||||
"features": "In order to parse data from a csv file, a header is required to let the module know which column is matching with known attribute fields / MISP types.\nThis header is part of the configuration of the module and should be filled out in MISP plugin settings, each field separated by COMMAS. Fields that do not match with any type known in MISP can be ignored in import, using a space or simply nothing between two separators (example: 'ip-src, , comment, ').\nThere is also one type that is confused and can be either a MISP attribute type or an attribute field: 'comment'. In this case, using 'attrComment' specifies that the attribute field 'comment' should be considered, otherwise it will be considered as the MISP attribute type.\n\nFor each MISP attribute type, an attribute is created.\nAttribute fields that are imported are the following: value, type, category, to-ids, distribution, comment, tag.",
|
||||
"features": "In order to parse data from a csv file, a header is required to let the module know which column is matching with known attribute fields / MISP types.\n\nThis header either comes from the csv file itself or is part of the configuration of the module and should be filled out in MISP plugin settings, each field separated by COMMAS. Fields that do not match with any type known in MISP or are not MISP attribute fields should be ignored in import, using a space or simply nothing between two separators (example: 'ip-src, , comment, ').\n\nIf the csv file already contains a header that does not start by a '#', you should tick the checkbox 'has_header' to avoid importing it and have potential issues. You can also redefine the header even if it is already contained in the file, by following the rules for headers explained earlier. One reason why you would redefine a header is for instance when you want to skip some fields, or some fields are not valid types.",
|
||||
"references": ["https://tools.ietf.org/html/rfc4180", "https://tools.ietf.org/html/rfc7111"],
|
||||
"input": "CSV format file.",
|
||||
"output": "MISP Event attributes"
|
||||
|
|
|
|||
|
|
@ -8,220 +8,259 @@ import os
|
|||
import base64
|
||||
|
||||
misperrors = {'error': 'Error'}
|
||||
moduleinfo = {'version': '0.1', 'author': 'Christian Studer',
|
||||
moduleinfo = {'version': '0.2', 'author': 'Christian Studer',
|
||||
'description': 'Import Attributes from a csv file.',
|
||||
'module-type': ['import']}
|
||||
moduleconfig = []
|
||||
userConfig = {'header': {
|
||||
'type': 'String',
|
||||
'message': 'Define the header of the csv file, with types (included in MISP attribute types or attribute fields) separated by commas.\nFor fields that do not match these types or that you want to skip, please use space or simply nothing between commas.\nFor instance: ip-src,domain, ,timestamp'},
|
||||
userConfig = {
|
||||
'header': {
|
||||
'type': 'String',
|
||||
'message': 'Define the header of the csv file, with types (included in MISP attribute types or attribute fields) separated by commas.\nFor fields that do not match these types or that you want to skip, please use space or simply nothing between commas.\nFor instance: ip-src,domain, ,timestamp'},
|
||||
'has_header': {
|
||||
'type': 'Boolean',
|
||||
'message': 'Tick this box ONLY if there is a header line, NOT COMMENTED, and all the fields of this header are respecting the recommendations above.'
|
||||
}}
|
||||
'message': 'Tick this box ONLY if there is a header line, NOT COMMENTED, and all the fields of this header are respecting the recommendations above.'},
|
||||
'special_delimiter': {
|
||||
'type': 'String',
|
||||
'message': 'IF THE DELIMITERS ARE NOT COMMAS, please specify which ones are used (for instance: ";", "|", "/", "\t" for tabs, etc).'
|
||||
}
|
||||
}
|
||||
mispattributes = {'userConfig': userConfig, 'inputSource': ['file'], 'format': 'misp_standard'}
|
||||
|
||||
duplicatedFields = {'mispType': {'mispComment': 'comment'},
|
||||
'attrField': {'attrComment': 'comment'}}
|
||||
attributesFields = ['type', 'value', 'category', 'to_ids', 'comment', 'distribution']
|
||||
misp_standard_csv_header = ['uuid', 'event_id', 'category', 'type', 'value', 'comment', 'to_ids', 'date',
|
||||
'object_relation', 'attribute_tag', 'object_uuid', 'object_name', 'object_meta_category']
|
||||
misp_context_additional_fields = ['event_info', 'event_member_org', 'event_source_org', 'event_distribution',
|
||||
'event_threat_level_id', 'event_analysis', 'event_date', 'event_tag']
|
||||
misp_extended_csv_header = misp_standard_csv_header + misp_context_additional_fields
|
||||
delimiters = [',', ';', '|', '/', '\t', ' ']
|
||||
|
||||
|
||||
class CsvParser():
|
||||
def __init__(self, header, has_header, data):
|
||||
data_header = data[0]
|
||||
def __init__(self, header, has_header, delimiter, data, from_misp, MISPtypes):
|
||||
self.misp_event = MISPEvent()
|
||||
if data_header == misp_standard_csv_header or data_header == misp_extended_csv_header:
|
||||
self.header = misp_standard_csv_header if data_header == misp_standard_csv_header else misp_extended_csv_header[:13]
|
||||
self.from_misp = True
|
||||
self.data = data[1:]
|
||||
else:
|
||||
self.from_misp = False
|
||||
self.has_header = has_header
|
||||
if header:
|
||||
self.header = header
|
||||
self.fields_number = len(header)
|
||||
self.parse_data(data)
|
||||
else:
|
||||
self.has_delimiter = True
|
||||
self.fields_number, self.delimiter, self.header = self.get_delimiter_from_header(data[0])
|
||||
self.data = data
|
||||
descFilename = os.path.join(pymisp_path[0], 'data/describeTypes.json')
|
||||
with open(descFilename, 'r') as f:
|
||||
self.MispTypes = json.loads(f.read())['result'].get('types')
|
||||
for h in self.header:
|
||||
if not (h in self.MispTypes or h in misp_extended_csv_header):
|
||||
misperrors['error'] = 'Wrong header field: {}. Please use a header value that can be recognized by MISP (or alternatively skip it using a whitespace).'.format(h)
|
||||
return misperrors
|
||||
|
||||
def get_delimiter_from_header(self, data):
|
||||
delimiters_count = {}
|
||||
for d in delimiters:
|
||||
length = data.count(d)
|
||||
if length > 0:
|
||||
delimiters_count[d] = data.count(d)
|
||||
if len(delimiters_count) == 0:
|
||||
length = 0
|
||||
delimiter = None
|
||||
header = [data]
|
||||
else:
|
||||
length, delimiter = max((n, v) for v, n in delimiters_count.items())
|
||||
header = data.split(delimiter)
|
||||
return length + 1, delimiter, header
|
||||
|
||||
def parse_data(self, data):
|
||||
return_data = []
|
||||
if self.fields_number == 1:
|
||||
for line in data:
|
||||
line = line.split('#')[0].strip()
|
||||
if line:
|
||||
return_data.append(line)
|
||||
self.delimiter = None
|
||||
else:
|
||||
self.delimiter_count = dict([(d, 0) for d in delimiters])
|
||||
for line in data:
|
||||
line = line.split('#')[0].strip()
|
||||
if line:
|
||||
self.parse_delimiter(line)
|
||||
return_data.append(line)
|
||||
# find which delimiter is used
|
||||
self.delimiter = self.find_delimiter()
|
||||
if self.fields_number == 0:
|
||||
self.header = return_data[0].split(self.delimiter)
|
||||
self.data = return_data[1:] if self.has_header else return_data
|
||||
|
||||
def parse_delimiter(self, line):
|
||||
for d in delimiters:
|
||||
if line.count(d) >= (self.fields_number - 1):
|
||||
self.delimiter_count[d] += 1
|
||||
|
||||
def find_delimiter(self):
|
||||
_, delimiter = max((n, v) for v, n in self.delimiter_count.items())
|
||||
return delimiter
|
||||
self.header = header
|
||||
self.has_header = has_header
|
||||
self.delimiter = delimiter
|
||||
self.data = data
|
||||
self.from_misp = from_misp
|
||||
self.MISPtypes = MISPtypes
|
||||
self.fields_number = len(self.header)
|
||||
self.__score_mapping = {0: self.__create_standard_misp,
|
||||
1: self.__create_attribute_with_ids,
|
||||
2: self.__create_attribute_with_tags,
|
||||
3: self.__create_attribute_with_ids_and_tags}
|
||||
|
||||
def parse_csv(self):
|
||||
if self.from_misp:
|
||||
self.build_misp_event()
|
||||
if self.header == misp_standard_csv_header:
|
||||
self.__parse_misp_csv()
|
||||
else:
|
||||
attribute_fields = misp_standard_csv_header[:1] + misp_standard_csv_header[2:10]
|
||||
object_fields = ['object_id'] + misp_standard_csv_header[10:]
|
||||
attribute_indexes = []
|
||||
object_indexes = []
|
||||
for i in range(len(self.header)):
|
||||
if self.header[i] in attribute_fields:
|
||||
attribute_indexes.append(i)
|
||||
elif self.header[i] in object_fields:
|
||||
object_indexes.append(i)
|
||||
if object_indexes:
|
||||
if not any(field in self.header for field in ('object_uuid', 'object_id')) or 'object_name' not in self.header:
|
||||
for line in self.data:
|
||||
for index in object_indexes:
|
||||
if line[index].strip():
|
||||
return {'error': 'It is not possible to import MISP objects from your csv file if you do not specify any object identifier and object name to separate each object from each other.'}
|
||||
if 'object_relation' not in self.header:
|
||||
return {'error': 'In order to import MISP objects, an object relation for each attribute contained in an object is required.'}
|
||||
self.__build_misp_event(attribute_indexes, object_indexes)
|
||||
else:
|
||||
self.buildAttributes()
|
||||
attribute_fields = attribute_fields = misp_standard_csv_header[:1] + misp_standard_csv_header[2:9]
|
||||
attribute_indexes = []
|
||||
types_indexes = []
|
||||
for i in range(len(self.header)):
|
||||
if self.header[i] in attribute_fields:
|
||||
attribute_indexes.append(i)
|
||||
elif self.header[i] in self.MISPtypes:
|
||||
types_indexes.append(i)
|
||||
self.__parse_external_csv(attribute_indexes, types_indexes)
|
||||
self.__finalize_results()
|
||||
return {'success': 1}
|
||||
|
||||
def build_misp_event(self):
|
||||
################################################################################
|
||||
# Parsing csv data with MISP fields, #
|
||||
# but a custom header #
|
||||
################################################################################
|
||||
|
||||
def __build_misp_event(self, attribute_indexes, object_indexes):
|
||||
score = self.__get_score()
|
||||
if object_indexes:
|
||||
objects = {}
|
||||
id_name = 'object_id' if 'object_id' in self.header else 'object_uuid'
|
||||
object_id_index = self.header.index(id_name)
|
||||
name_index = self.header.index('object_name')
|
||||
for line in self.data:
|
||||
attribute = self.__score_mapping[score](line, attribute_indexes)
|
||||
object_id = line[object_id_index]
|
||||
if object_id:
|
||||
if object_id not in objects:
|
||||
misp_object = MISPObject(line[name_index])
|
||||
if id_name == 'object_uuid':
|
||||
misp_object.uuid = object_id
|
||||
objects[object_id] = misp_object
|
||||
objects[object_id].add_attribute(**attribute)
|
||||
else:
|
||||
self.event.add_attribute(**attribute)
|
||||
for misp_object in objects.values():
|
||||
self.misp_event.add_object(**misp_object)
|
||||
else:
|
||||
for line in self.data:
|
||||
attribute = self.__score_mapping[score](line, attribute_indexes)
|
||||
self.misp_event.add_attribute(**attribute)
|
||||
|
||||
################################################################################
|
||||
# Parsing csv data containing fields that are not #
|
||||
# MISP attributes or objects standard fields #
|
||||
# (but should be MISP attribute types!!) #
|
||||
################################################################################
|
||||
|
||||
def __parse_external_csv(self, attribute_indexes, types_indexes):
|
||||
score = self.__get_score()
|
||||
if attribute_indexes:
|
||||
for line in self.data:
|
||||
try:
|
||||
base_attribute = self.__score_mapping[score](line, attribute_indexes)
|
||||
except IndexError:
|
||||
continue
|
||||
for index in types_indexes:
|
||||
attribute = {'type': self.header[index], 'value': line[index]}
|
||||
attribute.update(base_attribute)
|
||||
self.misp_event.add_attribute(**attribute)
|
||||
else:
|
||||
for line in self.data:
|
||||
for index in types_indexes:
|
||||
self.misp_event.add_attribute(**{'type': self.header[index], 'value': line[index]})
|
||||
|
||||
################################################################################
|
||||
# Parsing standard MISP csv format #
|
||||
################################################################################
|
||||
|
||||
def __parse_misp_csv(self):
|
||||
objects = {}
|
||||
header_length = len(self.header)
|
||||
attribute_fields = self.header[:1] + self.header[2:6] + self.header[7:8]
|
||||
attribute_fields = self.header[:1] + self.header[2:8]
|
||||
for line in self.data:
|
||||
attribute = {}
|
||||
try:
|
||||
a_uuid, _, a_category, a_type, value, comment, to_ids, timestamp, relation, tag, o_uuid, o_name, o_category = line[:header_length]
|
||||
except ValueError:
|
||||
continue
|
||||
for t, v in zip(attribute_fields, (a_uuid, a_category, a_type, value, comment, timestamp)):
|
||||
attribute[t] = v.strip('"')
|
||||
attribute['to_ids'] = True if to_ids == '1' else False
|
||||
a_uuid, _, category, _type, value, comment, ids, timestamp, relation, tag, o_uuid, name, _ = line[:self.fields_number]
|
||||
attribute = {t: v.strip('"') for t, v in zip(attribute_fields, (a_uuid, category, _type, value, comment, ids, timestamp))}
|
||||
attribute['to_ids'] = True if attribute['to_ids'] == '1' else False
|
||||
if tag:
|
||||
attribute['Tag'] = [{'name': t.strip()} for t in tag.split(',')]
|
||||
if relation:
|
||||
if o_uuid not in objects:
|
||||
objects[o_uuid] = MISPObject(o_name)
|
||||
objects[o_uuid] = MISPObject(name)
|
||||
objects[o_uuid].add_attribute(relation, **attribute)
|
||||
else:
|
||||
self.misp_event.add_attribute(**attribute)
|
||||
for uuid, misp_object in objects.items():
|
||||
misp_object.uuid = uuid
|
||||
self.misp_event.add_object(**misp_object)
|
||||
self.finalize_results()
|
||||
|
||||
def buildAttributes(self):
|
||||
# if there is only 1 field of data
|
||||
if self.delimiter is None:
|
||||
mispType = self.header[0]
|
||||
for data in self.data:
|
||||
d = data.strip()
|
||||
if d:
|
||||
self.misp_event.add_attribute(**{'type': mispType, 'value': d})
|
||||
else:
|
||||
# split fields that should be recognized as misp attribute types from the others
|
||||
list2pop, misp, head = self.findMispTypes()
|
||||
# for each line of data
|
||||
for data in self.data:
|
||||
datamisp = []
|
||||
datasplit = data.split(self.delimiter)
|
||||
# in case there is an empty line or an error
|
||||
if len(datasplit) != self.fields_number:
|
||||
continue
|
||||
# pop from the line data that matches with a misp type, using the list of indexes
|
||||
for l in list2pop:
|
||||
datamisp.append(datasplit.pop(l).strip())
|
||||
# for each misp type, we create an attribute
|
||||
for m, dm in zip(misp, datamisp):
|
||||
attribute = {'type': m, 'value': dm}
|
||||
for h, ds in zip(head, datasplit):
|
||||
if h:
|
||||
attribute[h] = ds.strip()
|
||||
self.misp_event.add_attribute(**attribute)
|
||||
self.finalize_results()
|
||||
################################################################################
|
||||
# Utility functions #
|
||||
################################################################################
|
||||
|
||||
def findMispTypes(self):
|
||||
list2pop = []
|
||||
misp = []
|
||||
head = []
|
||||
for h in reversed(self.header):
|
||||
n = self.header.index(h)
|
||||
# fields that are misp attribute types
|
||||
if h in self.MispTypes:
|
||||
list2pop.append(n)
|
||||
misp.append(h)
|
||||
# handle confusions between misp attribute types and attribute fields
|
||||
elif h in duplicatedFields['mispType']:
|
||||
# fields that should be considered as misp attribute types
|
||||
list2pop.append(n)
|
||||
misp.append(duplicatedFields['mispType'].get(h))
|
||||
elif h in duplicatedFields['attrField']:
|
||||
# fields that should be considered as attribute fields
|
||||
head.append(duplicatedFields['attrField'].get(h))
|
||||
# or, it could be an attribute field
|
||||
elif h in attributesFields:
|
||||
head.append(h)
|
||||
# otherwise, it is not defined
|
||||
else:
|
||||
head.append('')
|
||||
# return list of indexes of the misp types, list of the misp types, remaining fields that will be attribute fields
|
||||
return list2pop, misp, list(reversed(head))
|
||||
def __create_attribute_with_ids(self, line, indexes):
|
||||
attribute = self.__create_standard_misp(line, indexes)
|
||||
return self.__deal_with_ids(attribute)
|
||||
|
||||
def finalize_results(self):
|
||||
def __create_attribute_with_ids_and_tags(self, line, indexes):
|
||||
attribute = self.__deal_with_ids(self.__create_standard_misp(line, indexes))
|
||||
return self.__deal_with_tags(attribute)
|
||||
|
||||
def __create_attribute_with_tags(self, line, indexes):
|
||||
attribute = self.__create_standard_misp(line, indexes)
|
||||
return self.__deal_with_tags(attribute)
|
||||
|
||||
def __create_standard_misp(self, line, indexes):
|
||||
return {self.header[index]: line[index] for index in indexes if line[index]}
|
||||
|
||||
@staticmethod
|
||||
def __deal_with_ids(attribute):
|
||||
attribute['to_ids'] = True if attribute['to_ids'] == '1' else False
|
||||
return attribute
|
||||
|
||||
@staticmethod
|
||||
def __deal_with_tags(attribute):
|
||||
attribute['Tag'] = [{'name': tag.strip()} for tag in attribute['Tag'].split(',')]
|
||||
return attribute
|
||||
|
||||
def __get_score(self):
|
||||
score = 1 if 'to_ids' in self.header else 0
|
||||
if 'attribute_tag' in self.header:
|
||||
score += 2
|
||||
return score
|
||||
|
||||
def __finalize_results(self):
|
||||
event = json.loads(self.misp_event.to_json())
|
||||
self.results = {key: event[key] for key in ('Attribute', 'Object') if (key in event and event[key])}
|
||||
|
||||
|
||||
def __any_mandatory_misp_field(header):
|
||||
return any(field in header for field in ('type', 'value'))
|
||||
|
||||
|
||||
def __special_parsing(data, delimiter):
|
||||
return list(line.split(delimiter) for line in csv.reader(io.TextIOWrapper(io.BytesIO(data.encode()), encoding='utf-8')) if line and not line.startswith('#'))
|
||||
|
||||
|
||||
def __standard_parsing(data):
|
||||
return list(line for line in csv.reader(io.TextIOWrapper(io.BytesIO(data.encode()), encoding='utf-8')) if line and not line[0].startswith('#'))
|
||||
|
||||
|
||||
def handler(q=False):
|
||||
if q is False:
|
||||
return False
|
||||
request = json.loads(q)
|
||||
if request.get('data'):
|
||||
data = base64.b64decode(request['data']).decode('utf-8')
|
||||
data = [line for line in csv.reader(io.TextIOWrapper(io.BytesIO(data.encode()), encoding='utf-8'))]
|
||||
else:
|
||||
misperrors['error'] = "Unsupported attributes type"
|
||||
return misperrors
|
||||
has_header = request['config'].get('has_header')
|
||||
has_header = True if has_header == '1' else False
|
||||
if not request.get('config') and not request['config'].get('header'):
|
||||
header = request['config']['header'].split(',') if request['config'].get('header').strip() else []
|
||||
delimiter = request['config']['special_delimiter'] if request['config'].get('special_delimiter').strip() else ','
|
||||
data = __standard_parsing(data) if delimiter == ',' else __special_parsing(data, delimiter)
|
||||
if not header:
|
||||
if has_header:
|
||||
header = []
|
||||
header = data.pop(0)
|
||||
else:
|
||||
misperrors['error'] = "Configuration error"
|
||||
misperrors['error'] = "Configuration error. Provide a header or use the one within the csv file and tick the checkbox 'Has_header'."
|
||||
return misperrors
|
||||
else:
|
||||
header = request['config'].get('header').split(',')
|
||||
header = [c.strip() for c in header]
|
||||
csv_parser = CsvParser(header, has_header, data)
|
||||
header = [h.strip() for h in header]
|
||||
if has_header:
|
||||
del data[0]
|
||||
if header == misp_standard_csv_header or header == misp_extended_csv_header:
|
||||
header = misp_standard_csv_header
|
||||
descFilename = os.path.join(pymisp_path[0], 'data/describeTypes.json')
|
||||
with open(descFilename, 'r') as f:
|
||||
MISPtypes = json.loads(f.read())['result'].get('types')
|
||||
for h in header:
|
||||
if not any((h in MISPtypes, h in misp_extended_csv_header, h in ('', ' ', '_', 'object_id'))):
|
||||
misperrors['error'] = 'Wrong header field: {}. Please use a header value that can be recognized by MISP (or alternatively skip it using a whitespace).'.format(h)
|
||||
return misperrors
|
||||
from_misp = all((h in misp_extended_csv_header or h in ('', ' ', '_', 'object_id') for h in header))
|
||||
if from_misp:
|
||||
if not __any_mandatory_misp_field(header):
|
||||
misperrors['error'] = 'Please make sure the data you try to import can be identified with a type/value combinaison.'
|
||||
return misperrors
|
||||
else:
|
||||
if __any_mandatory_misp_field(header):
|
||||
wrong_types = tuple(wrong_type for wrong_type in ('type', 'value') if wrong_type in header)
|
||||
misperrors['error'] = 'Error with the following header: {}. It contains the following field(s): {}, which is(are) already provided by the usage of at least on MISP attribute type in the header.'.format(header, 'and'.join(wrong_types))
|
||||
return misperrors
|
||||
csv_parser = CsvParser(header, has_header, delimiter, data, from_misp, MISPtypes)
|
||||
# build the attributes
|
||||
csv_parser.parse_csv()
|
||||
result = csv_parser.parse_csv()
|
||||
if 'error' in result:
|
||||
return result
|
||||
return {'results': csv_parser.results}
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue