mirror of https://github.com/MISP/PyMISP
633 lines
32 KiB
Python
633 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from .exceptions import MISPServerError, PyMISPUnexpectedResponse, PyMISPNotImplementedYet
|
|
from .api import PyMISP, everything_broken
|
|
from .mispevent import MISPEvent, MISPAttribute, MISPSighting, MISPLog, MISPObject, MISPUser, MISPOrganisation, MISPShadowAttribute
|
|
from typing import TypeVar, Optional, Tuple, List, Dict, Union
|
|
from datetime import date, datetime
|
|
import csv
|
|
|
|
import logging
|
|
from urllib.parse import urljoin
|
|
|
|
SearchType = TypeVar('SearchType', str, int)
|
|
# str: string to search / list: values to search (OR) / dict: {'OR': [list], 'NOT': [list], 'AND': [list]}
|
|
SearchParameterTypes = TypeVar('SearchParameterTypes', str, List[SearchType], Dict[str, SearchType])
|
|
DateTypes = TypeVar('DateTypes', datetime, date, SearchType, float)
|
|
DateInterval = TypeVar('DateInterval', DateTypes, Tuple[DateTypes, DateTypes])
|
|
|
|
ToIDSType = TypeVar('ToIDSType', str, int, bool)
|
|
|
|
logger = logging.getLogger('pymisp')
|
|
|
|
|
|
class ExpandedPyMISP(PyMISP):
|
|
|
|
def build_complex_query(self, or_parameters: Optional[List[SearchType]]=None,
|
|
and_parameters: Optional[List[SearchType]]=None,
|
|
not_parameters: Optional[List[SearchType]]=None):
|
|
to_return = {}
|
|
if and_parameters:
|
|
to_return['AND'] = and_parameters
|
|
if not_parameters:
|
|
to_return['NOT'] = not_parameters
|
|
if or_parameters:
|
|
to_return['OR'] = or_parameters
|
|
return to_return
|
|
|
|
def toggle_warninglist(self, warninglist_id: List[int]=None, warninglist_name: List[str]=None, force_enable: bool=None):
|
|
'''Toggle (enable/disable) the status of a warninglist by ID.
|
|
:param warninglist_id: ID of the WarningList
|
|
:param force_enable: Force the warning list in the enabled state (does nothing is already enabled)
|
|
'''
|
|
return super().toggle_warninglist(warninglist_id, warninglist_name, force_enable)
|
|
|
|
def make_timestamp(self, value: DateTypes):
|
|
if isinstance(value, datetime):
|
|
return datetime.timestamp()
|
|
elif isinstance(value, date):
|
|
return datetime.combine(value, datetime.max.time()).timestamp()
|
|
elif isinstance(value, str):
|
|
if value.isdigit():
|
|
return value
|
|
else:
|
|
try:
|
|
float(value)
|
|
return value
|
|
except ValueError:
|
|
# The value can also be '1d', '10h', ...
|
|
return value
|
|
else:
|
|
return value
|
|
|
|
def _check_response(self, response):
|
|
"""Check if the response from the server is not an unexpected error"""
|
|
if response.status_code >= 500:
|
|
logger.critical(everything_broken.format(response.request.headers, response.request.body, response.text))
|
|
raise MISPServerError('Error code 500:\n{}'.format(response.text))
|
|
elif 400 <= response.status_code < 500:
|
|
# 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)}
|
|
|
|
# At this point, we had no error.
|
|
|
|
try:
|
|
response = response.json()
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug(response)
|
|
if isinstance(response, dict) and response.get('response') is not None:
|
|
# Cleanup.
|
|
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 {'errors': 'The response is empty.'}
|
|
return response.text
|
|
|
|
def get_event(self, event_id: int):
|
|
event = super().get_event(event_id)
|
|
e = MISPEvent()
|
|
e.load(event)
|
|
return e
|
|
|
|
def add_object(self, event_id: int, misp_object: MISPObject):
|
|
created_object = super().add_object(event_id, misp_object)
|
|
if isinstance(created_object, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {created_object}')
|
|
elif 'errors' in created_object:
|
|
return created_object
|
|
o = MISPObject(misp_object.name)
|
|
o.from_dict(**created_object)
|
|
return o
|
|
|
|
def update_object(self, misp_object: MISPObject):
|
|
updated_object = super().edit_object(misp_object)
|
|
if isinstance(updated_object, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {updated_object}')
|
|
elif 'errors' in updated_object:
|
|
return updated_object
|
|
o = MISPObject(misp_object.name)
|
|
o.from_dict(**updated_object)
|
|
return o
|
|
|
|
def get_object(self, object_id: int):
|
|
"""Get an object
|
|
|
|
:param obj_id: Object id to get
|
|
"""
|
|
misp_object = super().get_object(object_id)
|
|
if isinstance(misp_object, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {misp_object}')
|
|
elif 'errors' in misp_object:
|
|
return misp_object
|
|
o = MISPObject(misp_object['Object']['name'])
|
|
o.from_dict(**misp_object)
|
|
return o
|
|
|
|
def add_event(self, event: MISPEvent):
|
|
created_event = super().add_event(event)
|
|
if isinstance(created_event, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {created_event}')
|
|
elif 'errors' in created_event:
|
|
return created_event
|
|
e = MISPEvent()
|
|
e.load(created_event)
|
|
return e
|
|
|
|
def update_event(self, event: MISPEvent):
|
|
updated_event = super().update_event(event.uuid, event)
|
|
if isinstance(updated_event, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {updated_event}')
|
|
elif 'errors' in updated_event:
|
|
return updated_event
|
|
e = MISPEvent()
|
|
e.load(updated_event)
|
|
return e
|
|
|
|
def get_attribute(self, attribute_id: int):
|
|
attribute = super().get_attribute(attribute_id)
|
|
a = MISPAttribute()
|
|
a.from_dict(**attribute)
|
|
return a
|
|
|
|
def add_attribute(self, event_id: int, attribute: MISPAttribute):
|
|
url = urljoin(self.root_url, 'attributes/add/{}'.format(event_id))
|
|
response = self._prepare_request('POST', url, data=attribute)
|
|
new_attribute = self._check_response(response)
|
|
if isinstance(new_attribute, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {new_attribute}')
|
|
elif 'errors' in new_attribute:
|
|
return new_attribute
|
|
a = MISPAttribute()
|
|
a.from_dict(**new_attribute)
|
|
return a
|
|
|
|
def add_attribute_proposal(self, event_id: int, attribute: MISPAttribute):
|
|
url = urljoin(self.root_url, 'shadow_attributes/add/{}'.format(event_id))
|
|
response = self._prepare_request('POST', url, attribute)
|
|
new_attribute_proposal = self._check_response(response)
|
|
if isinstance(new_attribute_proposal, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {new_attribute_proposal}')
|
|
elif 'errors' in new_attribute_proposal:
|
|
return new_attribute_proposal
|
|
a = MISPShadowAttribute()
|
|
a.from_dict(**new_attribute_proposal)
|
|
return a
|
|
|
|
def update_attribute(self, attribute: MISPAttribute):
|
|
updated_attribute = super().update_attribute(attribute.uuid, attribute)
|
|
if isinstance(updated_attribute, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {updated_attribute}')
|
|
elif 'errors' in updated_attribute:
|
|
return updated_attribute
|
|
a = MISPAttribute()
|
|
a.from_dict(**updated_attribute)
|
|
return a
|
|
|
|
def update_attribute_proposal(self, attribute_id: int, attribute: MISPAttribute):
|
|
url = urljoin(self.root_url, 'shadow_attributes/edit/{}'.format(attribute_id))
|
|
# FIXME: Inconsistency on MISP side
|
|
attribute = {'ShadowAttribute': attribute}
|
|
response = self._prepare_request('POST', url, attribute)
|
|
attribute_proposal = self._check_response(response)
|
|
if isinstance(attribute_proposal, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {attribute_proposal}')
|
|
elif 'errors' in attribute_proposal:
|
|
return attribute_proposal
|
|
a = MISPShadowAttribute()
|
|
a.from_dict(**attribute_proposal)
|
|
return a
|
|
|
|
def get_attribute_proposal(self, proposal_id: int):
|
|
url = urljoin(self.root_url, 'shadow_attributes/view/{}'.format(proposal_id))
|
|
response = self._prepare_request('GET', url)
|
|
attribute_proposal = self._check_response(response)
|
|
if isinstance(attribute_proposal, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {attribute_proposal}')
|
|
elif 'errors' in attribute_proposal:
|
|
return attribute_proposal
|
|
a = MISPShadowAttribute()
|
|
a.from_dict(**attribute_proposal)
|
|
return a
|
|
|
|
def accept_attribute_proposal(self, proposal_id: int):
|
|
url = urljoin(self.root_url, 'shadow_attributes/accept/{}'.format(proposal_id))
|
|
response = self._prepare_request('POST', url)
|
|
r = self._check_response(response)
|
|
if isinstance(r, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {r}')
|
|
return r
|
|
|
|
def discard_attribute_proposal(self, proposal_id: int):
|
|
url = urljoin(self.root_url, 'shadow_attributes/discard/{}'.format(proposal_id))
|
|
response = self._prepare_request('POST', url)
|
|
r = self._check_response(response)
|
|
if isinstance(r, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {r}')
|
|
return r
|
|
|
|
def add_user(self, user: MISPUser):
|
|
user = super().add_user(user)
|
|
if isinstance(user, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {user}')
|
|
elif 'errors' in user:
|
|
return user
|
|
u = MISPUser()
|
|
u.from_dict(**user)
|
|
return u
|
|
|
|
def get_user(self, userid='me'):
|
|
user = super().get_user(userid)
|
|
if isinstance(user, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {user}')
|
|
elif 'errors' in user:
|
|
return user
|
|
u = MISPUser()
|
|
u.from_dict(**user)
|
|
return u
|
|
|
|
def add_organisation(self, organisation: MISPOrganisation):
|
|
organisation = super().add_organisation(organisation)
|
|
if isinstance(organisation, str):
|
|
raise PyMISPUnexpectedResponse(f'Unexpected response from server: {organisation}')
|
|
elif 'errors' in organisation:
|
|
return organisation
|
|
o = MISPOrganisation()
|
|
o.from_dict(**organisation)
|
|
return o
|
|
|
|
def search_sightings(self, context: Optional[str]=None,
|
|
context_id: Optional[SearchType]=None,
|
|
type_sighting: Optional[str]=None,
|
|
date_from: Optional[DateTypes]=None,
|
|
date_to: Optional[DateTypes]=None,
|
|
publish_timestamp: Optional[DateInterval]=None, last: Optional[DateInterval]=None,
|
|
org: Optional[SearchType]=None,
|
|
source: Optional[str]=None,
|
|
include_attribute: Optional[bool]=None,
|
|
include_event_meta: Optional[bool]=None,
|
|
pythonify: Optional[bool]=False
|
|
):
|
|
'''Search sightings
|
|
|
|
:param context: The context of the search. Can be either "attribute", "event", or nothing (will then match on events and attributes).
|
|
:param context_id: Only relevant if context is either "attribute" or "event". Then it is the relevant ID.
|
|
:param type_sighting: Type of sighting
|
|
:param date_from: Events with the date set to a date after the one specified. This filter will use the date of the event.
|
|
:param date_to: Events with the date set to a date before the one specified. This filter will use the date of the event.
|
|
:param publish_timestamp: Restrict the results by the last publish timestamp (newer than).
|
|
:param org: Search by the creator organisation by supplying the organisation identifier.
|
|
:param source: Source of the sighting
|
|
:param include_attribute: Include the attribute.
|
|
:param include_event_meta: Include the meta information of the event.
|
|
|
|
Deprecated:
|
|
|
|
:param last: synonym for publish_timestamp
|
|
|
|
:Example:
|
|
|
|
>>> misp.search_sightings(publish_timestamp='30d') # search sightings for the last 30 days on the instance
|
|
[ ... ]
|
|
>>> misp.search_sightings(context='attribute', context_id=6, include_attribute=True) # return list of sighting for attribute 6 along with the attribute itself
|
|
[ ... ]
|
|
>>> misp.search_sightings(context='event', context_id=17, include_event_meta=True, org=2) # return list of sighting for event 17 filtered with org id 2
|
|
'''
|
|
query = {'returnFormat': 'json'}
|
|
if context is not None:
|
|
if context not in ['attribute', 'event']:
|
|
raise ValueError('context has to be in {}'.format(', '.join(['attribute', 'event'])))
|
|
url_path = f'sightings/restSearch/{context}'
|
|
else:
|
|
url_path = 'sightings/restSearch'
|
|
query['id'] = context_id
|
|
query['type'] = type_sighting
|
|
query['from'] = date_from
|
|
query['to'] = date_to
|
|
query['last'] = publish_timestamp
|
|
query['org_id'] = org
|
|
query['source'] = source
|
|
query['includeAttribute'] = include_attribute
|
|
query['includeEvent'] = include_event_meta
|
|
|
|
url = urljoin(self.root_url, url_path)
|
|
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')):
|
|
return normalized_response
|
|
elif pythonify:
|
|
to_return = []
|
|
for s in normalized_response:
|
|
entries = {}
|
|
s_data = s['Sighting']
|
|
if include_event_meta:
|
|
e = s_data.pop('Event')
|
|
me = MISPEvent()
|
|
me.from_dict(**e)
|
|
entries['event'] = me
|
|
if include_attribute:
|
|
a = s_data.pop('Attribute')
|
|
ma = MISPAttribute()
|
|
ma.from_dict(**a)
|
|
entries['attribute'] = ma
|
|
ms = MISPSighting()
|
|
ms.from_dict(**s_data)
|
|
entries['sighting'] = ms
|
|
to_return.append(entries)
|
|
return to_return
|
|
else:
|
|
return normalized_response
|
|
|
|
def search(self, controller: str='events', return_format: str='json',
|
|
limit: Optional[int]=None, page: Optional[int]=None,
|
|
value: Optional[SearchParameterTypes]=None,
|
|
type_attribute: Optional[SearchParameterTypes]=None,
|
|
category: Optional[SearchParameterTypes]=None,
|
|
org: Optional[SearchParameterTypes]=None,
|
|
tags: Optional[SearchParameterTypes]=None,
|
|
quick_filter: Optional[str]=None, quickFilter: Optional[str]=None,
|
|
date_from: Optional[DateTypes]=None,
|
|
date_to: Optional[DateTypes]=None,
|
|
eventid: Optional[SearchType]=None,
|
|
with_attachments: Optional[bool]=None, withAttachments: Optional[bool]=None,
|
|
metadata: Optional[bool]=None,
|
|
uuid: Optional[str]=None,
|
|
publish_timestamp: Optional[DateInterval]=None, last: Optional[DateInterval]=None,
|
|
timestamp: Optional[DateInterval]=None,
|
|
published: Optional[bool]=None,
|
|
enforce_warninglist: Optional[bool]=None, enforceWarninglist: Optional[bool]=None,
|
|
to_ids: Optional[Union[ToIDSType, List[ToIDSType]]]=None,
|
|
deleted: Optional[str]=None,
|
|
include_event_uuid: Optional[str]=None, includeEventUuid: Optional[str]=None,
|
|
event_timestamp: Optional[DateTypes]=None,
|
|
sg_reference_only: Optional[bool]=None,
|
|
eventinfo: Optional[str]=None,
|
|
searchall: Optional[bool]=None,
|
|
requested_attributes: Optional[str]=None,
|
|
include_context: Optional[bool]=None, includeContext: Optional[bool]=None,
|
|
headerless: Optional[bool]=None,
|
|
pythonify: Optional[bool]=False,
|
|
**kwargs):
|
|
'''Search in the MISP instance
|
|
|
|
:param returnFormat: Set the return format of the search (Currently supported: json, xml, openioc, suricata, snort - more formats are being moved to restSearch with the goal being that all searches happen through this API). Can be passed as the first parameter after restSearch or via the JSON payload.
|
|
:param limit: Limit the number of results returned, depending on the scope (for example 10 attributes or 10 full events).
|
|
:param page: If a limit is set, sets the page to be returned. page 3, limit 100 will return records 201->300).
|
|
:param value: Search for the given value in the attributes' value field.
|
|
:param type_attribute: The attribute type, any valid MISP attribute type is accepted.
|
|
:param category: The attribute category, any valid MISP attribute category is accepted.
|
|
:param org: Search by the creator organisation by supplying the organisation identifier.
|
|
:param tags: Tags to search or to exclude. You can pass a list, or the output of `build_complex_query`
|
|
:param quick_filter: The string passed to this field will ignore all of the other arguments. MISP will return an xml / json (depending on the header sent) of all events that have a sub-string match on value in the event info, event orgc, or any of the attribute value1 / value2 fields, or in the attribute comment.
|
|
:param date_from: Events with the date set to a date after the one specified. This filter will use the date of the event.
|
|
:param date_to: Events with the date set to a date before the one specified. This filter will use the date of the event.
|
|
:param eventid: The events that should be included / excluded from the search
|
|
:param with_attachments: If set, encodes the attachments / zipped malware samples as base64 in the data field within each attribute
|
|
:param metadata: Only the metadata (event, tags, relations) is returned, attributes and proposals are omitted.
|
|
:param uuid: Restrict the results by uuid.
|
|
:param publish_timestamp: Restrict the results by the last publish timestamp (newer than).
|
|
:param timestamp: Restrict the results by the timestamp (last edit). Any event with a timestamp newer than the given timestamp will be returned. In case you are dealing with /attributes as scope, the attribute's timestamp will be used for the lookup.
|
|
:param published: Set whether published or unpublished events should be returned. Do not set the parameter if you want both.
|
|
:param enforce_warninglist: Remove any attributes from the result that would cause a hit on a warninglist entry.
|
|
:param to_ids: By default all attributes are returned that match the other filter parameters, irregardless of their to_ids setting. To restrict the returned data set to to_ids only attributes set this parameter to 1. 0 for the ones with to_ids set to False.
|
|
:param deleted: If this parameter is set to 1, it will return soft-deleted attributes along with active ones. By using "only" as a parameter it will limit the returned data set to soft-deleted data only.
|
|
:param include_event_uuid: Instead of just including the event ID, also include the event UUID in each of the attributes.
|
|
:param event_timestamp: Only return attributes from events that have received a modification after the given timestamp.
|
|
:param sg_reference_only: If this flag is set, sharing group objects will not be included, instead only the sharing group ID is set.
|
|
:param eventinfo: Filter on the event's info field.
|
|
:param searchall: Search for a full or a substring (delimited by % for substrings) in the event info, event tags, attribute tags, attribute values or attribute comment fields.
|
|
:param requested_attributes: [CSV only] Select the fields that you wish to include in the CSV export. By setting event level fields additionally, includeContext is not required to get event metadata.
|
|
:param include_context: [CSV Only] Include the event data with each attribute.
|
|
:param headerless: [CSV Only] The CSV created when this setting is set to true will not contain the header row.
|
|
:param pythonify: Returns a list of PyMISP Objects instead of the plain json output. Warning: it might use a lot of RAM
|
|
|
|
Deprecated:
|
|
|
|
:param quickFilter: synponym for quick_filter
|
|
:param withAttachments: synonym for with_attachments
|
|
:param last: synonym for publish_timestamp
|
|
:param enforceWarninglist: synonym for enforce_warninglist
|
|
:param includeEventUuid: synonym for include_event_uuid
|
|
:param includeContext: synonym for include_context
|
|
|
|
'''
|
|
|
|
return_formats = ['openioc', 'json', 'xml', 'suricata', 'snort', 'text', 'rpz', 'csv', 'cache', 'stix', 'stix2']
|
|
|
|
if controller not in ['events', 'attributes', 'objects', 'sightings']:
|
|
raise ValueError('controller has to be in {}'.format(', '.join(['events', 'attributes', 'objects'])))
|
|
|
|
# Deprecated stuff / synonyms
|
|
if quickFilter is not None:
|
|
quick_filter = quickFilter
|
|
if withAttachments is not None:
|
|
with_attachments = withAttachments
|
|
if last is not None:
|
|
publish_timestamp = last
|
|
if enforceWarninglist is not None:
|
|
enforce_warninglist = enforceWarninglist
|
|
if includeEventUuid is not None:
|
|
include_event_uuid = includeEventUuid
|
|
if includeContext is not None:
|
|
include_context = includeContext
|
|
|
|
# Add all the parameters in kwargs are aimed at modules, or other 3rd party components, and cannot be sanitized.
|
|
# They are passed as-is.
|
|
query = kwargs
|
|
|
|
if return_format not in return_formats:
|
|
raise ValueError('return_format has to be in {}'.format(', '.join(return_formats)))
|
|
query['returnFormat'] = return_format
|
|
|
|
query['page'] = page
|
|
query['limit'] = limit
|
|
query['value'] = value
|
|
query['type'] = type_attribute
|
|
query['category'] = category
|
|
query['org'] = org
|
|
query['tags'] = tags
|
|
query['quickFilter'] = quick_filter
|
|
query['from'] = self.make_timestamp(date_from)
|
|
query['to'] = self.make_timestamp(date_to)
|
|
query['eventid'] = eventid
|
|
query['withAttachments'] = with_attachments
|
|
query['metadata'] = metadata
|
|
query['uuid'] = uuid
|
|
if publish_timestamp is not None:
|
|
if isinstance(publish_timestamp, (list, tuple)):
|
|
query['publish_timestamp'] = (self.make_timestamp(publish_timestamp[0]), self.make_timestamp(publish_timestamp[1]))
|
|
else:
|
|
query['publish_timestamp'] = self.make_timestamp(publish_timestamp)
|
|
if timestamp is not None:
|
|
if isinstance(timestamp, (list, tuple)):
|
|
query['timestamp'] = (self.make_timestamp(timestamp[0]), self.make_timestamp(timestamp[1]))
|
|
else:
|
|
query['timestamp'] = self.make_timestamp(timestamp)
|
|
query['published'] = published
|
|
query['enforceWarninglist'] = enforce_warninglist
|
|
if to_ids is not None:
|
|
if int(to_ids) not in [0, 1]:
|
|
raise ValueError('to_ids has to be in {}'.format(', '.join([0, 1])))
|
|
query['to_ids'] = to_ids
|
|
query['deleted'] = deleted
|
|
query['includeEventUuid'] = include_event_uuid
|
|
if event_timestamp is not None:
|
|
if isinstance(event_timestamp, (list, tuple)):
|
|
query['event_timestamp'] = (self.make_timestamp(event_timestamp[0]), self.make_timestamp(event_timestamp[1]))
|
|
else:
|
|
query['event_timestamp'] = self.make_timestamp(event_timestamp)
|
|
query['sgReferenceOnly'] = sg_reference_only
|
|
query['eventinfo'] = eventinfo
|
|
query['searchall'] = searchall
|
|
query['requested_attributes'] = requested_attributes
|
|
query['includeContext'] = include_context
|
|
query['headerless'] = headerless
|
|
url = urljoin(self.root_url, f'{controller}/restSearch')
|
|
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)
|
|
elif isinstance(normalized_response, str) or (isinstance(normalized_response, dict)
|
|
and normalized_response.get('errors')):
|
|
return normalized_response
|
|
elif return_format == 'json' and pythonify:
|
|
# The response is in json, we can convert it to a list of pythonic MISP objects
|
|
to_return = []
|
|
if controller == 'events':
|
|
for e in normalized_response:
|
|
me = MISPEvent()
|
|
me.load(e)
|
|
to_return.append(me)
|
|
elif controller == 'attributes':
|
|
for a in normalized_response.get('Attribute'):
|
|
ma = MISPAttribute()
|
|
ma.from_dict(**a)
|
|
to_return.append(ma)
|
|
elif controller == 'objects':
|
|
raise PyMISPNotImplementedYet('Not implemented yet')
|
|
return to_return
|
|
else:
|
|
return normalized_response
|
|
|
|
def _csv_to_dict(self, csv_content):
|
|
'''Makes a list of dict out of a csv file (requires headers)'''
|
|
fieldnames, lines = csv_content.split('\n', 1)
|
|
fieldnames = fieldnames.split(',')
|
|
to_return = []
|
|
for line in csv.reader(lines.split('\n')):
|
|
if line:
|
|
to_return.append({fname: value for fname, value in zip(fieldnames, line)})
|
|
return to_return
|
|
|
|
def search_logs(self, limit: Optional[int]=None, page: Optional[int]=None,
|
|
log_id: Optional[int]=None, title: Optional[str]=None,
|
|
created: Optional[DateTypes]=None, model: Optional[str]=None,
|
|
action: Optional[str]=None, user_id: Optional[int]=None,
|
|
change: Optional[str]=None, email: Optional[str]=None,
|
|
org: Optional[str]=None, description: Optional[str]=None,
|
|
ip: Optional[str]=None, pythonify: Optional[bool]=False):
|
|
'''Search in logs
|
|
|
|
Note: to run substring queries simply append/prepend/encapsulate the search term with %
|
|
|
|
:param limit: Limit the number of results returned, depending on the scope (for example 10 attributes or 10 full events).
|
|
:param page: If a limit is set, sets the page to be returned. page 3, limit 100 will return records 201->300).
|
|
:param log_id: Log ID
|
|
:param title: Log Title
|
|
:param created: Creation timestamp
|
|
:param model: Model name that generated the log entry
|
|
:param action: The thing that was done
|
|
:param user_id: ID of the user doing the action
|
|
:param change: Change that occured
|
|
:param email: Email of the user
|
|
:param org: Organisation of the User doing the action
|
|
:param description: Description of the action
|
|
:param ip: Origination IP of the User doing the action
|
|
:param pythonify: Returns a list of PyMISP Objects instead or the plain json output. Warning: it might use a lot of RAM
|
|
'''
|
|
query = locals()
|
|
query.pop('self')
|
|
query.pop('pythonify')
|
|
if log_id is not None:
|
|
query['id'] = query.pop('log_id')
|
|
|
|
url = urljoin(self.root_url, 'admin/logs/index')
|
|
response = self._prepare_request('POST', url, data=query)
|
|
normalized_response = self._check_response(response)
|
|
if not pythonify:
|
|
return normalized_response
|
|
|
|
to_return = []
|
|
for l in normalized_response:
|
|
ml = MISPLog()
|
|
ml.from_dict(**l['Log'])
|
|
to_return.append(ml)
|
|
return to_return
|
|
|
|
def search_index(self, published: Optional[bool]=None, eventid: Optional[SearchType]=None,
|
|
tags: Optional[SearchParameterTypes]=None,
|
|
date_from: Optional[DateTypes]=None,
|
|
date_to: Optional[DateTypes]=None,
|
|
eventinfo: Optional[str]=None,
|
|
threatlevel: Optional[List[SearchType]]=None,
|
|
distribution: Optional[List[SearchType]]=None,
|
|
analysis: Optional[List[SearchType]]=None,
|
|
org: Optional[SearchParameterTypes]=None,
|
|
timestamp: Optional[DateInterval]=None,
|
|
pythonify: Optional[bool]=None):
|
|
"""Search only at the index level. Using ! in front of a value means NOT (default is OR)
|
|
|
|
:param published: Set whether published or unpublished events should be returned. Do not set the parameter if you want both.
|
|
:param eventid: The events that should be included / excluded from the search
|
|
:param tags: Tags to search or to exclude. You can pass a list, or the output of `build_complex_query`
|
|
:param date_from: Events with the date set to a date after the one specified. This filter will use the date of the event.
|
|
:param date_to: Events with the date set to a date before the one specified. This filter will use the date of the event.
|
|
:param eventinfo: Filter on the event's info field.
|
|
:param threatlevel: Threat level(s) (1,2,3,4) | list
|
|
:param distribution: Distribution level(s) (0,1,2,3) | list
|
|
:param analysis: Analysis level(s) (0,1,2) | list
|
|
:param org: Search by the creator organisation by supplying the organisation identifier.
|
|
:param timestamp: Restrict the results by the timestamp (last edit). Any event with a timestamp newer than the given timestamp will be returned. In case you are dealing with /attributes as scope, the attribute's timestamp will be used for the lookup.
|
|
:param pythonify: Returns a list of PyMISP Objects instead or the plain json output. Warning: it might use a lot of RAM
|
|
"""
|
|
query = locals()
|
|
query.pop('self')
|
|
query.pop('pythonify')
|
|
if query.get('date_from'):
|
|
query['datefrom'] = self.make_timestamp(query.pop('date_from'))
|
|
if query.get('date_to'):
|
|
query['dateuntil'] = self.make_timestamp(query.pop('date_to'))
|
|
|
|
if query.get('timestamp') is not None:
|
|
timestamp = query.pop('timestamp')
|
|
if isinstance(timestamp, (list, tuple)):
|
|
query['timestamp'] = (self.make_timestamp(timestamp[0]), self.make_timestamp(timestamp[1]))
|
|
else:
|
|
query['timestamp'] = self.make_timestamp(timestamp)
|
|
|
|
url = urljoin(self.root_url, 'events/index')
|
|
response = self._prepare_request('POST', url, data=query)
|
|
normalized_response = self._check_response(response)
|
|
|
|
if not pythonify:
|
|
return normalized_response
|
|
to_return = []
|
|
for e_meta in normalized_response:
|
|
me = MISPEvent()
|
|
me.from_dict(**e_meta)
|
|
to_return.append(me)
|
|
return to_return
|
|
|
|
def set_default_role(self, role_id: int):
|
|
url = urljoin(self.root_url, f'/admin/roles/set_default/{role_id}')
|
|
response = self._prepare_request('POST', url)
|
|
return self._check_response(response)
|