#!/usr/bin/env python # -*- coding: utf-8 -*- """ Python API using the REST interface of MISP """ import json import datetime import os import base64 import re try: from urllib.parse import urljoin except ImportError: from urlparse import urljoin from io import BytesIO import zipfile import warnings import functools try: import requests HAVE_REQUESTS = True except ImportError: HAVE_REQUESTS = False from . import __version__ # Least dirty way to support python 2 and 3 try: basestring except NameError: basestring = str class PyMISPError(Exception): def __init__(self, message): super(PyMISPError, self).__init__(message) self.message = message class NewEventError(PyMISPError): pass class NewAttributeError(PyMISPError): pass class SearchError(PyMISPError): pass class MissingDependency(PyMISPError): pass class NoURL(PyMISPError): pass class NoKey(PyMISPError): pass def deprecated(func): '''This is a decorator which can be used to mark functions as deprecated. It will result in a warning being emitted when the function is used.''' @functools.wraps(func) def new_func(*args, **kwargs): warnings.warn_explicit( "Call to deprecated function {}.".format(func.__name__), category=DeprecationWarning, filename=func.__code__.co_filename, lineno=func.__code__.co_firstlineno + 1 ) return func(*args, **kwargs) return new_func class PyMISP(object): """ Python API for MISP :param url: URL of the MISP instance you want to connect to :param key: API key of the user you want to use :param ssl: can be True or False (to check ot not the validity of the certificate. Or a CA_BUNDLE in case of self signed certiifcate (the concatenation of all the *.crt of the chain) :param out_type: Type of object (json or xml) """ def __init__(self, url, key, ssl=True, out_type='json', debug=False): if not url: raise NoURL('Please provide the URL of your MISP instance.') if not key: raise NoKey('Please provide your authorization key.') self.root_url = url self.key = key self.ssl = ssl self.out_type = out_type self.debug = debug self.categories = ['Internal reference', 'Targeting data', 'Antivirus detection', 'Payload delivery', 'Payload installation', 'Artifacts dropped', 'Persistence mechanism', 'Network activity', 'Payload type', 'Attribution', 'External analysis', 'Other'] self.types = ['md5', 'sha1', 'sha256', 'filename', 'filename|md5', 'filename|sha1', 'filename|sha256', 'ip-src', 'ip-dst', 'hostname', 'domain', 'url', 'user-agent', 'http-method', 'regkey', 'regkey|value', 'AS', 'snort', 'pattern-in-file', 'pattern-in-traffic', 'pattern-in-memory', 'named pipe', 'mutex', 'vulnerability', 'attachment', 'malware-sample', 'link', 'comment', 'text', 'email-src', 'email-dst', 'email-subject', 'email-attachment', 'yara', 'target-user', 'target-email', 'target-machine', 'target-org', 'target-location', 'target-external', 'other', 'threat-actor'] try: # Make sure the MISP instance is working and the URL is valid self.get_version() except Exception as e: raise PyMISPError('Unable to connect to MISP ({}). Please make sure the API key and the URL are correct (http/https is required): {}'.format(self.root_url, e)) def __prepare_session(self, force_out=None): """ Prepare the headers of the session :param force_out: force the type of the expect output (overwrite the constructor) """ if not HAVE_REQUESTS: raise MissingDependency('Missing dependency, install requests (`pip install requests`)') if force_out is not None: out = force_out else: out = self.out_type session = requests.Session() session.verify = self.ssl session.headers.update( {'Authorization': self.key, 'Accept': 'application/' + out, 'content-type': 'application/' + out}) return session def flatten_error_messages(self, response): messages = [] if response.get('error'): if isinstance(response['error'], list): for e in response['errors']: messages.append(e['error']['value'][0]) else: messages.append(['error']) elif response.get('errors'): if isinstance(response['errors'], dict): for where, errors in response['errors'].items(): for e in errors: for type_e, msgs in e.items(): for m in msgs: messages.append('Error in {}: {}'.format(where, m)) return messages def _check_response(self, response): if response.status_code >= 500: response.raise_for_status() try: to_return = response.json() except: if self.debug: print(response.text) raise PyMISPError('Unknown error: {}'.format(response.text)) errors = [] if isinstance(to_return, list): to_return = {'response': to_return} if to_return.get('error'): if not isinstance(to_return['error'], list): errors.append(to_return['error']) else: errors += to_return['error'] if 400 <= response.status_code < 500: if to_return.get('error') is None and to_return.get('message'): errors.append(to_return['message']) else: errors.append(basestring(response.status_code)) errors += self.flatten_error_messages(to_return) if errors: to_return['errors'] = errors if self.debug: print(json.dumps(to_return, indent=4)) return to_return # ################################################ # ############### Simple REST API ################ # ################################################ def get_index(self, force_out=None, filters=None): """ Return the index. Warning, there's a limit on the number of results """ session = self.__prepare_session(force_out) url = urljoin(self.root_url, 'events/index') if filters is not None: filters = json.dumps(filters) print(filters) return session.post(url, data=filters) else: return session.get(url) def get_event(self, event_id, force_out=None): """ Get an event :param event_id: Event id to get """ session = self.__prepare_session(force_out) url = urljoin(self.root_url, 'events/{}'.format(event_id)) return session.get(url) def add_event(self, event, force_out=None): """ Add a new event :param event: Event as JSON object / string or XML to add """ session = self.__prepare_session(force_out) url = urljoin(self.root_url, 'events') if self.out_type == 'json': if isinstance(event, basestring): return session.post(url, data=event) else: return session.post(url, data=json.dumps(event)) else: return session.post(url, data=event) def update_event(self, event_id, event, force_out=None): """ Update an event :param event_id: Event id to update :param event: Event as JSON object / string or XML to add """ session = self.__prepare_session(force_out) url = urljoin(self.root_url, 'events/{}'.format(event_id)) if self.out_type == 'json': if isinstance(event, basestring): return session.post(url, data=event) else: return session.post(url, data=json.dumps(event)) else: return session.post(url, data=event) def delete_event(self, event_id, force_out=None): """ Delete an event :param event_id: Event id to delete """ session = self.__prepare_session(force_out) url = urljoin(self.root_url, 'events/{}'.format(event_id)) return session.delete(url) def delete_attribute(self, attribute_id, force_out=None): session = self.__prepare_session(force_out) url = urljoin(self.root_url, 'attributes/{}'.format(attribute_id)) return session.delete(url) # ############################################## # ######### Event handling (Json only) ######### # ############################################## def _prepare_full_event(self, distribution, threat_level_id, analysis, info, date=None, published=False): to_return = {'Event': {}} # Setup details of a new event if distribution not in [0, 1, 2, 3]: raise NewEventError('{} is invalid, the distribution has to be in 0, 1, 2, 3'.format(distribution)) if threat_level_id not in [1, 2, 3, 4]: raise NewEventError('{} is invalid, the threat_level_id has to be in 1, 2, 3, 4'.format(threat_level_id)) if analysis not in [0, 1, 2]: raise NewEventError('{} is invalid, the analysis has to be in 0, 1, 2'.format(analysis)) if date is None: date = datetime.date.today().isoformat() if published not in [True, False]: raise NewEventError('{} is invalid, published has to be True or False'.format(published)) to_return['Event'] = {'distribution': distribution, 'info': info, 'date': date, 'published': published, 'threat_level_id': threat_level_id, 'analysis': analysis} return to_return def _prepare_full_attribute(self, category, type_value, value, to_ids, comment=None, distribution=None): to_return = {} if category not in self.categories: raise NewAttributeError('{} is invalid, category has to be in {}'.format(category, (', '.join(self.categories)))) to_return['category'] = category if type_value not in self.types: raise NewAttributeError('{} is invalid, type_value has to be in {}'.format(type_value, (', '.join(self.types)))) to_return['type'] = type_value if to_ids not in [True, False]: raise NewAttributeError('{} is invalid, to_ids has to be True or False'.format(to_ids)) to_return['to_ids'] = to_ids if distribution is not None: distribution = int(distribution) # If None: take the default value of the event if distribution not in [None, 0, 1, 2, 3, 5]: raise NewAttributeError('{} is invalid, the distribution has to be in 0, 1, 2, 3, 5 or None'.format(distribution)) if distribution is not None: to_return['distribution'] = distribution to_return['value'] = value if comment is not None: to_return['comment'] = comment return to_return def _prepare_update(self, event): # Cleanup the received event to make it publishable event['Event'].pop('locked', None) event['Event'].pop('attribute_count', None) event['Event'].pop('RelatedEvent', None) event['Event'].pop('orgc', None) event['Event'].pop('ShadowAttribute', None) event['Event'].pop('org', None) event['Event'].pop('proposal_email_lock', None) event['Event'].pop('publish_timestamp', None) event['Event'].pop('published', None) event['Event'].pop('timestamp', None) event['Event']['id'] = int(event['Event']['id']) return event # ########## Helpers ########## def get(self, eid): response = self.get_event(int(eid), 'json') return self._check_response(response) def update(self, event): eid = event['Event']['id'] response = self.update_event(eid, event, 'json') return self._check_response(response) def new_event(self, distribution=None, threat_level_id=None, analysis=None, info=None, date=None, published=False): data = self._prepare_full_event(distribution, threat_level_id, analysis, info, date, published) response = self.add_event(data, 'json') return self._check_response(response) def publish(self, event): if event['Event']['published']: return {'error': 'Already published'} event = self._prepare_update(event) event['Event']['published'] = True response = self.update_event(event['Event']['id'], event, 'json') return self._check_response(response) def add_tag(self, event, tag): session = self.__prepare_session('json') to_post = {'request': {'Event': {'id': event['Event']['id'], 'tag': tag}}} response = session.post(urljoin(self.root_url, 'events/addTag'), data=json.dumps(to_post)) return self._check_response(response) def change_threat_level(self, event, threat_level_id): event['Event']['threat_level_id'] = threat_level_id self._prepare_update(event) response = self.update_event(event['Event']['id'], event) return self._check_response(response) # ##### File attributes ##### def _send_attributes(self, event, attributes, proposal=False): if proposal: response = self.proposal_add(event['Event']['id'], attributes) else: event = self._prepare_update(event) for a in attributes: if a.get('distribution') is None: a['distribution'] = event['Event']['distribution'] event['Event']['Attribute'] = attributes response = self.update_event(event['Event']['id'], event, 'json') return self._check_response(response) def add_hashes(self, event, category='Artifacts dropped', filename=None, md5=None, sha1=None, sha256=None, comment=None, to_ids=True, distribution=None, proposal=False): categories = ['Payload delivery', 'Artifacts dropped', 'Payload installation', 'External analysis'] if category not in categories: raise NewAttributeError('{} is invalid, category has to be in {}'.format(category, (', '.join(categories)))) attributes = [] type_value = '{}' value = '{}' if filename: type_value = 'filename|{}' value = filename + '|{}' if md5: attributes.append(self._prepare_full_attribute(category, type_value.format('md5'), value.format(md5), to_ids, comment, distribution)) if sha1: attributes.append(self._prepare_full_attribute(category, type_value.format('sha1'), value.format(sha1), to_ids, comment, distribution)) if sha256: attributes.append(self._prepare_full_attribute(category, type_value.format('sha256'), value.format(sha256), to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_regkey(self, event, regkey, rvalue=None, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False): type_value = '{}' value = '{}' if rvalue: type_value = 'regkey|value' value = '{}|{}'.format(regkey, rvalue) else: type_value = 'regkey' value = regkey attributes = [] attributes.append(self._prepare_full_attribute(category, type_value, value, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_pattern(self, event, pattern, in_file=True, in_memory=False, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] if in_file: attributes.append(self._prepare_full_attribute(category, 'pattern-in-file', pattern, to_ids, comment, distribution)) if in_memory: attributes.append(self._prepare_full_attribute(category, 'pattern-in-memory', pattern, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_pipe(self, event, named_pipe, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] if not named_pipe.startswith('\\.\\pipe\\'): named_pipe = '\\.\\pipe\\{}'.format(named_pipe) attributes.append(self._prepare_full_attribute(category, 'named pipe', named_pipe, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_mutex(self, event, mutex, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] if not mutex.startswith('\\BaseNamedObjects\\'): mutex = '\\BaseNamedObjects\\{}'.format(mutex) attributes.append(self._prepare_full_attribute(category, 'mutex', mutex, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) # ##### Network attributes ##### def add_ipdst(self, event, ipdst, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'ip-dst', ipdst, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_ipsrc(self, event, ipsrc, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'ip-src', ipsrc, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_hostname(self, event, hostname, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'hostname', hostname, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_domain(self, event, domain, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'domain', domain, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_url(self, event, url, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'url', url, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_useragent(self, event, useragent, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'user-agent', useragent, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_traffic_pattern(self, event, pattern, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'pattern-in-traffic', pattern, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_snort(self, event, snort, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute(category, 'snort', snort, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) # ##### Email attributes ##### def add_email_src(self, event, email, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Payload delivery', 'email-src', email, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_email_dst(self, event, email, category='Payload delivery', to_ids=True, comment=None, distribution=None, proposal=False): categories = ['Payload delivery', 'Network activity'] if category not in categories: raise NewAttributeError('{} is invalid, category has to be in {}'.format(category, (', '.join(categories)))) attributes = [] attributes.append(self._prepare_full_attribute(category, 'email-dst', email, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_email_subject(self, event, email, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Payload delivery', 'email-subject', email, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_email_attachment(self, event, email, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Payload delivery', 'email-attachment', email, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) # ##### Target attributes ##### def add_target_email(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Targeting data', 'target-email', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_target_user(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Targeting data', 'target-user', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_target_machine(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Targeting data', 'target-machine', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_target_org(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Targeting data', 'target-org', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_target_location(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Targeting data', 'target-location', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_target_external(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Targeting data', 'target-external', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) # ##### Attribution attributes ##### def add_threat_actor(self, event, target, to_ids=True, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Attribution', 'threat-actor', target, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) # ##### Internal reference attributes ##### def add_internal_link(self, event, reference, to_ids=False, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Internal reference', 'link', reference, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_internal_comment(self, event, reference, to_ids=False, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Internal reference', 'comment', reference, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_internal_text(self, event, reference, to_ids=False, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Internal reference', 'text', reference, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) def add_internal_other(self, event, reference, to_ids=False, comment=None, distribution=None, proposal=False): attributes = [] attributes.append(self._prepare_full_attribute('Internal reference', 'other', reference, to_ids, comment, distribution)) return self._send_attributes(event, attributes, proposal) # ################################################## # ######### Upload samples through the API ######### # ################################################## def _create_event(self, distribution, threat_level_id, analysis, info): # Setup details of a new event if distribution not in [0, 1, 2, 3]: raise NewEventError('{} is invalid, the distribution has to be in 0, 1, 2, 3'.format(distribution)) if threat_level_id not in [0, 1, 2, 3]: raise NewEventError('{} is invalid, the threat_level_id has to be in 0, 1, 2, 3'.format(threat_level_id)) if analysis not in [0, 1, 2]: raise NewEventError('{} is invalid, the analysis has to be in 0, 1, 2'.format(analysis)) return {'distribution': int(distribution), 'info': info, 'threat_level_id': int(threat_level_id), 'analysis': analysis} def prepare_attribute(self, event_id, distribution, to_ids, category, info, analysis, threat_level_id): to_post = {'request': {}} authorized_categs = ['Payload delivery', 'Artifacts dropped', 'Payload installation', 'External analysis', 'Network activity', 'Antivirus detection'] if event_id is not None: try: event_id = int(event_id) except: pass if not isinstance(event_id, int): # New event to_post['request'] = self._create_event(distribution, threat_level_id, analysis, info) else: to_post['request']['event_id'] = int(event_id) if to_ids not in [True, False]: raise NewAttributeError('{} is invalid, to_ids has to be True or False'.format(to_ids)) to_post['request']['to_ids'] = to_ids if category not in authorized_categs: raise NewAttributeError('{} is invalid, category has to be in {}'.format(category, (', '.join(authorized_categs)))) to_post['request']['category'] = category return to_post def _encode_file_to_upload(self, path): with open(path, 'rb') as f: return base64.b64encode(f.read()) def upload_sample(self, filename, filepath, event_id, distribution, to_ids, category, info, analysis, threat_level_id): to_post = self.prepare_attribute(event_id, distribution, to_ids, category, info, analysis, threat_level_id) to_post['request']['files'] = [{'filename': filename, 'data': self._encode_file_to_upload(filepath)}] return self._upload_sample(to_post) def upload_samplelist(self, filepaths, event_id, distribution, to_ids, category, info, analysis, threat_level_id): to_post = self.prepare_attribute(event_id, distribution, to_ids, category, info, analysis, threat_level_id) files = [] for path in filepaths: if not os.path.isfile(path): continue files.append({'filename': os.path.basename(path), 'data': self._encode_file_to_upload(path)}) to_post['request']['files'] = files return self._upload_sample(to_post) def _upload_sample(self, to_post): session = self.__prepare_session('json') url = urljoin(self.root_url, 'events/upload_sample') response = session.post(url, data=json.dumps(to_post)) return self._check_response(response) # ############################ # ######## Proposals ######### # ############################ def __query_proposal(self, session, path, id, attribute=None): path = path.strip('/') url = urljoin(self.root_url, 'shadow_attributes/{}/{}'.format(path, id)) query = None if path in ['add', 'edit']: query = {'request': {'ShadowAttribute': attribute}} if path == 'view': response = session.get(url) else: if query is not None: response = session.post(url, data=json.dumps(query)) else: response = session.post(url) return self._check_response(response) def proposal_view(self, event_id=None, proposal_id=None): session = self.__prepare_session('json') if proposal_id is not None and event_id is not None: return {'error': 'You can only view an event ID or a proposal ID'} if event_id is not None: id = event_id else: id = proposal_id return self.__query_proposal(session, 'view', id) def proposal_add(self, event_id, attribute): session = self.__prepare_session('json') return self.__query_proposal(session, 'add', event_id, attribute) def proposal_edit(self, attribute_id, attribute): session = self.__prepare_session('json') return self.__query_proposal(session, 'edit', attribute_id, attribute) def proposal_accept(self, proposal_id): session = self.__prepare_session('json') return self.__query_proposal(session, 'accept', proposal_id) def proposal_discard(self, proposal_id): session = self.__prepare_session('json') return self.__query_proposal(session, 'discard', proposal_id) # ############################## # ######## REST Search ######### # ############################## def __query(self, session, path, query): if query.get('error') is not None: return query url = urljoin(self.root_url, 'events/{}'.format(path.lstrip('/'))) query = {'request': query} response = session.post(url, data=json.dumps(query)) return self._check_response(response) def search_index(self, published=None, eventid=None, tag=None, datefrom=None, dateto=None, eventinfo=None, threatlevel=None, distribution=None, analysis=None, attribute=None, org=None): """ Search only at the index level. Use ! infront of value as NOT, default OR :param published: Published (0,1) :param eventid: Evend ID(s) | str or list :param tag: Tag(s) | str or list :param datefrom: First date, in format YYYY-MM-DD :param datefrom: Last date, in format YYYY-MM-DD :param eventinfo: Event info(s) to match | str or list :param threatlevel: Threat level(s) (1,2,3,4) | str or list :param distribution: Distribution level(s) (0,1,2,3) | str or list :param analysis: Analysis level(s) (0,1,2) | str or list :param org: Organisation(s) | str or list """ allowed = {'published': published, 'eventid': eventid, 'tag': tag, 'Dateto': dateto, 'Datefrom': datefrom, 'eventinfo': eventinfo, 'threatlevel': threatlevel, 'distribution': distribution, 'analysis': analysis, 'attribute': attribute, 'org': org} rule_levels = {'distribution': ["0", "1", "2", "3", "!0", "!1", "!2", "!3"], 'threatlevel': ["1", "2", "3", "4", "!1", "!2", "!3", "!4"], 'analysis': ["0", "1", "2", "!0", "!1", "!2"]} buildup_url = "events/index" for rule in allowed.keys(): if allowed[rule] is not None: if not isinstance(allowed[rule], list): allowed[rule] = [allowed[rule]] allowed[rule] = map(str, allowed[rule]) if rule in rule_levels: if not set(allowed[rule]).issubset(rule_levels[rule]): raise SearchError('Values in your {} are invalid, has to be in {}'.format(rule, ', '.join(str(x) for x in rule_levels[rule]))) if type(allowed[rule]) == list: joined = '|'.join(str(x) for x in allowed[rule]) buildup_url += '/search{}:{}'.format(rule, joined) else: buildup_url += '/search{}:{}'.format(rule, allowed[rule]) session = self.__prepare_session('json') url = urljoin(self.root_url, buildup_url) response = session.get(url) return self._check_response(response) def search_all(self, value): query = {'value': value, 'searchall': 1} session = self.__prepare_session('json') return self.__query(session, 'restSearch/download', query) def __prepare_rest_search(self, values, not_values): """ Prepare a search, generate the chain processed by the server :param values: Values to search :param not_values: Values that should not be in the response """ to_return = '' if values is not None: if not isinstance(values, list): to_return += values else: to_return += '&&'.join(values) if not_values is not None: if len(to_return) > 0: to_return += '&&!' else: to_return += '!' if not isinstance(values, list): to_return += not_values else: to_return += '&&!'.join(not_values) return to_return def search(self, values=None, not_values=None, type_attribute=None, category=None, org=None, tags=None, not_tags=None, date_from=None, date_to=None, last=None): """ Search via the Rest API :param values: values to search for :param not_values: values *not* to search for :param type_attribute: Type of attribute :param category: Category to search :param org: Org reporting the event :param tags: Tags to search for :param not_tags: Tags *not* to search for :param date_from: First date :param date_to: Last date :param last: Last updated events (for example 5d or 12h or 30m) """ val = self.__prepare_rest_search(values, not_values).replace('/', '|') tag = self.__prepare_rest_search(tags, not_tags).replace(':', ';') query = {} if len(val) != 0: query['value'] = val if len(tag) != 0: query['tags'] = tag if type_attribute is not None: query['type'] = type_attribute if category is not None: query['category'] = category if org is not None: query['org'] = org if date_from is not None: if isinstance(date_from, datetime.date) or isinstance(date_to, datetime.datetime): query['from'] = date_from.strftime('%Y-%m-%d') else: query['from'] = date_from if date_to is not None: if isinstance(date_to, datetime.date) or isinstance(date_to, datetime.datetime): query['to'] = date_to.strftime('%Y-%m-%d') else: query['to'] = date_to if last is not None: query['last'] = last session = self.__prepare_session('json') return self.__query(session, 'restSearch/download', query) def get_attachement(self, event_id): """ Get attachement of an event (not sample) :param event_id: Event id from where the attachements will be fetched """ attach = urljoin(self.root_url, 'attributes/downloadAttachment/download/{}'.format(event_id)) session = self.__prepare_session('json') return session.get(attach) def get_yara(self, event_id): to_post = {'request': {'eventid': event_id, 'type': 'yara'}} session = self.__prepare_session('json') response = session.post(urljoin(self.root_url, 'attributes/restSearch'), data=json.dumps(to_post)) result = self._check_response(response) if result.get('error') is not None: return False, result.get('error') if not result.get('response'): return False, result.get('message') rules = '\n\n'.join([a['value'] for a in result['response']['Attribute']]) return True, rules def download_samples(self, sample_hash=None, event_id=None, all_samples=False): to_post = {'request': {'hash': sample_hash, 'eventID': event_id, 'allSamples': all_samples}} session = self.__prepare_session('json') response = session.post(urljoin(self.root_url, 'attributes/downloadSample'), data=json.dumps(to_post)) result = self._check_response(response) if result.get('error') is not None: return False, result.get('error') if not result.get('result'): return False, result.get('message') details = [] for f in result['result']: decoded = base64.b64decode(f['base64']) zipped = BytesIO(decoded) try: archive = zipfile.ZipFile(zipped) try: # New format unzipped = BytesIO(archive.open(f['md5'], pwd='infected').read()) except KeyError: # Old format unzipped = BytesIO(archive.open(f['filename'], pwd='infected').read()) details.append([f['event_id'], f['filename'], unzipped]) except zipfile.BadZipfile: # In case the sample isn't zipped details.append([f['event_id'], f['filename'], zipped]) return True, details def download_last(self, last): """ Download the last updated events. :param last: can be defined in days, hours, minutes (for example 5d or 12h or 30m) """ return self.search(last=last) # ############## Suricata ############### def download_all_suricata(self): """ Download all suricata rules events. """ suricata_rules = urljoin(self.root_url, 'events/nids/suricata/download') session = self.__prepare_session('rules') return session.get(suricata_rules) def download_suricata_rule_event(self, event_id): """ Download one suricata rule event. :param event_id: ID of the event to download (same as get) """ template = urljoin(self.root_url, 'events/nids/suricata/download/{}'.format(event_id)) session = self.__prepare_session('rules') return session.get(template) # ########## Tags ########## def get_all_tags(self, quiet=False): session = self.__prepare_session('json') url = urljoin(self.root_url, 'tags') response = session.get(url) r = self._check_response(response) if not quiet or r.get('errors'): return r else: to_return = [] for tag in r['Tag']: to_return.append(tag['name']) return to_return def new_tag(self, name=None, colour="#00ace6", exportable=False): to_post = {'Tag': {'name': name, 'colour': colour, 'exportable': exportable}} session = self.__prepare_session('json') url = urljoin(self.root_url, 'tags/add') response = session.post(url, data=json.dumps(to_post)) return self._check_response(response) # ########## Version ########## def get_api_version(self): """ Returns the current version of PyMISP installed on the system """ return {'version': __version__} def get_api_version_master(self): """ Get the most recent version of PyMISP from github """ r = requests.get('https://raw.githubusercontent.com/MISP/PyMISP/master/pymisp/__init__.py') if r.status_code == 200: version = re.findall("__version__ = '(.*)'", r.text) return {'version': version[0]} else: return {'error': 'Impossible to retrieve the version of the master branch.'} def get_version(self): """ Returns the version of the instance. """ session = self.__prepare_session('json') url = urljoin(self.root_url, 'servers/getVersion') response = session.get(url) return self._check_response(response) def get_version_master(self): """ Get the most recent version from github """ r = requests.get('https://raw.githubusercontent.com/MISP/MISP/master/VERSION.json') if r.status_code == 200: master_version = json.loads(r.text) return {'version': '{}.{}.{}'.format(master_version['major'], master_version['minor'], master_version['hotfix'])} else: return {'error': 'Impossible to retrieve the version of the master branch.'} # ############## Export Attributes in text #################################### def get_all_attributes_txt(self, type_attr): session = self.__prepare_session('txt') url = urljoin(self.root_url, 'attributes/text/download/%s' % type_attr) response = session.get(url) return response # ############## Deprecated (Pure XML API should not be used) ################## @deprecated def download_all(self): """ Download all event from the instance """ xml = urljoin(self.root_url, 'events/xml/download') session = self.__prepare_session('xml') return session.get(xml) @deprecated def download(self, event_id, with_attachement=False): """ Download one event in XML :param event_id: Event id of the event to download (same as get) """ if with_attachement: attach = 'true' else: attach = 'false' template = urljoin(self.root_url, 'events/xml/download/{}/{}'.format(event_id, attach)) session = self.__prepare_session('xml') return session.get(template)