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

Search paramaters validation. 

Fix #96
pull/98/head
Raphaël Vinot 2017-07-18 12:07:47 +02:00
parent 59f9b37b5e
commit fe829b8f4d
1 changed files with 35 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
pymisp/api.py
View File

@ -920,78 +920,63 @@ class PyMISP(object):
:param deleted: also return the deleted attributes :param deleted: also return the deleted attributes
:param async_callback: The function to run when results are returned :param async_callback: The function to run when results are returned
""" """
query = {}
# Event: array('value', 'type', 'category', 'org', 'tags', 'from', 'to', 'last', 'eventid', 'withAttachments', 'uuid', 'publish_timestamp', 'timestamp', 'enforceWarninglist', 'searchall', 'metadata', 'published'); # Event: array('value', 'type', 'category', 'org', 'tags', 'from', 'to', 'last', 'eventid', 'withAttachments', 'uuid', 'publish_timestamp', 'timestamp', 'enforceWarninglist', 'searchall', 'metadata', 'published');
# Attribute: array('value', 'type', 'category', 'org', 'tags', 'from', 'to', 'last', 'eventid', 'withAttachments', 'uuid', 'publish_timestamp', 'timestamp', 'enforceWarninglist', 'to_ids', 'deleted'); # Attribute: array('value', 'type', 'category', 'org', 'tags', 'from', 'to', 'last', 'eventid', 'withAttachments', 'uuid', 'publish_timestamp', 'timestamp', 'enforceWarninglist', 'to_ids', 'deleted');
val = self.__prepare_rest_search(kwargs.get('values'), kwargs.get('not_values')) val = self.__prepare_rest_search(kwargs.pop('values', None), kwargs.pop('not_values', None))
query = {}
if len(val) != 0: if len(val) != 0:
query['value'] = val query['value'] = val
if kwargs.get('type_attribute'): query['type'] = kwargs.pop('type_attribute', None)
query['type'] = kwargs.get('type_attribute') query['category'] = kwargs.pop('category', None)
query['org'] = kwargs.pop('org', None)
if kwargs.get('category'): tag = self.__prepare_rest_search(kwargs.pop('tags', None), kwargs.pop('not_tags', None))
query['category'] = kwargs.get('category')
if kwargs.get('org') is not None:
query['org'] = kwargs.get('org')
tag = self.__prepare_rest_search(kwargs.get('tags'), kwargs.get('not_tags'))
if len(tag) != 0: if len(tag) != 0:
query['tags'] = tag query['tags'] = tag
if kwargs.get('date_from'): date_from = kwargs.pop('date_from', None)
if isinstance(kwargs.get('date_from'), datetime.date) or isinstance(kwargs.get('date_from'), datetime.datetime): if date_from:
query['from'] = kwargs.get('date_from').strftime('%Y-%m-%d') if isinstance(date_from, datetime.date) or isinstance(date_from, datetime.datetime):
query['from'] = date_from.strftime('%Y-%m-%d')
else: else:
query['from'] = kwargs.get('date_from') query['from'] = date_from
if kwargs.get('date_to'): date_to = kwargs.pop('date_to', None)
if isinstance(kwargs.get('date_to'), datetime.date) or isinstance(kwargs.get('date_to'), datetime.datetime): if date_to:
query['to'] = kwargs.get('date_to').strftime('%Y-%m-%d') if isinstance(date_to, datetime.date) or isinstance(date_to, datetime.datetime):
query['to'] = date_to.strftime('%Y-%m-%d')
else: else:
query['to'] = kwargs.get('date_to') query['to'] = date_to
if kwargs.get('last'): query['last'] = kwargs.pop('last', None)
query['last'] = kwargs.get('last') query['eventid'] = kwargs.pop('eventid', None)
query['withAttachments'] = kwargs.pop('withAttachments', None)
if kwargs.get('eventid'): uuid = kwargs.pop('uuid', None)
query['eventid'] = kwargs.get('eventid') if uuid:
if self._valid_uuid(uuid):
if kwargs.get('withAttachments'): query['uuid'] = uuid
query['withAttachments'] = kwargs.get('withAttachments')
if kwargs.get('uuid'):
if self._valid_uuid(kwargs.get('uuid')):
query['uuid'] = kwargs.get('uuid')
else: else:
return {'error': 'You must enter a valid uuid.'} return {'error': 'You must enter a valid uuid.'}
if kwargs.get('publish_timestamp'): query['publish_timestamp'] = kwargs.pop('publish_timestamp', None)
query['publish_timestamp'] = kwargs.get('publish_timestamp') query['timestamp'] = kwargs.pop('timestamp', None)
query['enforceWarninglist'] = kwargs.pop('enforceWarninglist', None)
if kwargs.get('timestamp'): query['to_ids'] = kwargs.pop('to_ids', None)
query['timestamp'] = kwargs.get('timestamp') query['deleted'] = kwargs.pop('deleted', None)
if kwargs.get('enforceWarninglist'):
query['enforceWarninglist'] = kwargs.get('enforceWarninglist')
if kwargs.get('to_ids') is not None:
query['to_ids'] = kwargs.get('to_ids')
if kwargs.get('deleted') is not None:
query['deleted'] = kwargs.get('deleted')
if controller == 'events': if controller == 'events':
# Event search only: # Event search only:
if kwargs.get('searchall'): query['searchall'] = kwargs.pop('searchall', None)
query['searchall'] = kwargs.get('searchall') query['metadata'] = kwargs.pop('metadata', None)
query['published'] = kwargs.pop('published', None)
if kwargs.get('metadata') is not None: # Cleanup
query['metadata'] = kwargs.get('metadata') query = {k: v for k, v in query.items() if v is not None}
if kwargs.get('published') is not None: if kwargs:
query['published'] = kwargs.get('published') raise SearchError('Unused parameter: {}'.format(', '.join(kwargs.keys())))
# Create a session, make it async if and only if we have a callback # Create a session, make it async if and only if we have a callback
session = self.__prepare_session(async_implemented=(async_callback is not None)) session = self.__prepare_session(async_implemented=(async_callback is not None))