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

Merge remote-tracking branch 'MISP/master'

pull/157/head
Christophe Vandeplas 2017-12-20 13:29:36 +01:00
parent 582fd28702 78c156bb6f
commit 2fb4170152
2 changed files with 133 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
pymisp/abstract.py
View File

@ -29,6 +29,7 @@ class AbstractMISP(collections.MutableMapping):
def __init__(self, **kwargs):
super(AbstractMISP, self).__init__()
self.edited = True
def properties(self):
to_return = []
@ -43,6 +44,8 @@ class AbstractMISP(collections.MutableMapping):
if value is None:
continue
setattr(self, prop, value)
# We load an existing dictionary, marking it an not-edited
self.edited = False
def update_not_jsonable(self, *args):
self.__not_jsonable += args
@ -87,3 +90,19 @@ class AbstractMISP(collections.MutableMapping):
def __len__(self):
return len(self.to_dict())
@property
def edited(self):
return self.__edited
@edited.setter
def edited(self, val):
if isinstance(val, bool):
self.__edited = val
else:
raise Exception('edited can only be True or False')
def __setattr__(self, name, value):
if name in self.properties():
self.__edited = True
super(AbstractMISP, self).__setattr__(name, value)

229
pymisp/mispevent.py
View File

@ -2,7 +2,6 @@
# -*- coding: utf-8 -*-
import datetime
import time
import json
import os
import base64
@ -66,6 +65,17 @@ def _int_to_str(d):
return d
def _datetime_to_timestamp(d):
if isinstance(d, (int, str)):
# Assume we already have a timestamp
return d
if sys.version_info >= (3, 3):
return d.timestamp()
else:
from datetime import timezone # Only for Python < 3.3
return (d - datetime(1970, 1, 1, tzinfo=timezone.utc)).total_seconds()
class MISPAttribute(AbstractMISP):
def __init__(self, describe_types=None):
@ -80,31 +90,6 @@ class MISPAttribute(AbstractMISP):
self.__sane_default = describe_types['sane_defaults']
self.Tag = []
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.data = None
self.encrypt = False
self.id = None
self.event_id = None
self.uuid = None
self.timestamp = None
self.sharing_group_id = None
self.deleted = None
self.sig = None
self.SharingGroup = []
self.ShadowAttribute = []
self.disable_correlation = False
self.RelatedAttribute = []
self.Tag = []
def get_known_types(self):
return self._types
@ -129,7 +114,12 @@ class MISPAttribute(AbstractMISP):
self.deleted = True
def add_tag(self, tag):
self.Tag.append({'name': tag})
misp_tag = MISPTag()
if isinstance(tag, str):
misp_tag.from_dict(name=tag)
elif isinstance(tag, dict):
misp_tag.from_dict(**tag)
self.Tag.append(misp_tag)
def verify(self, gpg_uid):
if not has_pyme:
@ -203,15 +193,18 @@ class MISPAttribute(AbstractMISP):
if kwargs.get('sharing_group_id'):
self.sharing_group_id = int(kwargs.pop('sharing_group_id'))
if kwargs.get('Tag'):
self.Tag = [t for t in kwargs.pop('Tag', []) if t]
self.Tag = []
for tag in kwargs.pop('Tag'):
t = MISPTag()
t.from_dict(**tag)
self.Tag.append(t)
# If the user wants to disable correlation, let them. Defaults to False.
self.disable_correlation = kwargs.pop("disable_correlation", False)
if self.disable_correlation is None:
self.disable_correlation = False
for k, v in kwargs.items():
setattr(self, k, v)
super(MISPAttribute, self).from_dict(**kwargs)
def _prepare_new_malware_sample(self):
if '|' in self.value:
@ -275,20 +268,16 @@ class MISPAttribute(AbstractMISP):
# DEPRECATED
return self.to_dict()
def to_dict(self, with_timestamp=False):
to_return = {}
for attribute in self.properties():
val = getattr(self, attribute, None)
if val in [None, []]:
continue
def to_dict(self):
to_return = super(MISPAttribute, self).to_dict()
if to_return.get('data'):
to_return['data'] = base64.b64encode(self.data.getvalue()).decode()
if self.edited:
to_return.pop('timestamp', None)
elif to_return.get('timestamp'):
to_return['timestamp'] = _datetime_to_timestamp(self.timestamp)
if attribute == 'data':
to_return['data'] = base64.b64encode(self.data.getvalue()).decode()
elif attribute == 'timestamp':
if with_timestamp:
to_return['timestamp'] = int(time.mktime(self.timestamp.timetuple()))
else:
to_return[attribute] = val
to_return = _int_to_str(to_return)
return to_return
@ -314,36 +303,6 @@ class MISPEvent(AbstractMISP):
self.Object = []
self.RelatedEvent = []
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()
# All other keys
self.sig = None
self.global_sig = None
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 = []
self.Galaxy = None
self.Object = None
def get_known_types(self):
return self._types
@ -510,7 +469,11 @@ class MISPEvent(AbstractMISP):
sub_event.load(rel_event)
self.RelatedEvent.append(sub_event)
if kwargs.get('Tag'):
self.Tag = [t for t in kwargs.pop('Tag', []) if t]
self.Tag = []
for tag in kwargs.pop('Tag'):
t = MISPTag()
t.from_dict(**tag)
self.Tag.append(t)
if kwargs.get('Object'):
self.Object = []
for obj in kwargs.pop('Object'):
@ -518,41 +481,64 @@ class MISPEvent(AbstractMISP):
tmp_object.from_dict(**obj)
self.Object.append(tmp_object)
for k, v in kwargs.items():
setattr(self, k, v)
super(MISPEvent, self).from_dict(**kwargs)
def _json(self):
# DEPTECATED
return self.to_dict()
def to_dict(self, with_timestamp=False):
def to_dict(self):
for o in self.objects:
if o.edited:
self.edited = True
break
for a in self.attributes:
if a.edited:
self.edited = True
break
to_return = super(MISPEvent, self).to_dict()
if to_return.get('date'):
to_return['date'] = self.date.isoformat()
if with_timestamp and to_return.get('timestamp'):
if sys.version_info >= (3, 3):
to_return['timestamp'] = self.timestamp.timestamp()
else:
from datetime import timezone # Only for Python < 3.3
to_return['timestamp'] = (self.timestamp - datetime(1970, 1, 1, tzinfo=timezone.utc)).total_seconds()
else:
if self.edited:
to_return.pop('timestamp', None)
if with_timestamp and to_return.get('publish_timestamp'):
if sys.version_info >= (3, 3):
to_return['publish_timestamp'] = self.publish_timestamp.timestamp()
else:
from datetime import timezone # Only for Python < 3.3
to_return['publish_timestamp'] = (self.publish_timestamp - datetime(1970, 1, 1, tzinfo=timezone.utc)).total_seconds()
else:
to_return.pop('publish_timestamp', None)
elif to_return.get('timestamp'):
to_return['timestamp'] = _datetime_to_timestamp(self.timestamp)
if to_return.get('publish_timestamp'):
to_return['publish_timestamp'] = _datetime_to_timestamp(self.publish_timestamp)
to_return = _int_to_str(to_return)
to_return = {'Event': to_return}
return to_return
def add_tag(self, tag):
self.Tag.append({'name': tag})
misp_tag = MISPTag()
if isinstance(tag, str):
misp_tag.from_dict(name=tag)
elif isinstance(tag, dict):
misp_tag.from_dict(**tag)
self.Tag.append(misp_tag)
def get_attribute_tag(self, attribute_identifier):
'''Return the tags associated to an attribute or an object attribute.
:attribute_identifier: can be an ID, UUID, or the value.
'''
tags = []
for a in self.attributes + [attribute for o in self.objects for attribute in o.attributes]:
if ((hasattr(a, 'id') and a.id == attribute_identifier) or
(hasattr(a, 'uuid') and a.uuid == attribute_identifier) or
(hasattr(a, 'value') and attribute_identifier == a.value or
attribute_identifier in a.value.split('|'))):
tags += a.tags
return tags
def add_attribute_tag(self, tag, attribute_identifier):
'''Add a tag to an existing attribute, raise an Exception if the attribute doesn't exists.
:tag: Tag name
:attribute_identifier: can be an ID, UUID, or the value.
'''
attributes = []
for a in self.attributes:
if ((hasattr(a, 'id') and a.id == attribute_identifier) or
@ -626,6 +612,20 @@ class MISPEvent(AbstractMISP):
raise InvalidMISPObject("An object to add to an existing Event needs to be either a MISPObject, or a plain python dictionary")
class MISPTag(AbstractMISP):
def __init__(self):
super(MISPTag, self).__init__()
def __repr__(self):
if hasattr(self, 'name'):
return '<{self.__class__.__name__}(name={self.name})'.format(self=self)
return '<{self.__class__.__name__}(NotInitialized)'.format(self=self)
def from_dict(self, name, **kwargs):
self.name = name
super(MISPTag, self).from_dict(**kwargs)
class MISPObjectReference(AbstractMISP):
def __init__(self):
@ -641,8 +641,7 @@ class MISPObjectReference(AbstractMISP):
self.referenced_uuid = referenced_uuid
self.relationship_type = relationship_type
self.comment = comment
for k, v in kwargs.items():
setattr(self, k, v)
super(MISPObjectReference, self).from_dict(**kwargs)
class MISPUser(AbstractMISP):
@ -650,20 +649,12 @@ class MISPUser(AbstractMISP):
def __init__(self):
super(MISPUser, self).__init__()
def from_dict(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
class MISPOrganisation(AbstractMISP):
def __init__(self):
super(MISPOrganisation, self).__init__()
def from_dict(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
class MISPObjectAttribute(MISPAttribute):
@ -778,22 +769,30 @@ class MISPObject(AbstractMISP):
else:
self.__known_template = False
for key, value in kwargs.items():
if key == 'Attribute':
for v in value:
self.add_attribute(**v)
elif key == 'ObjectReference':
for v in value:
self.add_reference(**v)
else:
setattr(self, key, value)
if kwargs.get('Attribute'):
for a in kwargs.pop('Attribute'):
self.add_attribute(**a)
if kwargs.get('ObjectReference'):
for r in kwargs.pop('ObjectReference'):
self.add_reference(**r)
super(MISPObject, self).from_dict(**kwargs)
def to_dict(self, strict=False):
# Set the expected key (Attributes)
self.Attribute = self.attributes
for a in self.attributes:
if a.edited:
self.edited = True
break
if strict or self.__strict and self.__known_template:
self._validate()
return super(MISPObject, self).to_dict()
# Set the expected key (Attributes)
self.Attribute = self.attributes
to_return = super(MISPObject, self).to_dict()
if self.edited:
to_return.pop('timestamp', None)
elif to_return.get('timestamp'):
to_return['timestamp'] = _datetime_to_timestamp(self.timestamp)
return to_return
def to_json(self, strict=False):
if strict or self.__strict and self.__known_template: