From 45684cfd5f703957dd86057d9185b9abe36b0c17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rapha=C3=ABl=20Vinot?= Date: Sat, 21 Oct 2017 20:22:38 +0200 Subject: [PATCH] chg: Update comments --- pymisp/api.py | 106 +++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 97 insertions(+), 9 deletions(-) diff --git a/pymisp/api.py b/pymisp/api.py index 00e8b56..83105c6 100644 --- a/pymisp/api.py +++ b/pymisp/api.py @@ -147,7 +147,7 @@ class PyMISP(object): self.sane_default = self.describe_types['sane_defaults'] def __prepare_session(self, output='json', async_implemented=False): - """Prepare the headers of the session""" + """Prepare the session headers""" if not HAVE_REQUESTS: raise MissingDependency('Missing dependency, install requests (`pip install requests`)') @@ -170,6 +170,10 @@ class PyMISP(object): # ##################### def flatten_error_messages(self, response): + """Dirty dirty method to normalize the error messages between the API calls. + Any response containing the a key 'error' or 'errors' failed at some point, + we make one single list out of it. + """ messages = [] if response.get('error'): if isinstance(response['error'], list): @@ -204,6 +208,7 @@ class PyMISP(object): return messages def _check_response(self, response): + """Check if the response from the server is not an unexpected error""" if response.status_code >= 500: response.raise_for_status() try: @@ -243,6 +248,7 @@ class PyMISP(object): return value if isinstance(value, (tuple, list)) else (value,) def _make_mispevent(self, event): + """Transform a Json MISP event into a MISPEvent""" if not isinstance(event, MISPEvent): e = MISPEvent(self.describe_types) e.load(event) @@ -251,6 +257,7 @@ class PyMISP(object): return e def _prepare_full_event(self, distribution, threat_level_id, analysis, info, date=None, published=False, orgc_id=None, org_id=None, sharing_group_id=None): + """Initialize a new MISPEvent from scratch""" misp_event = MISPEvent(self.describe_types) misp_event.set_all_values(info=info, distribution=distribution, threat_level_id=threat_level_id, analysis=analysis, date=date, orgc_id=orgc_id, org_id=org_id, sharing_group_id=sharing_group_id) @@ -259,6 +266,7 @@ class PyMISP(object): return misp_event def _prepare_full_attribute(self, category, type_value, value, to_ids, comment=None, distribution=5, **kwargs): + """Initialize a new MISPAttribute from scratch""" misp_attribute = MISPAttribute(self.describe_types) misp_attribute.set_all_values(type=type_value, value=value, category=category, to_ids=to_ids, comment=comment, distribution=distribution, **kwargs) @@ -353,12 +361,14 @@ class PyMISP(object): return self._check_response(response) def delete_attribute(self, attribute_id): + """Delete an attribute by ID""" session = self.__prepare_session() url = urljoin(self.root_url, 'attributes/{}'.format(attribute_id)) response = session.delete(url) return self._check_response(response) def pushEventToZMQ(self, event_id): + """Force push an event on ZMQ""" session = self.__prepare_session() url = urljoin(self.root_url, 'events/pushEventToZMQ/{}.json'.format(event_id)) response = session.post(url) @@ -369,9 +379,11 @@ class PyMISP(object): # ############################################## def get(self, eid): + """Get an event by event ID""" return self.get_event(eid) def update(self, event): + """Update an event by ID""" e = self._make_mispevent(event) if e.uuid: eid = e.uuid @@ -401,21 +413,25 @@ class PyMISP(object): return self._check_response(response) def change_threat_level(self, event, threat_level_id): + """Change the threat level of an event""" e = self._make_mispevent(event) e.threat_level_id = threat_level_id return self.update(e) def change_sharing_group(self, event, sharing_group_id): + """Change the sharing group of an event""" e = self._make_mispevent(event) e.distribution = 4 # Needs to be 'Sharing group' e.sharing_group_id = sharing_group_id return self.update(e) def new_event(self, distribution=None, threat_level_id=None, analysis=None, info=None, date=None, published=False, orgc_id=None, org_id=None, sharing_group_id=None): + """Create and add a new event""" misp_event = self._prepare_full_event(distribution, threat_level_id, analysis, info, date, published, orgc_id, org_id, sharing_group_id) return self.add_event(misp_event) def tag(self, uuid, tag): + """Tag an event or an attribute""" if not self._valid_uuid(uuid): raise PyMISPError('Invalid UUID') session = self.__prepare_session() @@ -425,6 +441,7 @@ class PyMISP(object): return self._check_response(response) def untag(self, uuid, tag): + """Untag an event or an attribute""" if not self._valid_uuid(uuid): raise PyMISPError('Invalid UUID') session = self.__prepare_session() @@ -436,6 +453,7 @@ class PyMISP(object): # ##### File attributes ##### def _send_attributes(self, event, attributes, proposal=False): + """Helper to add new attributes to an existing events""" eventID_to_update = None if isinstance(event, MISPEvent): if hasattr(event, 'id'): @@ -465,12 +483,14 @@ class PyMISP(object): return response def add_named_attribute(self, event, type_value, value, category=None, to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add one or more attributes to an existing event""" attributes = [] for value in self._one_or_more(value): attributes.append(self._prepare_full_attribute(category, type_value, value, to_ids, comment, distribution, **kwargs)) return self._send_attributes(event, attributes, proposal) def add_hashes(self, event, category='Artifacts dropped', filename=None, md5=None, sha1=None, sha256=None, ssdeep=None, comment=None, to_ids=True, distribution=None, proposal=False, **kwargs): + """Add hashe(s) to an existing event""" attributes = [] type_value = '{}' @@ -490,12 +510,15 @@ class PyMISP(object): return self._send_attributes(event, attributes, proposal) def av_detection_link(self, event, link, category='Antivirus detection', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add AV detection link(s)""" return self.add_named_attribute(event, 'link', link, category, to_ids, comment, distribution, proposal, **kwargs) def add_detection_name(self, event, name, category='Antivirus detection', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add AV detection name(s)""" return self.add_named_attribute(event, 'text', name, category, to_ids, comment, distribution, proposal, **kwargs) def add_filename(self, event, filename, category='Artifacts dropped', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add filename(s)""" return self.add_named_attribute(event, 'filename', filename, category, to_ids, comment, distribution, proposal, **kwargs) def add_attachment(self, event, attachment, category='Artifacts dropped', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): @@ -538,6 +561,7 @@ class PyMISP(object): return self.add_named_attribute(event, 'attachment', filename, category, to_ids, comment, distribution, proposal, data=encodedData, **kwargs) def add_regkey(self, event, regkey, rvalue=None, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add a registry key""" if rvalue: type_value = 'regkey|value' value = '{}|{}'.format(regkey, rvalue) @@ -550,6 +574,7 @@ class PyMISP(object): return self._send_attributes(event, attributes, proposal) def add_regkeys(self, event, regkeys_values, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add a registry keys""" attributes = [] for regkey, rvalue in regkeys_values.items(): @@ -564,12 +589,14 @@ class PyMISP(object): 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, **kwargs): + """Add a pattern(s) in file or in memory""" if not (in_file or in_memory): raise PyMISPError('Invalid pattern type: please use in_memory=True or in_file=True') itemtype = 'pattern-in-file' if in_file else 'pattern-in-memory' return self.add_named_attribute(event, itemtype, pattern, category, to_ids, comment, distribution, proposal, **kwargs) def add_pipe(self, event, named_pipe, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add pipes(s)""" def scrub(s): if not s.startswith('\\.\\pipe\\'): s = '\\.\\pipe\\{}'.format(s) @@ -578,6 +605,7 @@ class PyMISP(object): return self.add_named_attribute(event, 'named pipe', attributes, category, to_ids, comment, distribution, proposal, **kwargs) def add_mutex(self, event, mutex, category='Artifacts dropped', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add mutex(es)""" def scrub(s): if not s.startswith('\\BaseNamedObjects\\'): s = '\\BaseNamedObjects\\{}'.format(s) @@ -586,98 +614,125 @@ class PyMISP(object): return self.add_named_attribute(event, 'mutex', attributes, category, to_ids, comment, distribution, proposal, **kwargs) def add_yara(self, event, yara, category='Payload delivery', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add yara rule(es)""" return self.add_named_attribute(event, 'yara', yara, category, to_ids, comment, distribution, proposal, **kwargs) # ##### Network attributes ##### def add_ipdst(self, event, ipdst, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add destination IP(s)""" return self.add_named_attribute(event, 'ip-dst', ipdst, category, to_ids, comment, distribution, proposal, **kwargs) def add_ipsrc(self, event, ipsrc, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add source IP(s)""" return self.add_named_attribute(event, 'ip-src', ipsrc, category, to_ids, comment, distribution, proposal, **kwargs) def add_hostname(self, event, hostname, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add hostname(s)""" return self.add_named_attribute(event, 'hostname', hostname, category, to_ids, comment, distribution, proposal, **kwargs) def add_domain(self, event, domain, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add domain(s)""" return self.add_named_attribute(event, 'domain', domain, category, to_ids, comment, distribution, proposal, **kwargs) def add_domain_ip(self, event, domain, ip, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add domain|ip""" if isinstance(ip, str): ip = [ip] composed = list(map(lambda x: '%s|%s' % (domain, x), ip)) return self.add_named_attribute(event, 'domain|ip', composed, category, to_ids, comment, distribution, proposal, **kwargs) def add_domains_ips(self, event, domain_ips, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add multiple domain|ip""" composed = list(map(lambda x: '%s|%s' % (x[0], x[1]), domain_ips.items())) return self.add_named_attribute(event, 'domain|ip', composed, category, to_ids, comment, distribution, proposal, **kwargs) def add_url(self, event, url, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add url(s)""" return self.add_named_attribute(event, 'url', url, category, to_ids, comment, distribution, proposal, **kwargs) def add_useragent(self, event, useragent, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add user agent(s)""" return self.add_named_attribute(event, 'user-agent', useragent, category, to_ids, comment, distribution, proposal, **kwargs) def add_traffic_pattern(self, event, pattern, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add pattern(s) in traffic""" return self.add_named_attribute(event, 'pattern-in-traffic', pattern, category, to_ids, comment, distribution, proposal, **kwargs) def add_snort(self, event, snort, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add SNORT rule(s)""" return self.add_named_attribute(event, 'snort', snort, category, to_ids, comment, distribution, proposal, **kwargs) def add_net_other(self, event, netother, category='Network activity', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add a free text entry""" return self.add_named_attribute(event, 'other', netother, category, to_ids, comment, distribution, proposal, **kwargs) # ##### Email attributes ##### def add_email_src(self, event, email, category='Payload delivery', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add a source email""" return self.add_named_attribute(event, 'email-src', email, category, to_ids, comment, distribution, proposal, **kwargs) def add_email_dst(self, event, email, category='Payload delivery', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add a destination email""" return self.add_named_attribute(event, 'email-dst', email, category, to_ids, comment, distribution, proposal, **kwargs) def add_email_subject(self, event, email, category='Payload delivery', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an email subject""" return self.add_named_attribute(event, 'email-subject', email, category, to_ids, comment, distribution, proposal, **kwargs) def add_email_attachment(self, event, email, category='Payload delivery', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an email atachment""" return self.add_named_attribute(event, 'email-attachment', email, category, to_ids, comment, distribution, proposal, **kwargs) # ##### Target attributes ##### def add_target_email(self, event, target, category='Targeting data', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an target email""" return self.add_named_attribute(event, 'target-email', target, category, to_ids, comment, distribution, proposal, **kwargs) def add_target_user(self, event, target, category='Targeting data', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an target user""" return self.add_named_attribute(event, 'target-user', target, category, to_ids, comment, distribution, proposal, **kwargs) def add_target_machine(self, event, target, category='Targeting data', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an target machine""" return self.add_named_attribute(event, 'target-machine', target, category, to_ids, comment, distribution, proposal, **kwargs) def add_target_org(self, event, target, category='Targeting data', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an target organisation""" return self.add_named_attribute(event, 'target-org', target, category, to_ids, comment, distribution, proposal, **kwargs) def add_target_location(self, event, target, category='Targeting data', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an target location""" return self.add_named_attribute(event, 'target-location', target, category, to_ids, comment, distribution, proposal, **kwargs) def add_target_external(self, event, target, category='Targeting data', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an target external""" return self.add_named_attribute(event, 'target-external', target, category, to_ids, comment, distribution, proposal, **kwargs) # ##### Attribution attributes ##### def add_threat_actor(self, event, target, category='Attribution', to_ids=True, comment=None, distribution=None, proposal=False, **kwargs): + """Add an threat actor""" return self.add_named_attribute(event, 'threat-actor', target, category, to_ids, comment, distribution, proposal, **kwargs) # ##### Internal reference attributes ##### def add_internal_link(self, event, reference, category='Internal reference', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add an internal link""" return self.add_named_attribute(event, 'link', reference, category, to_ids, comment, distribution, proposal, **kwargs) def add_internal_comment(self, event, reference, category='Internal reference', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add an internal comment""" return self.add_named_attribute(event, 'comment', reference, category, to_ids, comment, distribution, proposal, **kwargs) def add_internal_text(self, event, reference, category='Internal reference', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add an internal text""" return self.add_named_attribute(event, 'text', reference, category, to_ids, comment, distribution, proposal, **kwargs) def add_internal_other(self, event, reference, category='Internal reference', to_ids=False, comment=None, distribution=None, proposal=False, **kwargs): + """Add an internal reference (type other)""" return self.add_named_attribute(event, 'other', reference, category, to_ids, comment, distribution, proposal, **kwargs) # ################################################## @@ -686,6 +741,7 @@ class PyMISP(object): def _prepare_upload(self, event_id, distribution, to_ids, category, comment, info, analysis, threat_level_id): + """Helper to prepare a sample to upload""" to_post = {'request': {}} if event_id is not None: @@ -716,6 +772,7 @@ class PyMISP(object): return to_post def _encode_file_to_upload(self, filepath_or_bytes): + """Helper to encode a file to upload""" if isinstance(filepath_or_bytes, basestring) and os.path.isfile(filepath_or_bytes): with open(filepath_or_bytes, 'rb') as f: binblob = f.read() @@ -726,6 +783,7 @@ class PyMISP(object): def upload_sample(self, filename, filepath_or_bytes, event_id, distribution=None, to_ids=True, category=None, comment=None, info=None, analysis=None, threat_level_id=None): + """Upload a sample""" to_post = self._prepare_upload(event_id, distribution, to_ids, category, comment, info, analysis, threat_level_id) to_post['request']['files'] = [{'filename': filename, 'data': self._encode_file_to_upload(filepath_or_bytes)}] @@ -734,6 +792,7 @@ class PyMISP(object): def upload_samplelist(self, filepaths, event_id, distribution=None, to_ids=True, category=None, comment=None, info=None, analysis=None, threat_level_id=None): + """Upload a list of samples""" to_post = self._prepare_upload(event_id, distribution, to_ids, category, comment, info, analysis, threat_level_id) files = [] @@ -745,6 +804,7 @@ class PyMISP(object): return self._upload_sample(to_post) def _upload_sample(self, to_post): + """Helper to upload a sample""" session = self.__prepare_session() url = urljoin(self.root_url, 'events/upload_sample') response = session.post(url, data=json.dumps(to_post)) @@ -755,6 +815,7 @@ class PyMISP(object): # ############################ def __query_proposal(self, session, path, id, attribute=None): + """Helper to prepare a query to handle proposals""" url = urljoin(self.root_url, 'shadow_attributes/{}/{}'.format(path, id)) if path in ['add', 'edit']: query = {'request': {'ShadowAttribute': attribute}} @@ -766,6 +827,7 @@ class PyMISP(object): return self._check_response(response) def proposal_view(self, event_id=None, proposal_id=None): + """View a proposal""" session = self.__prepare_session() 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'} @@ -776,18 +838,22 @@ class PyMISP(object): return self.__query_proposal(session, 'view', id) def proposal_add(self, event_id, attribute): + """Add a proposal""" session = self.__prepare_session() return self.__query_proposal(session, 'add', event_id, attribute) def proposal_edit(self, attribute_id, attribute): + """Edit a proposal""" session = self.__prepare_session() return self.__query_proposal(session, 'edit', attribute_id, attribute) def proposal_accept(self, proposal_id): + """Accept a proposal""" session = self.__prepare_session() return self.__query_proposal(session, 'accept', proposal_id) def proposal_discard(self, proposal_id): + """Discard a proposal""" session = self.__prepare_session() return self.__query_proposal(session, 'discard', proposal_id) @@ -796,6 +862,7 @@ class PyMISP(object): # ############################## def change_toids(self, attribute_uuid, to_ids): + """Change the toids flag""" if to_ids not in [0, 1]: raise Exception('to_ids can only be 0 or 1') query = {"to_ids": to_ids} @@ -807,6 +874,7 @@ class PyMISP(object): # ############################## def freetext(self, event_id, string, adhereToWarninglists=False, distribution=None): + """Pass a text to the freetext importer""" query = {"value": string} wl_params = [False, True, 'soft'] if adhereToWarninglists not in wl_params: @@ -823,6 +891,7 @@ class PyMISP(object): # ############################## def __query(self, session, path, query, controller='events', async_callback=None): + """Helper to prepare a search query""" if query.get('error') is not None: return query if controller not in ['events', 'attributes']: @@ -897,6 +966,7 @@ class PyMISP(object): return res def search_all(self, value): + """Search a value in the whole database""" query = {'value': value, 'searchall': 1} session = self.__prepare_session() return self.__query(session, 'restSearch/download', query) @@ -1030,6 +1100,7 @@ class PyMISP(object): return response.content def get_yara(self, event_id): + """Get the yara rules from an event""" to_post = {'request': {'eventid': event_id, 'type': 'yara'}} session = self.__prepare_session() response = session.post(urljoin(self.root_url, 'attributes/restSearch'), data=json.dumps(to_post)) @@ -1042,6 +1113,7 @@ class PyMISP(object): return True, rules def download_samples(self, sample_hash=None, event_id=None, all_samples=False): + """Download samples, by hash or event ID. If there are multiple samples in one event, use the all_samples switch""" to_post = {'request': {'hash': sample_hash, 'eventID': event_id, 'allSamples': all_samples}} session = self.__prepare_session() response = session.post(urljoin(self.root_url, 'attributes/downloadSample'), data=json.dumps(to_post)) @@ -1079,6 +1151,7 @@ class PyMISP(object): # ########## Tags ########## def get_all_tags(self, quiet=False): + """Get all the tags used on the instance""" session = self.__prepare_session() url = urljoin(self.root_url, 'tags') response = session.get(url) @@ -1092,6 +1165,7 @@ class PyMISP(object): return to_return def new_tag(self, name=None, colour="#00ace6", exportable=False): + """Create a new tag""" to_post = {'Tag': {'name': name, 'colour': colour, 'exportable': exportable}} session = self.__prepare_session() url = urljoin(self.root_url, 'tags/add') @@ -1168,18 +1242,21 @@ class PyMISP(object): # ############## Sightings ################## def sighting_per_id(self, attribute_id): + """Add a sighting to an attribute (by attribute ID)""" session = self.__prepare_session() url = urljoin(self.root_url, 'sightings/add/{}'.format(attribute_id)) response = session.post(url) return self._check_response(response) def sighting_per_uuid(self, attribute_uuid): + """Add a sighting to an attribute (by attribute UUID)""" session = self.__prepare_session() url = urljoin(self.root_url, 'sightings/add/{}'.format(attribute_uuid)) response = session.post(url) return self._check_response(response) def set_sightings(self, sightings): + """Push a sighting (python dictionary)""" if isinstance(sightings, dict): sightings = json.dumps(sightings) session = self.__prepare_session() @@ -1188,6 +1265,7 @@ class PyMISP(object): return self._check_response(response) def sighting_per_json(self, json_file): + """Push a sighting (JSON file)""" with open(json_file, 'r') as f: jdata = json.load(f) return self.set_sightings(jdata) @@ -1195,6 +1273,7 @@ class PyMISP(object): # ############## Sharing Groups ################## def get_sharing_groups(self): + """Get the existing sharing groups""" session = self.__prepare_session() url = urljoin(self.root_url, 'sharing_groups.json') response = session.get(url) @@ -1474,6 +1553,7 @@ class PyMISP(object): # ############## Roles ################## def get_roles_list(self): + """Get the list of existing roles""" session = self.__prepare_session() url = urljoin(self.root_url, '/roles') response = session.get(url) @@ -1482,6 +1562,7 @@ class PyMISP(object): # ############## Tags ################## def get_tags_list(self): + """Get the list of existing tags""" session = self.__prepare_session() url = urljoin(self.root_url, '/tags') response = session.get(url) @@ -1542,86 +1623,93 @@ class PyMISP(object): # ########################### def fetch_feed(self, feed_id): + """Fetch one single feed""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/fetchFromFeed/{}'.format(feed_id)) response = session.get(url) return self._check_response(response) def view_feeds(self): + """Get the content of all the feeds""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds') response = session.get(url) return self._check_response(response) def view_feed(self, feed_ids): + """Get the content of a single feed""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/view/{}'.format(feed_ids)) response = session.get(url) return self._check_response(response) def cache_feeds_all(self): + """ Cache all the feeds""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/cacheFeeds/all') response = session.get(url) return self._check_response(response) def cache_feed(self, feed_id): + """Cache a specific feed""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/cacheFeeds/{}'.format(feed_id)) response = session.get(url) return self._check_response(response) def cache_feeds_freetext(self): + """Cache all the freetext feeds""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/cacheFeeds/freetext') response = session.get(url) return self._check_response(response) def cache_feeds_misp(self): + """Cache all the MISP feeds""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/cacheFeeds/misp') response = session.get(url) return self._check_response(response) def compare_feeds(self): + """Generate the comparison matrix for all the MISP feeds""" session = self.__prepare_session() url = urljoin(self.root_url, 'feeds/compareFeeds') response = session.get(url) return self._check_response(response) - # ########################### - # ### Cache All Feeds ### - # ########################### - def cache_all_feeds(self): - session = self.__prepare_session() - url = urljoin(self.root_url, 'feeds/cacheFeeds/all') - response = session.post(url) - return self._check_response(response) + """Alias for cache_feeds_all""" + # DEPRECATED + return self.cache_feeds_all() # ################### # ### Objects ### # ################### def add_object(self, event_id, template_id, misp_object): + """Add an object""" session = self.__prepare_session() url = urljoin(self.root_url, 'objects/add/{}/{}'.format(event_id, template_id)) response = session.post(url, data=misp_object.to_json()) return self._check_response(response) def add_object_reference(self, misp_object_reference): + """Add a reference to an object""" session = self.__prepare_session() url = urljoin(self.root_url, 'object_references/add') response = session.post(url, data=misp_object_reference.to_json()) return self._check_response(response) def get_object_templates_list(self): + """Returns the list of Object templates available on the MISP instance""" session = self.__prepare_session() url = urljoin(self.root_url, 'objectTemplates') response = session.get(url) return self._check_response(response)['response'] def get_object_template_id(self, object_uuid): + """Gets the template ID corresponting the UUID passed as parameter""" templates = self.get_object_templates_list() for t in templates: if t['ObjectTemplate']['uuid'] == object_uuid: