Browse Source

fix: csv import rework & improvement

- More efficient parsing
- Support of multiple csv formats
- Possibility to customise headers
- More improvement to come for external csv file
features_csvimport
chrisr3d 2 years ago
parent
commit
cfc6438c47
No known key found for this signature in database GPG Key ID: 6BBED1B63A6D639F
  1. 348
      misp_modules/modules/import_mod/csvimport.py

348
misp_modules/modules/import_mod/csvimport.py

@ -8,17 +8,22 @@ 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'},
@ -33,195 +38,230 @@ 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)
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:
if self.header == misp_standard_csv_header:
self.__parse_misp_csv()
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]
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:
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
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}
################################################################################
#### 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:
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
for line in self.data:
attribute = self.__score_mapping[score](line, attribute_indexes)
self.misp_event.add_attribute(**attribute)
def parse_csv(self):
if self.from_misp:
self.build_misp_event()
################################################################################
#### 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:
base_attribute = self.__score_mapping[score](line, attribute_indexes)
for index in types_indexes:
attribute = {'type': self.header[index], 'value': line[index]}
attribute.update(base_attribute)
self.misp_event.add_attribute(**attribute)
else:
self.buildAttributes()
for line in self.data:
for index in types_indexes:
self.misp_event.add_attribute(**{'type': self.header[index], 'value': line[index]})
def build_misp_event(self):
################################################################################
#### 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('"')
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 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()
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 finalize_results(self):
################################################################################
#### Utility functions ####
################################################################################
def __create_attribute_with_ids(self, line, indexes):
attribute = self.__create_standard_misp(line, indexes)
return self.__deal_with_ids(attribute)
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')))
def __standard_parsing(data):
return list(line for line in csv.reader(io.TextIOWrapper(io.BytesIO(data.encode()), encoding='utf-8')))
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 = [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:
header = request['config'].get('header').split(',')
header = [c.strip() for c in header]
csv_parser = CsvParser(header, has_header, data)
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…
Cancel
Save