mirror of https://github.com/MISP/PyMISP
337 lines
13 KiB
Python
337 lines
13 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import datetime
|
|
import time
|
|
import json
|
|
from json import JSONEncoder
|
|
import os
|
|
try:
|
|
from dateutil.parser import parse
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
import jsonschema
|
|
except ImportError:
|
|
pass
|
|
|
|
from .exceptions import PyMISPError, NewEventError, NewAttributeError
|
|
|
|
|
|
class MISPAttribute(object):
|
|
|
|
def __init__(self, describe_types):
|
|
self.categories = describe_types['categories']
|
|
self.types = describe_types['types']
|
|
self.category_type_mapping = describe_types['category_type_mappings']
|
|
self.sane_default = describe_types['sane_defaults']
|
|
self._reinitialize_attribute()
|
|
|
|
def _reinitialize_attribute(self):
|
|
# Default values
|
|
self.category = None
|
|
self.type = None
|
|
self.value = None
|
|
self.to_ids = False
|
|
self.comment = ''
|
|
self.distribution = 5
|
|
|
|
# other possible values
|
|
self.id = None
|
|
self.uuid = None
|
|
self.timestamp = None
|
|
self.sharing_group_id = None
|
|
self.deleted = None
|
|
self.SharingGroup = []
|
|
self.ShadowAttribute = []
|
|
|
|
def set_all_values(self, **kwargs):
|
|
if kwargs.get('type') and kwargs.get('category'):
|
|
if kwargs['type'] not in self.category_type_mapping[kwargs['category']]:
|
|
raise NewAttributeError('{} and {} is an invalid combinaison, type for this category has to be in {}'.capitalizeformat(self.type, self.category, (', '.join(self.category_type_mapping[self.category]))))
|
|
# Required
|
|
if kwargs.get('type'):
|
|
self.type = kwargs['type']
|
|
if self.type not in self.types:
|
|
raise NewAttributeError('{} is invalid, type has to be in {}'.format(self.type, (', '.join(self.types))))
|
|
else:
|
|
raise NewAttributeError('The type of the attribute is required.')
|
|
|
|
type_defaults = self.sane_default[self.type]
|
|
|
|
if kwargs.get('value'):
|
|
self.value = kwargs['value']
|
|
else:
|
|
raise NewAttributeError('The value of the attribute is required.')
|
|
|
|
# Default values
|
|
if kwargs.get('category'):
|
|
self.category = kwargs['category']
|
|
if self.category not in self.categories:
|
|
raise NewAttributeError('{} is invalid, category has to be in {}'.format(self.category, (', '.join(self.categories))))
|
|
else:
|
|
self.category = type_defaults['default_category']
|
|
|
|
if kwargs.get('to_ids'):
|
|
self.to_ids = kwargs['to_ids']
|
|
if not isinstance(self.to_ids, bool):
|
|
raise NewAttributeError('{} is invalid, to_ids has to be True or False'.format(self.to_ids))
|
|
else:
|
|
self.to_ids = bool(int(type_defaults['to_ids']))
|
|
if kwargs.get('comment'):
|
|
self.comment = kwargs['comment']
|
|
if kwargs.get('distribution'):
|
|
self.distribution = int(kwargs['distribution'])
|
|
if self.distribution not in [0, 1, 2, 3, 5]:
|
|
raise NewAttributeError('{} is invalid, the distribution has to be in 0, 1, 2, 3, 5'.format(self.distribution))
|
|
|
|
# other possible values
|
|
if kwargs.get('id'):
|
|
self.id = int(kwargs['id'])
|
|
if kwargs.get('uuid'):
|
|
self.uuid = kwargs['uuid']
|
|
if kwargs.get('timestamp'):
|
|
self.timestamp = datetime.datetime.fromtimestamp(int(kwargs['timestamp']))
|
|
if kwargs.get('sharing_group_id'):
|
|
self.sharing_group_id = int(kwargs['sharing_group_id'])
|
|
if kwargs.get('deleted'):
|
|
self.deleted = kwargs['deleted']
|
|
if kwargs.get('SharingGroup'):
|
|
self.SharingGroup = kwargs['SharingGroup']
|
|
if kwargs.get('ShadowAttribute'):
|
|
self.ShadowAttribute = kwargs['ShadowAttribute']
|
|
|
|
def _json(self):
|
|
to_return = {'type': self.type, 'category': self.category, 'to_ids': self.to_ids,
|
|
'distribution': self.distribution, 'value': self.value,
|
|
'comment': self.comment}
|
|
if self.sharing_group_id:
|
|
to_return['sharing_group_id'] = self.sharing_group_id
|
|
return to_return
|
|
|
|
def _json_full(self):
|
|
to_return = self._json()
|
|
if self.id:
|
|
to_return['id'] = self.id
|
|
if self.uuid:
|
|
to_return['uuid'] = self.uuid
|
|
if self.timestamp:
|
|
to_return['timestamp'] = int(time.mktime(self.timestamp.timetuple()))
|
|
if self.deleted is not None:
|
|
to_return['deleted'] = self.deleted
|
|
if self.ShadowAttribute:
|
|
to_return['ShadowAttribute'] = self.ShadowAttribute
|
|
if self.SharingGroup:
|
|
to_return['SharingGroup'] = self.SharingGroup
|
|
return to_return
|
|
|
|
|
|
class EncodeUpdate(JSONEncoder):
|
|
def default(self, obj):
|
|
try:
|
|
return obj._json()
|
|
except AttributeError:
|
|
return JSONEncoder.default(self, obj)
|
|
|
|
|
|
class EncodeFull(JSONEncoder):
|
|
def default(self, obj):
|
|
try:
|
|
return obj._json_full()
|
|
except AttributeError:
|
|
return JSONEncoder.default(self, obj)
|
|
|
|
|
|
class MISPEvent(object):
|
|
|
|
def __init__(self, describe_types=None):
|
|
self.ressources_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'data')
|
|
self.json_schema = json.load(open(os.path.join(self.ressources_path, 'schema.json'), 'r'))
|
|
self.json_schema_lax = json.load(open(os.path.join(self.ressources_path, 'schema-lax.json'), 'r'))
|
|
if not describe_types:
|
|
t = json.load(open(os.path.join(self.ressources_path, 'describeTypes.json'), 'r'))
|
|
describe_types = t['result']
|
|
self.describe_types = describe_types
|
|
self.categories = describe_types['categories']
|
|
self.types = describe_types['types']
|
|
self.category_type_mapping = describe_types['category_type_mappings']
|
|
self.sane_default = describe_types['sane_defaults']
|
|
self.new = True
|
|
self.dump_full = False
|
|
|
|
self._reinitialize_event()
|
|
|
|
def _reinitialize_event(self):
|
|
# Default values for a valid event to send to a MISP instance
|
|
self.distribution = 3
|
|
self.threat_level_id = 2
|
|
self.analysis = 0
|
|
self.info = None
|
|
self.published = False
|
|
self.date = datetime.date.today()
|
|
self.attributes = []
|
|
|
|
# All other keys
|
|
self.id = None
|
|
self.orgc_id = None
|
|
self.org_id = None
|
|
self.uuid = None
|
|
self.attribute_count = None
|
|
self.timestamp = None
|
|
self.proposal_email_lock = None
|
|
self.locked = None
|
|
self.publish_timestamp = None
|
|
self.sharing_group_id = None
|
|
self.Org = None
|
|
self.Orgc = None
|
|
self.ShadowAttribute = []
|
|
self.RelatedEvent = []
|
|
self.Tag = []
|
|
|
|
def load(self, json_event):
|
|
self.new = False
|
|
self.dump_full = True
|
|
if isinstance(json_event, str):
|
|
loaded = json.loads(json_event)
|
|
if loaded.get('response'):
|
|
event = loaded.get('response')[0]
|
|
else:
|
|
event = loaded
|
|
if not event:
|
|
raise PyMISPError('Invalid event')
|
|
else:
|
|
event = json_event
|
|
jsonschema.validate(event, self.json_schema_lax)
|
|
e = event.get('Event')
|
|
self._reinitialize_event()
|
|
self.set_all_values(**e)
|
|
|
|
def set_all_values(self, **kwargs):
|
|
# Required value
|
|
if kwargs.get('info'):
|
|
self.info = kwargs['info']
|
|
else:
|
|
raise NewAttributeError('The info field of the new event is required.')
|
|
|
|
# Default values for a valid event to send to a MISP instance
|
|
if kwargs.get('distribution') is not None:
|
|
self.distribution = int(kwargs['distribution'])
|
|
if self.distribution not in [0, 1, 2, 3]:
|
|
raise NewEventError('{} is invalid, the distribution has to be in 0, 1, 2, 3'.format(self.distribution))
|
|
if kwargs.get('threat_level_id') is not None:
|
|
self.threat_level_id = int(kwargs['threat_level_id'])
|
|
if self.threat_level_id not in [1, 2, 3, 4]:
|
|
raise NewEventError('{} is invalid, the threat_level has to be in 1, 2, 3, 4'.format(self.threat_level_id))
|
|
if kwargs.get('analysis') is not None:
|
|
self.analysis = int(kwargs['analysis'])
|
|
if self.analysis not in [0, 1, 2]:
|
|
raise NewEventError('{} is invalid, the analysis has to be in 0, 1, 2'.format(self.analysis))
|
|
if kwargs.get('published') is not None:
|
|
self.publish()
|
|
if kwargs.get('date'):
|
|
if isinstance(kwargs['date'], str):
|
|
self.date = parse(kwargs['date'])
|
|
elif isinstance(kwargs['date'], datetime.datetime):
|
|
self.date = kwargs['date'].date()
|
|
elif isinstance(kwargs['date'], datetime.date):
|
|
self.date = kwargs['date']
|
|
else:
|
|
raise NewEventError('Invalid format for the date: {} - {}'.format(kwargs['date'], type(kwargs['date'])))
|
|
if kwargs.get('Attribute'):
|
|
for a in kwargs['Attribute']:
|
|
attribute = MISPAttribute(self.describe_types)
|
|
attribute.set_all_values(**a)
|
|
self.attributes.append(attribute)
|
|
|
|
# All other keys
|
|
if kwargs.get('id'):
|
|
self.id = int(kwargs['id'])
|
|
if kwargs.get('orgc_id'):
|
|
self.orgc_id = int(kwargs['orgc_id'])
|
|
if kwargs.get('org_id'):
|
|
self.org_id = int(kwargs['org_id'])
|
|
if kwargs.get('uuid'):
|
|
self.uuid = kwargs['uuid']
|
|
if kwargs.get('attribute_count'):
|
|
self.attribute_count = int(kwargs['attribute_count'])
|
|
if kwargs.get('timestamp'):
|
|
self.timestamp = datetime.datetime.fromtimestamp(int(kwargs['timestamp']))
|
|
if kwargs.get('proposal_email_lock'):
|
|
self.proposal_email_lock = kwargs['proposal_email_lock']
|
|
if kwargs.get('locked'):
|
|
self.locked = kwargs['locked']
|
|
if kwargs.get('publish_timestamp'):
|
|
self.publish_timestamp = datetime.datetime.fromtimestamp(int(kwargs['publish_timestamp']))
|
|
if kwargs.get('sharing_group_id'):
|
|
self.sharing_group_id = int(kwargs['sharing_group_id'])
|
|
if kwargs.get('Org'):
|
|
self.Org = kwargs['Org']
|
|
if kwargs.get('Orgc'):
|
|
self.Orgc = kwargs['Orgc']
|
|
if kwargs.get('ShadowAttribute'):
|
|
self.ShadowAttribute = kwargs['ShadowAttribute']
|
|
if kwargs.get('RelatedEvent'):
|
|
self.RelatedEvent = kwargs['RelatedEvent']
|
|
if kwargs.get('Tag'):
|
|
self.Tag = kwargs['Tag']
|
|
|
|
def _json(self):
|
|
to_return = {'Event': {}}
|
|
to_return['Event'] = {'distribution': self.distribution, 'info': self.info,
|
|
'date': self.date.isoformat(), 'published': self.published,
|
|
'threat_level_id': self.threat_level_id,
|
|
'analysis': self.analysis, 'Attribute': []}
|
|
if self.id:
|
|
to_return['Event']['id'] = self.id
|
|
if self.orgc_id:
|
|
to_return['Event']['orgc_id'] = self.orgc_id
|
|
if self.org_id:
|
|
to_return['Event']['org_id'] = self.org_id
|
|
if self.uuid:
|
|
to_return['Event']['uuid'] = self.uuid
|
|
if self.sharing_group_id:
|
|
to_return['Event']['sharing_group_id'] = self.sharing_group_id
|
|
if self.Tag:
|
|
to_return['Event']['Tag'] = self.Tag
|
|
to_return['Event']['Attribute'] = [a._json() for a in self.attributes]
|
|
jsonschema.validate(to_return, self.json_schema)
|
|
return to_return
|
|
|
|
def _json_full(self):
|
|
to_return = self._json()
|
|
if self.locked is not None:
|
|
to_return['Event']['locked'] = self.locked
|
|
if self.attribute_count:
|
|
to_return['Event']['attribute_count'] = self.attribute_count
|
|
if self.RelatedEvent:
|
|
to_return['Event']['RelatedEvent'] = self.RelatedEvent
|
|
if self.Org:
|
|
to_return['Event']['Org'] = self.Org
|
|
if self.Orgc:
|
|
to_return['Event']['Orgc'] = self.Orgc
|
|
if self.ShadowAttribute:
|
|
to_return['Event']['ShadowAttribute'] = self.ShadowAttribute
|
|
if self.proposal_email_lock is not None:
|
|
to_return['Event']['proposal_email_lock'] = self.proposal_email_lock
|
|
if self.locked is not None:
|
|
to_return['Event']['locked'] = self.locked
|
|
if self.publish_timestamp:
|
|
to_return['Event']['publish_timestamp'] = int(time.mktime(self.publish_timestamp.timetuple()))
|
|
if self.timestamp:
|
|
to_return['Event']['timestamp'] = int(time.mktime(self.timestamp.timetuple()))
|
|
to_return['Event']['Attribute'] = [a._json_full() for a in self.attributes]
|
|
jsonschema.validate(to_return, self.json_schema)
|
|
return to_return
|
|
|
|
def publish(self):
|
|
self.published = True
|
|
|
|
def unpublish(self):
|
|
self.published = False
|
|
|
|
def add_attribute(self, type, value, **kwargs):
|
|
attribute = MISPAttribute(self.describe_types)
|
|
attribute.set_all_values(type=type, value=value, **kwargs)
|
|
self.attributes.append(attribute)
|