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

chg: Remove compat for MISP 2.4.52, cleanup.

pull/317/head
Raphaël Vinot 2018-12-18 11:04:36 +01:00
parent 864aa5fe70
commit a09915d850
5 changed files with 26 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pymisp/__init__.py
View File

@ -32,7 +32,7 @@ def deprecated(func):
try:
from .exceptions import PyMISPError, NewEventError, NewAttributeError, MissingDependency, NoURL, NoKey, InvalidMISPObject, UnknownMISPObjectTemplate, PyMISPInvalidFormat, MISPServerError, PyMISPNotImplementedYet, PyMISPUnexpectedResponse # noqa
from .exceptions import PyMISPError, NewEventError, NewAttributeError, MissingDependency, NoURL, NoKey, InvalidMISPObject, UnknownMISPObjectTemplate, PyMISPInvalidFormat, MISPServerError, PyMISPNotImplementedYet, PyMISPUnexpectedResponse, PyMISPEmptyResponse # noqa
from .api import PyMISP # noqa
from .abstract import AbstractMISP, MISPEncode, MISPTag, Distribution, ThreatLevel, Analysis # noqa
from .mispevent import MISPEvent, MISPAttribute, MISPObjectReference, MISPObjectAttribute, MISPObject, MISPUser, MISPOrganisation, MISPSighting, MISPLog # noqa

9
pymisp/api.py
View File

@ -15,7 +15,7 @@ from io import BytesIO, open
import zipfile
from . import __version__, deprecated
from .exceptions import PyMISPError, SearchError, NoURL, NoKey
from .exceptions import PyMISPError, SearchError, NoURL, NoKey, PyMISPEmptyResponse
from .mispevent import MISPEvent, MISPAttribute, MISPUser, MISPOrganisation, MISPSighting, MISPFeed, MISPObject
from .abstract import AbstractMISP, MISPEncode
@ -157,6 +157,11 @@ class PyMISP(object):
if data is None:
req = requests.Request(request_type, url)
else:
if not isinstance(data, str):
if isinstance(data, dict):
# Remove None values.
data = {k: v for k, v in data.items() if v is not None}
data = json.dumps(data)
req = requests.Request(request_type, url, data=data)
if self.asynch and background_callback is not None:
local_session = FuturesSession
@ -227,6 +232,8 @@ class PyMISP(object):
json_response = response.json()
except ValueError:
# If the server didn't return a JSON blob, we've a problem.
if not len(response.text):
raise PyMISPEmptyResponse('The server returned an empty response. \n{}\n{}\n'.format(response.request.headers, response.request.body))
raise PyMISPError(everything_broken.format(response.request.headers, response.request.body, response.text))
errors = []

29
pymisp/aping.py
View File

@ -6,7 +6,6 @@ from .api import PyMISP, everything_broken
from .mispevent import MISPEvent, MISPAttribute, MISPSighting, MISPLog
from typing import TypeVar, Optional, Tuple, List, Dict
from datetime import date, datetime
import json
import csv
import logging
@ -70,7 +69,7 @@ class ExpandedPyMISP(PyMISP):
# The server returns a json message with the error details
error_message = response.json()
logger.error(f'Something went wrong ({response.status_code}): {error_message}')
return {'errors': [(response.status_code, error_message)]}
return {'error': (response.status_code, error_message)}
# At this point, we had no error.
@ -80,11 +79,15 @@ class ExpandedPyMISP(PyMISP):
logger.debug(response)
if isinstance(response, dict) and response.get('response') is not None:
# Cleanup.
return response.get('response')
response = response['response']
return response
except Exception:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(response.text)
if not len(response.content):
# Empty response
logger.error('Got an empty response.')
return {'error': 'The response is empty.'}
return response.text
def get_event(self, event_id: int):
@ -178,10 +181,7 @@ class ExpandedPyMISP(PyMISP):
query['includeEvent'] = include_event_meta
url = urljoin(self.root_url, url_path)
# Remove None values.
# TODO: put that in self._prepare_request
query = {k: v for k, v in query.items() if v is not None}
response = self._prepare_request('POST', url, data=json.dumps(query))
response = self._prepare_request('POST', url, data=query)
normalized_response = self._check_response(response)
if isinstance(normalized_response, str) or (isinstance(normalized_response, dict) and
normalized_response.get('errors')):
@ -354,10 +354,7 @@ class ExpandedPyMISP(PyMISP):
query['includeContext'] = include_context
query['headerless'] = headerless
url = urljoin(self.root_url, f'{controller}/restSearch')
# Remove None values.
# TODO: put that in self._prepare_request
query = {k: v for k, v in query.items() if v is not None}
response = self._prepare_request('POST', url, data=json.dumps(query))
response = self._prepare_request('POST', url, data=query)
normalized_response = self._check_response(response)
if return_format == 'csv' and pythonify and not headerless:
return self._csv_to_dict(normalized_response)
@ -426,10 +423,7 @@ class ExpandedPyMISP(PyMISP):
query['id'] = query.pop('log_id')
url = urljoin(self.root_url, 'admin/logs/index')
# Remove None values.
# TODO: put that in self._prepare_request
query = {k: v for k, v in query.items() if v is not None}
response = self._prepare_request('POST', url, data=json.dumps(query))
response = self._prepare_request('POST', url, data=query)
normalized_response = self._check_response(response)
if not pythonify:
return normalized_response
@ -483,10 +477,7 @@ class ExpandedPyMISP(PyMISP):
query['timestamp'] = self.make_timestamp(timestamp)
url = urljoin(self.root_url, 'events/index')
# Remove None values.
# TODO: put that in self._prepare_request
query = {k: v for k, v in query.items() if v is not None}
response = self._prepare_request('POST', url, data=json.dumps(query))
response = self._prepare_request('POST', url, data=query)
normalized_response = self._check_response(response)
if not pythonify:

4
pymisp/exceptions.py
View File

@ -67,3 +67,7 @@ class PyMISPNotImplementedYet(PyMISPError):
class PyMISPUnexpectedResponse(PyMISPError):
pass
class PyMISPEmptyResponse(PyMISPError):
pass

9
pymisp/mispevent.py
View File

@ -464,12 +464,7 @@ class MISPEvent(AbstractMISP):
event = json_event
if not event:
raise PyMISPError('Invalid event')
# Invalid event created by MISP up to 2.4.52 (attribute_count is none instead of '0')
if (event.get('Event') and
'attribute_count' in event.get('Event') and
event.get('Event').get('attribute_count') is None):
event['Event']['attribute_count'] = '0'
self.from_dict(**event['Event'])
self.from_dict(**event)
if validate:
jsonschema.validate(json.loads(self.to_json()), self.__json_schema)
@ -488,6 +483,8 @@ class MISPEvent(AbstractMISP):
raise NewEventError('Invalid format for the date: {} - {}'.format(date, type(date)))
def from_dict(self, **kwargs):
if kwargs.get('Event'):
kwargs = kwargs.get('Event')
# Required value
self.info = kwargs.pop('info', None)
if self.info is None: