CIRCL
/
AIL-framework
mirror of https://github.com/CIRCL/AIL-framework
2
1
Fork 0
Code Issues Releases Wiki Activity
AIL-framework/bin/export/MispImport.py

245 lines
8.5 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
# -*-coding:UTF-8 -*
import os
import sys
import uuid
import redis
sys.path.append(os.path.join(os.environ['AIL_BIN'], 'lib'))
sys.path.append(os.path.join(os.environ['AIL_BIN'], 'packages'))
import Item
import Cryptocurrency
import Pgp
import Decoded
import Domain
import Screenshot
import Correlate_object
import AILObjects
# MISP
from pymisp import MISPEvent, MISPObject, PyMISP
# # TODO: deplace me in another fil
def get_global_id(obj_type, obj_id, obj_subtype=None):
if obj_subtype:
return '{}:{}:{}'.format(obj_type, obj_subtype, obj_id)
else:
return '{}:{}'.format(obj_type, obj_id)
# sub type
# obj type
# obj value
def get_global_id_from_id(global_id):
obj_meta = {}
global_id = global_id.split(':', 3)
if len(global_id) > 2:
obj_meta['type'] = global_id[0]
obj_meta['subtype'] = global_id[1]
obj_meta['id'] = global_id[2]
else:
obj_meta['type'] = global_id[0]
obj_meta['subtype'] = None
obj_meta['id'] = global_id[1]
return obj_meta
def get_import_dir():
return os.path.join(os.environ['AIL_HOME'], 'temp/import')
def sanitize_import_file_path(filename):
IMPORT_FOLDER = get_import_dir()
filename = os.path.join(IMPORT_FOLDER, filename)
filename = os.path.realpath(filename)
# path traversal
if not os.path.commonprefix([filename, IMPORT_FOLDER]) == IMPORT_FOLDER:
return os.path.join(IMPORT_FOLDER, str(uuid.uuid4()) + '.json')
# check if file already exist
if os.path.isfile(filename):
return os.path.join(IMPORT_FOLDER, str(uuid.uuid4()) + '.json')
return filename
def get_misp_obj_tag(misp_obj):
if misp_obj.attributes:
misp_tags = misp_obj.attributes[0].tags
tags = []
for misp_tag in misp_tags:
tags.append(misp_tag.name)
return tags
else:
return []
def get_object_metadata(misp_obj):
obj_meta = {}
if 'first_seen' in misp_obj.keys():
obj_meta['first_seen'] = misp_obj.first_seen
if 'last_seen' in misp_obj.keys():
obj_meta['last_seen'] = misp_obj.last_seen
obj_meta['tags'] = get_misp_obj_tag(misp_obj)
return obj_meta
def unpack_item_obj(map_uuid_global_id, misp_obj):
obj_meta = get_object_metadata(misp_obj)
obj_id = None
io_content = None
for attribute in misp_obj.attributes:
if attribute.object_relation == 'raw-data':
obj_id = attribute.value # # TODO: sanitize
io_content = attribute.data # # TODO: check if type == io
if obj_id and io_content:
res = Item.create_item(obj_id, obj_meta, io_content)
map_uuid_global_id[misp_obj.uuid] = get_global_id('item', obj_id)
## TODO: handle multiple pgp in the same object
def unpack_obj_pgp(map_uuid_global_id, misp_obj):
# get obj sub type
obj_attr = misp_obj.attributes[0]
obj_id = obj_attr.value
if obj_attr.object_relation == 'key-id':
obj_subtype = 'key'
elif obj_attr.object_relation == 'user-id-name':
obj_subtype = 'name'
elif obj_attr.object_relation == 'user-id-email':
obj_subtype = 'mail'
if obj_id and obj_subtype:
obj_meta = get_object_metadata(misp_obj)
res = Pgp.pgp.create_correlation(obj_subtype, obj_id, obj_meta)
map_uuid_global_id[misp_obj.uuid] = get_global_id('pgp', obj_id, obj_subtype=obj_subtype)
def unpack_obj_cryptocurrency(map_uuid_global_id, misp_obj):
obj_id = None
obj_subtype = None
for attribute in misp_obj.attributes:
if attribute.object_relation == 'address': # # TODO: handle xmr address field
obj_id = attribute.value
elif attribute.object_relation == 'symbol':
obj_subtype = Cryptocurrency.get_cryptocurrency_type(attribute.value)
# valid cryptocurrency type
if obj_subtype and obj_id:
obj_meta = get_object_metadata(misp_obj)
res = Cryptocurrency.cryptocurrency.create_correlation(obj_subtype, obj_id, obj_meta)
map_uuid_global_id[misp_obj.uuid] = get_global_id('cryptocurrency', obj_id, obj_subtype=obj_subtype)
def get_obj_type_from_relationship(misp_obj):
obj_uuid = misp_obj.uuid
obj_type = None
for relation in misp_obj.ObjectReference:
if relation.object_uuid == obj_uuid:
if relation.relationship_type == "screenshot-of":
return 'screenshot'
if relation.relationship_type == "included-in":
obj_type = 'decoded'
return obj_type
# # TODO: covert md5 and sha1 to expected
def unpack_file(map_uuid_global_id, misp_obj):
obj_type = get_obj_type_from_relationship(misp_obj)
if obj_type:
obj_id = None
io_content = None
for attribute in misp_obj.attributes:
# get file content
if attribute.object_relation == 'attachment':
io_content = attribute.data
elif attribute.object_relation == 'malware-sample':
io_content = attribute.data
# # TODO: use/verify specified mimetype
elif attribute.object_relation == 'mimetype':
#print(attribute.value)
pass
# # TODO: support more
elif attribute.object_relation == 'sha1' and obj_type == 'decoded':
obj_id = attribute.value
elif attribute.object_relation == 'sha256' and obj_type == 'screenshot':
obj_id = attribute.value
if obj_id and io_content:
obj_meta = get_object_metadata(misp_obj)
if obj_type == 'screenshot':
Screenshot.create_screenshot(obj_id, obj_meta, io_content)
map_uuid_global_id[misp_obj.uuid] = get_global_id('image', obj_id)
else: #decoded
Decoded.create_decoded(obj_id, obj_meta, io_content)
map_uuid_global_id[misp_obj.uuid] = get_global_id('decoded', obj_id)
def get_misp_import_fct(map_uuid_global_id, misp_obj):
if misp_obj.name == 'ail-leak':
unpack_item_obj(map_uuid_global_id, misp_obj)
elif misp_obj.name == 'domain-ip':
pass
elif misp_obj.name == 'pgp-meta':
unpack_obj_pgp(map_uuid_global_id, misp_obj)
elif misp_obj.name == 'coin-address':
unpack_obj_cryptocurrency(map_uuid_global_id, misp_obj)
elif misp_obj.name == 'file':
unpack_file(map_uuid_global_id, misp_obj)
# import relationship between objects
def create_obj_relationships(map_uuid_global_id, misp_obj):
if misp_obj.uuid in map_uuid_global_id:
for relationship in misp_obj.ObjectReference:
if relationship.referenced_uuid in map_uuid_global_id:
obj_meta_src = get_global_id_from_id(map_uuid_global_id[relationship.object_uuid])
obj_meta_target = get_global_id_from_id(map_uuid_global_id[relationship.referenced_uuid])
if obj_meta_src == 'decoded' or obj_meta_src == 'item':
print('000000')
print(obj_meta_src)
print(obj_meta_target)
print('111111')
Correlate_object.create_obj_relationship(obj_meta_src['type'], obj_meta_src['id'], obj_meta_target['type'], obj_meta_target['id'],
obj1_subtype=obj_meta_src['subtype'], obj2_subtype=obj_meta_target['subtype'])
def create_map_all_obj_uuid_golbal_id(map_uuid_global_id):
for obj_uuid in map_uuid_global_id:
AILObjects.create_map_obj_uuid_golbal_id(obj_uuid, map_uuid_global_id[obj_uuid])
def import_objs_from_file(filepath):
map_uuid_global_id = {}
event_to_import = MISPEvent()
try:
event_to_import.load_file(filepath)
except:
return map_uuid_global_id
for misp_obj in event_to_import.objects:
get_misp_import_fct(map_uuid_global_id, misp_obj)
for misp_obj in event_to_import.objects:
create_obj_relationships(map_uuid_global_id, misp_obj)
create_map_all_obj_uuid_golbal_id(map_uuid_global_id)
return map_uuid_global_id
if __name__ == '__main__':
# misp = PyMISP('https://127.0.0.1:8443/', 'uXgcN42b7xuL88XqK5hubwD8Q8596VrrBvkHQzB0', False)
import_objs_from_file('test_import_item.json')
#Screenshot.delete_screenshot('a92d459f70c4dea8a14688f585a5e2364be8b91fbf924290ead361d9b909dcf1')
#Decoded.delete_decoded('d59a110ab233fe87cefaa0cf5603b047b432ee07')
#Pgp.pgp.delete_correlation('key', '0xA4BB02A75E6AF448')
#Item.delete_item('submitted/2020/02/10/b2485894-4325-469b-bc8f-6ad1c2dbb202.gz')
#Item.delete_item('archive/pastebin.com_pro/2020/02/10/K2cerjP4.gz')
Reference in New Issue View Git Blame Copy Permalink