#!/usr/bin/env python3 # -*- coding: utf-8 -*- import sys import unittest from pymisp.tools import make_binary_objects from datetime import datetime, timedelta, date from io import BytesIO import re import json from pathlib import Path import time from uuid import uuid4 import logging logging.disable(logging.CRITICAL) try: from pymisp import ExpandedPyMISP, MISPEvent, MISPOrganisation, MISPUser, Distribution, ThreatLevel, Analysis, MISPObject, MISPAttribute from pymisp.tools import CSVLoader, DomainIPObject, ASNObject except ImportError: if sys.version_info < (3, 6): print('This test suite requires Python 3.6+, breaking.') sys.exit(0) else: raise try: from keys import url, key verifycert = False travis_run = True except ImportError as e: print(e) url = 'http://localhost:8080' key = 'HRizIMmaxBOXAQSzKZ874rDWUsQEk4vGAGBoljQO' verifycert = False travis_run = False class TestComprehensive(unittest.TestCase): @classmethod def setUpClass(cls): cls.maxDiff = None # Connect as admin cls.admin_misp_connector = ExpandedPyMISP(url, key, verifycert, debug=False) # Creates an org organisation = MISPOrganisation() organisation.name = 'Test Org' cls.test_org = cls.admin_misp_connector.add_organisation(organisation) # Set the refault role (id 3 on the VM) cls.admin_misp_connector.set_default_role(3) # Creates a user user = MISPUser() user.email = 'testusr@user.local' user.org_id = cls.test_org.id cls.test_usr = cls.admin_misp_connector.add_user(user) cls.user_misp_connector = ExpandedPyMISP(url, cls.test_usr.authkey, verifycert, debug=False) # Creates a publisher user = MISPUser() user.email = 'testpub@user.local' user.org_id = cls.test_org.id user.role_id = 4 cls.test_pub = cls.admin_misp_connector.add_user(user) cls.pub_misp_connector = ExpandedPyMISP(url, cls.test_pub.authkey, verifycert) # Update all json stuff cls.admin_misp_connector.update_object_templates() cls.admin_misp_connector.update_galaxies() cls.admin_misp_connector.update_noticelists() cls.admin_misp_connector.update_warninglists() cls.admin_misp_connector.update_taxonomies() @classmethod def tearDownClass(cls): # Delete publisher cls.admin_misp_connector.delete_user(user_id=cls.test_pub.id) # Delete user cls.admin_misp_connector.delete_user(user_id=cls.test_usr.id) # Delete org cls.admin_misp_connector.delete_organisation(org_id=cls.test_org.id) def create_simple_event(self, force_timestamps=False): mispevent = MISPEvent(force_timestamps=force_timestamps) mispevent.info = 'This is a super simple test' mispevent.distribution = Distribution.your_organisation_only mispevent.threat_level_id = ThreatLevel.low mispevent.analysis = Analysis.completed mispevent.add_attribute('text', str(uuid4())) return mispevent def environment(self): first_event = MISPEvent() first_event.info = 'First event - org only - low - completed' first_event.distribution = Distribution.your_organisation_only first_event.threat_level_id = ThreatLevel.low first_event.analysis = Analysis.completed first_event.set_date("2017-12-31") first_event.add_attribute('text', str(uuid4())) first_event.attributes[0].add_tag('admin_only') first_event.attributes[0].add_tag('tlp:white___test') first_event.add_attribute('text', str(uuid4())) first_event.attributes[1].add_tag('unique___test') second_event = MISPEvent() second_event.info = 'Second event - org only - medium - ongoing' second_event.distribution = Distribution.your_organisation_only second_event.threat_level_id = ThreatLevel.medium second_event.analysis = Analysis.ongoing second_event.set_date("Aug 18 2018") second_event.add_attribute('text', str(uuid4())) second_event.attributes[0].add_tag('tlp:white___test') second_event.add_attribute('ip-dst', '1.1.1.1') second_event.attributes[1].add_tag('tlp:amber___test') # Same value as in first event. second_event.add_attribute('text', first_event.attributes[0].value) third_event = MISPEvent() third_event.info = 'Third event - all orgs - high - initial' third_event.distribution = Distribution.all_communities third_event.threat_level_id = ThreatLevel.high third_event.analysis = Analysis.initial third_event.set_date("Jun 25 2018") third_event.add_tag('tlp:white___test') third_event.add_attribute('text', str(uuid4())) third_event.attributes[0].add_tag('tlp:amber___test') third_event.attributes[0].add_tag('foo_double___test') third_event.add_attribute('ip-src', '8.8.8.8') third_event.attributes[1].add_tag('tlp:amber___test') third_event.add_attribute('ip-dst', '9.9.9.9') # Create first and third event as admin # usr won't be able to see the first one first = self.admin_misp_connector.add_event(first_event) third = self.admin_misp_connector.add_event(third_event) # Create second event as user second = self.user_misp_connector.add_event(second_event) return first, second, third def test_search_value_event(self): '''Search a value on the event controller * Test ACL admin user vs normal user in an other org * Make sure we have one match ''' try: first, second, third = self.environment() # Search as admin events = self.admin_misp_connector.search(value=first.attributes[0].value, pythonify=True) self.assertEqual(len(events), 2) for e in events: self.assertIn(e.id, [first.id, second.id]) # Search as user events = self.user_misp_connector.search(value=first.attributes[0].value, pythonify=True) self.assertEqual(len(events), 1) for e in events: self.assertIn(e.id, [second.id]) # Non-existing value events = self.user_misp_connector.search(value=str(uuid4()), pythonify=True) self.assertEqual(events, []) finally: # Delete events self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_value_attribute(self): '''Search value in attributes controller''' try: first, second, third = self.environment() # Search as admin attributes = self.admin_misp_connector.search(controller='attributes', value=first.attributes[0].value, pythonify=True) self.assertEqual(len(attributes), 2) for a in attributes: self.assertIn(a.event_id, [first.id, second.id]) # Search as user attributes = self.user_misp_connector.search(controller='attributes', value=first.attributes[0].value, pythonify=True) self.assertEqual(len(attributes), 1) for a in attributes: self.assertIn(a.event_id, [second.id]) # Non-existing value attributes = self.user_misp_connector.search(controller='attributes', value=str(uuid4()), pythonify=True) self.assertEqual(attributes, []) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_type_event(self): '''Search multiple events, search events containing attributes with specific types''' try: first, second, third = self.environment() # Search as admin events = self.admin_misp_connector.search(timestamp=first.timestamp.timestamp(), pythonify=True) self.assertEqual(len(events), 3) for e in events: self.assertIn(e.id, [first.id, second.id, third.id]) attributes_types_search = self.admin_misp_connector.build_complex_query(or_parameters=['ip-src', 'ip-dst']) events = self.admin_misp_connector.search(timestamp=first.timestamp.timestamp(), type_attribute=attributes_types_search, pythonify=True) self.assertEqual(len(events), 2) for e in events: self.assertIn(e.id, [second.id, third.id]) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_type_attribute(self): '''Search multiple attributes, search attributes with specific types''' try: first, second, third = self.environment() # Search as admin attributes = self.admin_misp_connector.search(controller='attributes', timestamp=first.timestamp.timestamp(), pythonify=True) self.assertEqual(len(attributes), 8) for a in attributes: self.assertIn(a.event_id, [first.id, second.id, third.id]) # Search as user attributes_types_search = self.admin_misp_connector.build_complex_query(or_parameters=['ip-src', 'ip-dst']) attributes = self.admin_misp_connector.search(controller='attributes', timestamp=first.timestamp.timestamp(), type_attribute=attributes_types_search, pythonify=True) self.assertEqual(len(attributes), 3) for a in attributes: self.assertIn(a.event_id, [second.id, third.id]) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_tag_event(self): '''Search Tags at events level''' try: first, second, third = self.environment() # Search as admin events = self.admin_misp_connector.search(tags='tlp:white___test', pythonify=True) self.assertEqual(len(events), 3) for e in events: self.assertIn(e.id, [first.id, second.id, third.id]) events = self.admin_misp_connector.search(tags='tlp:amber___test', pythonify=True) self.assertEqual(len(events), 2) for e in events: self.assertIn(e.id, [second.id, third.id]) events = self.admin_misp_connector.search(tags='admin_only', pythonify=True) self.assertEqual(len(events), 1) for e in events: self.assertIn(e.id, [first.id]) # Search as user events = self.user_misp_connector.search(tags='tlp:white___test', pythonify=True) self.assertEqual(len(events), 2) for e in events: self.assertIn(e.id, [second.id, third.id]) events = self.user_misp_connector.search(tags='tlp:amber___test', pythonify=True) self.assertEqual(len(events), 2) for e in events: self.assertIn(e.id, [second.id, third.id]) events = self.user_misp_connector.search(tags='admin_only', pythonify=True) self.assertEqual(events, []) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_tag_attribute(self): '''Search Tags at attributes level''' try: first, second, third = self.environment() # Search as admin attributes = self.admin_misp_connector.search(controller='attributes', tags='tlp:white___test', pythonify=True) self.assertEqual(len(attributes), 5) attributes = self.admin_misp_connector.search(controller='attributes', tags='tlp:amber___test', pythonify=True) self.assertEqual(len(attributes), 3) attributes = self.admin_misp_connector.search(tags='admin_only', pythonify=True) self.assertEqual(len(attributes), 1) # Search as user attributes = self.user_misp_connector.search(controller='attributes', tags='tlp:white___test', pythonify=True) self.assertEqual(len(attributes), 4) attributes = self.user_misp_connector.search(controller='attributes', tags='tlp:amber___test', pythonify=True) self.assertEqual(len(attributes), 3) attributes = self.user_misp_connector.search(tags='admin_only', pythonify=True) self.assertEqual(attributes, []) attributes_tags_search = self.admin_misp_connector.build_complex_query(or_parameters=['tlp:amber___test'], not_parameters=['tlp:white___test']) attributes = self.user_misp_connector.search(controller='attributes', tags=attributes_tags_search, pythonify=True) self.assertEqual(len(attributes), 1) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_tag_advanced_event(self): '''Advanced search Tags at events level''' try: first, second, third = self.environment() complex_query = self.admin_misp_connector.build_complex_query(or_parameters=['tlp:white___test'], not_parameters=['tlp:amber___test', 'foo_double___test']) events = self.admin_misp_connector.search(tags=complex_query, pythonify=True) self.assertEqual(len(events), 3) for e in events: self.assertIn(e.id, [first.id, second.id, third.id]) for a in e.attributes: self.assertEqual([t for t in a.tags if t.name == 'tlp:amber___test'], []) for a in e.attributes: self.assertEqual([t for t in a.tags if t.name == 'foo_double___test'], []) complex_query = self.admin_misp_connector.build_complex_query(or_parameters=['unique___test'], not_parameters=['tlp:white___test']) events = self.admin_misp_connector.search(tags=complex_query, pythonify=True) self.assertEqual(len(events), 1) for e in events: self.assertIn(e.id, [first.id, second.id]) for a in e.attributes: self.assertEqual([t for t in a.tags if t.name == 'tlp:white___test'], []) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_tag_advanced_attributes(self): '''Advanced search Tags at attributes level''' try: first, second, third = self.environment() complex_query = self.admin_misp_connector.build_complex_query(or_parameters=['tlp:white___test'], not_parameters=['tlp:amber___test', 'foo_double___test']) attributes = self.admin_misp_connector.search(controller='attributes', tags=complex_query, pythonify=True) self.assertEqual(len(attributes), 3) for a in attributes: self.assertEqual([t for t in a.tags if t.name == 'tlp:amber___test'], []) for a in attributes: self.assertEqual([t for t in a.tags if t.name == 'foo_double___test'], []) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_search_timestamp_event(self): '''Search specific update timestamps at events level''' # Creating event 1 - timestamp 5 min ago first = self.create_simple_event(force_timestamps=True) event_creation_timestamp_first = datetime.now() - timedelta(minutes=5) first.timestamp = event_creation_timestamp_first # Creating event 2 - timestamp 2 min ago second = self.create_simple_event(force_timestamps=True) event_creation_timestamp_second = datetime.now() - timedelta(minutes=2) second.timestamp = event_creation_timestamp_second try: first = self.user_misp_connector.add_event(first) second = self.user_misp_connector.add_event(second) # Search as user # # Test - last 4 min events = self.user_misp_connector.search(timestamp='4m', pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) self.assertEqual(events[0].timestamp.timestamp(), int(event_creation_timestamp_second.timestamp())) # # Test timestamp of 2nd event events = self.user_misp_connector.search(timestamp=event_creation_timestamp_second.timestamp(), pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) self.assertEqual(events[0].timestamp.timestamp(), int(event_creation_timestamp_second.timestamp())) # # Test interval -6 min -> -4 min events = self.user_misp_connector.search(timestamp=['6m', '4m'], pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) self.assertEqual(events[0].timestamp.timestamp(), int(event_creation_timestamp_first.timestamp())) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_search_timestamp_attribute(self): '''Search specific update timestamps at attributes level''' # Creating event 1 - timestamp 5 min ago first = self.create_simple_event(force_timestamps=True) event_creation_timestamp_first = datetime.now() - timedelta(minutes=5) first.timestamp = event_creation_timestamp_first first.attributes[0].timestamp = event_creation_timestamp_first # Creating event 2 - timestamp 2 min ago second = self.create_simple_event(force_timestamps=True) event_creation_timestamp_second = datetime.now() - timedelta(minutes=2) second.timestamp = event_creation_timestamp_second second.attributes[0].timestamp = event_creation_timestamp_second try: first = self.user_misp_connector.add_event(first) second = self.user_misp_connector.add_event(second) # Search as user # # Test - last 4 min attributes = self.user_misp_connector.search(controller='attributes', timestamp='4m', pythonify=True) self.assertEqual(len(attributes), 1) self.assertEqual(attributes[0].event_id, second.id) self.assertEqual(attributes[0].timestamp.timestamp(), int(event_creation_timestamp_second.timestamp())) # # Test timestamp of 2nd event attributes = self.user_misp_connector.search(controller='attributes', timestamp=event_creation_timestamp_second.timestamp(), pythonify=True) self.assertEqual(len(attributes), 1) self.assertEqual(attributes[0].event_id, second.id) self.assertEqual(attributes[0].timestamp.timestamp(), int(event_creation_timestamp_second.timestamp())) # # Test interval -6 min -> -4 min attributes = self.user_misp_connector.search(controller='attributes', timestamp=['6m', '4m'], pythonify=True) self.assertEqual(len(attributes), 1) self.assertEqual(attributes[0].event_id, first.id) self.assertEqual(attributes[0].timestamp.timestamp(), int(event_creation_timestamp_first.timestamp())) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_user_perms(self): '''Test publish rights''' try: first = self.create_simple_event() first.publish() # Add event as user, no publish rights first = self.user_misp_connector.add_event(first) self.assertFalse(first.published) # Add event as publisher first.publish() first = self.pub_misp_connector.update_event(first) self.assertTrue(first.published) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_search_publish_timestamp(self): '''Search for a specific publication timestamp, an interval, and invalid values.''' # Creating event 1 first = self.create_simple_event() first.publish() # Creating event 2 second = self.create_simple_event() second.publish() try: first = self.pub_misp_connector.add_event(first) time.sleep(10) second = self.pub_misp_connector.add_event(second) # Test invalid query events = self.pub_misp_connector.search(publish_timestamp='5x', pythonify=True) self.assertEqual(events, []) events = self.pub_misp_connector.search(publish_timestamp='ad', pythonify=True) self.assertEqual(events, []) events = self.pub_misp_connector.search(publish_timestamp='aaad', pythonify=True) self.assertEqual(events, []) # Test - last 4 min events = self.pub_misp_connector.search(publish_timestamp='5s', pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) # Test 5 sec before timestamp of 2nd event events = self.pub_misp_connector.search(publish_timestamp=(second.publish_timestamp.timestamp()), pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) # Test interval -6 min -> -4 min events = self.pub_misp_connector.search(publish_timestamp=[first.publish_timestamp.timestamp() - 5, second.publish_timestamp.timestamp() - 5], pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_default_distribution(self): '''The default distributions on the VM are This community only for the events and Inherit from event for attr/obj)''' if travis_run: return first = self.create_simple_event() del first.distribution o = first.add_object(name='file') o.add_attribute('filename', value='foo.exe') try: # Event create first = self.user_misp_connector.add_event(first) self.assertEqual(first.distribution, Distribution.this_community_only.value) self.assertEqual(first.attributes[0].distribution, Distribution.inherit.value) self.assertEqual(first.objects[0].distribution, Distribution.inherit.value) self.assertEqual(first.objects[0].attributes[0].distribution, Distribution.inherit.value) # Event edit first.add_attribute('ip-dst', '12.54.76.43') o = first.add_object(name='file') o.add_attribute('filename', value='foo2.exe') first = self.user_misp_connector.update_event(first) self.assertEqual(first.attributes[1].distribution, Distribution.inherit.value) self.assertEqual(first.objects[1].distribution, Distribution.inherit.value) self.assertEqual(first.objects[1].attributes[0].distribution, Distribution.inherit.value) # Attribute create attribute = self.user_misp_connector.add_named_attribute(first, 'comment', 'bar') # FIXME: Add helper that returns a list of MISPAttribute self.assertEqual(attribute[0]['Attribute']['distribution'], str(Distribution.inherit.value)) # Object - add o = MISPObject('file') o.add_attribute('filename', value='blah.exe') new_obj = self.user_misp_connector.add_object(first.id, o) # FIXME: Add helper that returns a MISPObject self.assertEqual(new_obj.distribution, int(Distribution.inherit.value)) self.assertEqual(new_obj.attributes[0].distribution, int(Distribution.inherit.value)) # Object - edit clean_obj = MISPObject(name=new_obj.name, strict=True) clean_obj.from_dict(**new_obj) clean_obj.add_attribute('filename', value='blah.exe') new_obj = self.user_misp_connector.update_object(clean_obj) for a in new_obj.attributes: self.assertEqual(a.distribution, int(Distribution.inherit.value)) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_simple_event(self): '''Search a bunch of parameters: * Value not existing * only return metadata * published yes/no * event id * uuid * creator org * substring search in value and eventinfo * quickfilter * date_from * date_to * deleted * to_ids * include_event_uuid warning list ''' first = self.create_simple_event() first.info = 'foo bar blah' # First has one text attribute second = self.create_simple_event() second.info = 'foo blah' second.set_date('2018-09-01') second.add_attribute('ip-src', '8.8.8.8') # second has two attributes: text and ip-src try: first = self.user_misp_connector.add_event(first) second = self.user_misp_connector.add_event(second) timeframe = [first.timestamp.timestamp() - 5, first.timestamp.timestamp() + 5] # Search event we just created in multiple ways. Make sure it doesn't catch it when it shouldn't events = self.user_misp_connector.search(timestamp=timeframe, pythonify=True) self.assertEqual(len(events), 2) self.assertEqual(events[0].id, first.id) self.assertEqual(events[1].id, second.id) events = self.user_misp_connector.search(timestamp=timeframe, value='nothere', pythonify=True) self.assertEqual(events, []) events = self.user_misp_connector.search(timestamp=timeframe, value=first.attributes[0].value, pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) events = self.user_misp_connector.search(timestamp=[first.timestamp.timestamp() - 50, first.timestamp.timestamp() - 10], value=first.attributes[0].value, pythonify=True) self.assertEqual(events, []) # Test return content events = self.user_misp_connector.search(timestamp=timeframe, metadata=False, pythonify=True) self.assertEqual(len(events), 2) self.assertEqual(len(events[0].attributes), 1) self.assertEqual(len(events[1].attributes), 2) events = self.user_misp_connector.search(timestamp=timeframe, metadata=True, pythonify=True) self.assertEqual(len(events), 2) self.assertEqual(len(events[0].attributes), 0) self.assertEqual(len(events[1].attributes), 0) # other things events = self.user_misp_connector.search(timestamp=timeframe, published=True, pythonify=True) self.assertEqual(events, []) events = self.user_misp_connector.search(timestamp=timeframe, published=False, pythonify=True) self.assertEqual(len(events), 2) events = self.user_misp_connector.search(eventid=first.id, pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) events = self.user_misp_connector.search(uuid=first.uuid, pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) events = self.user_misp_connector.search(org=first.orgc_id, pythonify=True) self.assertEqual(len(events), 2) # test like search events = self.user_misp_connector.search(timestamp=timeframe, value='%{}%'.format(first.attributes[0].value.split('-')[2]), pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) events = self.user_misp_connector.search(timestamp=timeframe, eventinfo='%bar blah%', pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) # quickfilter events = self.user_misp_connector.search(timestamp=timeframe, quickfilter='%foo blah%', pythonify=True) # FIXME: should return one event # print(events) # self.assertEqual(len(events), 1) # self.assertEqual(events[0].id, second.id) # date_from / date_to events = self.user_misp_connector.search(timestamp=timeframe, date_from=date.today().isoformat(), pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, first.id) events = self.user_misp_connector.search(timestamp=timeframe, date_from='2018-09-01', pythonify=True) self.assertEqual(len(events), 2) events = self.user_misp_connector.search(timestamp=timeframe, date_from='2018-09-01', date_to='2018-09-02', pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) # Category events = self.user_misp_connector.search(timestamp=timeframe, category='Network activity', pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) # toids events = self.user_misp_connector.search(timestamp=timeframe, to_ids='0', pythonify=True) self.assertEqual(len(events), 2) events = self.user_misp_connector.search(timestamp=timeframe, to_ids='1', pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) self.assertEqual(len(events[0].attributes), 1) # deleted second.attributes[1].delete() self.user_misp_connector.update_event(second) events = self.user_misp_connector.search(eventid=second.id, pythonify=True) self.assertEqual(len(events[0].attributes), 1) events = self.user_misp_connector.search(eventid=second.id, deleted=True, pythonify=True) self.assertEqual(len(events[0].attributes), 2) # include_event_uuid attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, include_event_uuid=True, pythonify=True) self.assertEqual(attributes[0].event_uuid, second.uuid) # event_timestamp time.sleep(1) second.add_attribute('ip-src', '8.8.8.9') second = self.user_misp_connector.update_event(second) events = self.user_misp_connector.search(event_timestamp=second.timestamp.timestamp(), pythonify=True) self.assertEqual(len(events), 1) # searchall second.add_attribute('text', 'This is a test for the full text search', comment='Test stuff comment') second = self.user_misp_connector.update_event(second) events = self.user_misp_connector.search(value='%for the full text%', searchall=True, pythonify=True) self.assertEqual(len(events), 1) # warninglist response = self.admin_misp_connector.toggle_warninglist(warninglist_name='%dns resolv%', force_enable=True) # enable ipv4 DNS. self.assertDictEqual(response, {'saved': True, 'success': '3 warninglist(s) enabled'}) second.add_attribute('ip-src', '1.11.71.4') second.add_attribute('ip-src', '9.9.9.9') second = self.user_misp_connector.update_event(second) events = self.user_misp_connector.search(eventid=second.id, pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) self.assertEqual(len(events[0].attributes), 5) events = self.user_misp_connector.search(eventid=second.id, enforce_warninglist=False, pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) self.assertEqual(len(events[0].attributes), 5) if not travis_run: # FIXME: This is failing on travis for no discernable reason... events = self.user_misp_connector.search(eventid=second.id, enforce_warninglist=True, pythonify=True) self.assertEqual(len(events), 1) self.assertEqual(events[0].id, second.id) self.assertEqual(len(events[0].attributes), 3) response = self.admin_misp_connector.toggle_warninglist(warninglist_name='%dns resolv%') # disable ipv4 DNS. self.assertDictEqual(response, {'saved': True, 'success': '3 warninglist(s) toggled'}) # Page / limit attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, page=1, limit=3, pythonify=True) self.assertEqual(len(attributes), 3) attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, page=2, limit=3, pythonify=True) self.assertEqual(len(attributes), 2) time.sleep(1) # make sure the next attribute is added one at least one second later # attachments with open('tests/testlive_comprehensive.py', 'rb') as f: first.add_attribute('malware-sample', value='testfile.py', data=BytesIO(f.read())) first = self.user_misp_connector.update_event(first) events = self.user_misp_connector.search(timestamp=first.timestamp.timestamp(), with_attachments=True, pythonify=True) self.assertEqual(len(events), 1) self.assertIs(type(events[0].attributes[-1].malware_binary), BytesIO) events = self.user_misp_connector.search(timestamp=first.timestamp.timestamp(), with_attachments=False, pythonify=True) self.assertEqual(len(events), 1) self.assertIs(events[0].attributes[-1].malware_binary, None) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_edit_attribute(self): first = self.create_simple_event() try: first.attributes[0].comment = 'This is the original comment' first = self.user_misp_connector.add_event(first) first.attributes[0].comment = 'This is the modified comment' attribute = self.user_misp_connector.update_attribute(first.attributes[0]) self.assertEqual(attribute.comment, 'This is the modified comment') attribute = self.user_misp_connector.change_comment(first.attributes[0].uuid, 'This is the modified comment, again') self.assertEqual(attribute['Attribute']['comment'], 'This is the modified comment, again') attribute = self.user_misp_connector.change_disable_correlation(first.attributes[0].uuid, True) self.assertEqual(attribute['Attribute']['disable_correlation'], True) attribute = self.user_misp_connector.change_disable_correlation(first.attributes[0].uuid, 0) self.assertEqual(attribute['Attribute']['disable_correlation'], False) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_sightings(self): first = self.create_simple_event() second = self.create_simple_event() try: first = self.user_misp_connector.add_event(first) second = self.user_misp_connector.add_event(second) current_ts = int(time.time()) self.user_misp_connector.sighting(value=first.attributes[0].value) self.user_misp_connector.sighting(value=second.attributes[0].value, source='Testcases', type='1') s = self.user_misp_connector.search_sightings(publish_timestamp=current_ts, include_attribute=True, include_event_meta=True, pythonify=True) self.assertEqual(len(s), 2) self.assertEqual(s[0]['event'].id, first.id) self.assertEqual(s[0]['attribute'].id, first.attributes[0].id) s = self.user_misp_connector.search_sightings(publish_timestamp=current_ts, source='Testcases', include_attribute=True, include_event_meta=True, pythonify=True) self.assertEqual(len(s), 1) self.assertEqual(s[0]['event'].id, second.id) self.assertEqual(s[0]['attribute'].id, second.attributes[0].id) s = self.user_misp_connector.search_sightings(publish_timestamp=current_ts, type_sighting='1', include_attribute=True, include_event_meta=True, pythonify=True) self.assertEqual(len(s), 1) self.assertEqual(s[0]['event'].id, second.id) self.assertEqual(s[0]['attribute'].id, second.attributes[0].id) s = self.user_misp_connector.search_sightings(context='event', context_id=first.id, pythonify=True) self.assertEqual(len(s), 1) self.assertEqual(s[0]['sighting'].event_id, str(first.id)) s = self.user_misp_connector.search_sightings(context='attribute', context_id=second.attributes[0].id, pythonify=True) self.assertEqual(len(s), 1) self.assertEqual(s[0]['sighting'].attribute_id, str(second.attributes[0].id)) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_search_csv(self): first = self.create_simple_event() first.attributes[0].comment = 'This is the original comment' second = self.create_simple_event() second.info = 'foo blah' second.set_date('2018-09-01') second.add_attribute('ip-src', '8.8.8.8') try: first = self.user_misp_connector.add_event(first) second = self.user_misp_connector.add_event(second) response = self.user_misp_connector.fast_publish(first.id, alert=False) self.assertEqual(response['errors'][1]['message'], 'You do not have permission to use this functionality.') # Default search, attribute with to_ids == True first.attributes[0].to_ids = True first = self.user_misp_connector.update_event(first) self.admin_misp_connector.fast_publish(first.id, alert=False) csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), pythonify=True) self.assertEqual(len(csv), 1) self.assertEqual(csv[0]['value'], first.attributes[0].value) # eventid csv = self.user_misp_connector.search(return_format='csv', eventid=first.id, pythonify=True) self.assertEqual(len(csv), 1) self.assertEqual(csv[0]['value'], first.attributes[0].value) # category csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), category='Other', pythonify=True) self.assertEqual(len(csv), 1) self.assertEqual(csv[0]['value'], first.attributes[0].value) csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), category='Person', pythonify=True) self.assertEqual(len(csv), 0) # type_attribute csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), type_attribute='text', pythonify=True) self.assertEqual(len(csv), 1) self.assertEqual(csv[0]['value'], first.attributes[0].value) csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), type_attribute='ip-src', pythonify=True) self.assertEqual(len(csv), 0) # context csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), include_context=True, pythonify=True) self.assertEqual(len(csv), 1) self.assertTrue('event_info' in csv[0]) # date_from date_to csv = self.user_misp_connector.search(return_format='csv', date_from=date.today().isoformat(), pythonify=True) self.assertEqual(len(csv), 1) self.assertEqual(csv[0]['value'], first.attributes[0].value) csv = self.user_misp_connector.search(return_format='csv', date_from='2018-09-01', date_to='2018-09-02', pythonify=True) self.assertEqual(len(csv), 2) # headerless csv = self.user_misp_connector.search(return_format='csv', date_from='2018-09-01', date_to='2018-09-02', headerless=True) # FIXME: The header is here. # print(csv) # Expects 2 lines after removing the empty ones. # self.assertEqual(len(csv.strip().split('\n')), 2) # include_context csv = self.user_misp_connector.search(return_format='csv', date_from='2018-09-01', date_to='2018-09-02', include_context=True, pythonify=True) event_context_keys = ['event_info', 'event_member_org', 'event_source_org', 'event_distribution', 'event_threat_level_id', 'event_analysis', 'event_date', 'event_tag', 'event_timestamp'] for k in event_context_keys: self.assertTrue(k in csv[0]) # requested_attributes columns = ['value', 'event_id'] csv = self.user_misp_connector.search(return_format='csv', date_from='2018-09-01', date_to='2018-09-02', requested_attributes=columns, pythonify=True) self.assertEqual(len(csv[0].keys()), 2) for k in columns: self.assertTrue(k in csv[0]) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_search_stix(self): first = self.create_simple_event() first.add_attribute('ip-src', '8.8.8.8') try: first = self.user_misp_connector.add_event(first) if not travis_run: stix = self.user_misp_connector.search(return_format='stix', eventid=first.id) found = re.findall('8.8.8.8', stix) self.assertTrue(found) stix2 = self.user_misp_connector.search(return_format='stix2', eventid=first.id) json.dumps(stix2, indent=2) self.assertEqual(stix2['objects'][-1]['pattern'], "[network-traffic:src_ref.type = 'ipv4-addr' AND network-traffic:src_ref.value = '8.8.8.8']") finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_upload_sample(self): first = self.create_simple_event() second = self.create_simple_event() third = self.create_simple_event() try: # Simple, not executable first = self.user_misp_connector.add_event(first) with open('tests/testlive_comprehensive.py', 'rb') as f: response = self.user_misp_connector.upload_sample(filename='testfile.py', filepath_or_bytes=f.read(), event_id=first.id) self.assertEqual(response['message'], 'Success, saved all attributes.') first = self.user_misp_connector.get_event(first.id) self.assertEqual(len(first.objects), 1) self.assertEqual(first.objects[0].name, 'file') # Simple, executable second = self.user_misp_connector.add_event(second) with open('tests/viper-test-files/test_files/whoami.exe', 'rb') as f: response = self.user_misp_connector.upload_sample(filename='whoami.exe', filepath_or_bytes=f.read(), event_id=second.id) self.assertEqual(response['message'], 'Success, saved all attributes.') second = self.user_misp_connector.get_event(second.id) self.assertEqual(len(second.objects), 1) self.assertEqual(second.objects[0].name, 'file') third = self.user_misp_connector.add_event(third) if not travis_run: # Advanced, executable with open('tests/viper-test-files/test_files/whoami.exe', 'rb') as f: response = self.user_misp_connector.upload_sample(filename='whoami.exe', filepath_or_bytes=f.read(), event_id=third.id, advanced_extraction=True) self.assertEqual(response['message'], 'Success, saved all attributes.') third = self.user_misp_connector.get_event(third.id) self.assertEqual(len(third.objects), 7) self.assertEqual(third.objects[0].name, 'pe-section') finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_update_object(self): first = self.create_simple_event() ip_dom = MISPObject('domain-ip') ip_dom.add_attribute('domain', value='google.fr') ip_dom.add_attribute('ip', value='8.8.8.8') first.add_object(ip_dom) try: # Update with full event first = self.user_misp_connector.add_event(first) first.objects[0].add_attribute('ip', value='8.9.9.8') first.objects[0].add_attribute('ip', '8.9.9.10') first = self.user_misp_connector.update_event(first) self.assertEqual(first.objects[0].attributes[2].value, '8.9.9.8') self.assertEqual(first.objects[0].attributes[3].value, '8.9.9.10') # Update object only misp_object = self.user_misp_connector.get_object(first.objects[0].id) misp_object.attributes[2].value = '8.9.9.9' misp_object = self.user_misp_connector.update_object(misp_object) self.assertEqual(misp_object.attributes[2].value, '8.9.9.9') # Test with add_attributes second = self.create_simple_event() ip_dom = MISPObject('domain-ip') ip_dom.add_attribute('domain', value='google.fr') ip_dom.add_attributes('ip', {'value': '10.8.8.8', 'to_ids': False}, '10.9.8.8') ip_dom.add_attributes('ip', '11.8.8.8', '11.9.8.8') second.add_object(ip_dom) second = self.user_misp_connector.add_event(second) self.assertEqual(len(second.objects[0].attributes), 5) self.assertFalse(second.objects[0].attributes[1].to_ids) self.assertTrue(second.objects[0].attributes[2].to_ids) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) def test_domain_ip_object(self): first = self.create_simple_event() try: dom_ip_obj = DomainIPObject({'ip': ['1.1.1.1', {'value': '2.2.2.2', 'to_ids': False}], 'first-seen': '20190101', 'last-seen': '2019-02-03', 'domain': 'circl.lu'}) first.add_object(dom_ip_obj) first = self.user_misp_connector.add_event(first) self.assertEqual(len(first.objects[0].attributes), 5) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_asn_object(self): first = self.create_simple_event() try: dom_ip_obj = ASNObject({'asn': '12345', 'first-seen': '20190101', 'last-seen': '2019-02-03'}) first.add_object(dom_ip_obj) first = self.user_misp_connector.add_event(first) self.assertEqual(len(first.objects[0].attributes), 3) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_object_template(self): r = self.admin_misp_connector.update_object_templates() self.assertEqual(type(r), list) if not travis_run: template = self.admin_misp_connector.get_object_template('688c46fb-5edb-40a3-8273-1af7923e2215') self.assertEqual(template['ObjectTemplate']['uuid'], '688c46fb-5edb-40a3-8273-1af7923e2215') def test_tags(self): # Get list tags = self.admin_misp_connector.get_tags_list() self.assertTrue(isinstance(tags, list)) # Get tag for tag in tags: if not tag['hide_tag']: break tag = self.admin_misp_connector.get_tag(tags[0]['id']) self.assertTrue('name' in tag) r = self.admin_misp_connector.disable_tag(tag['id']) self.assertTrue(r['Tag']['hide_tag']) r = self.admin_misp_connector.enable_tag(tag['id']) self.assertFalse(r['Tag']['hide_tag']) def test_add_event_with_attachment_object_controller(self): first = self.create_simple_event() try: first = self.user_misp_connector.add_event(first) fo, peo, seos = make_binary_objects('tests/viper-test-files/test_files/whoami.exe') for s in seos: r = self.user_misp_connector.add_object(first.id, s) self.assertEqual(r.name, 'pe-section', r) r = self.user_misp_connector.add_object(first.id, peo) self.assertEqual(r.name, 'pe', r) for ref in peo.ObjectReference: r = self.user_misp_connector.add_object_reference(ref) self.assertTrue('ObjectReference' in r, r) r = self.user_misp_connector.add_object(first.id, fo) self.assertEqual(r.name, 'file', r) for ref in fo.ObjectReference: r = self.user_misp_connector.add_object_reference(ref) self.assertTrue('ObjectReference' in r, r) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_add_event_with_attachment(self): first = self.create_simple_event() try: first = self.user_misp_connector.add_event(first) file_obj, bin_obj, sections = make_binary_objects('tests/viper-test-files/test_files/whoami.exe', standalone=False) first.add_object(file_obj) first.add_object(bin_obj) for s in sections: first.add_object(s) self.assertEqual(len(first.objects[0].references), 1) self.assertEqual(first.objects[0].references[0].relationship_type, 'included-in') first = self.user_misp_connector.update_event(first) self.assertEqual(len(first.objects[0].references), 1) self.assertEqual(first.objects[0].references[0].relationship_type, 'included-in') finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_taxonomies(self): # Make sure we're up-to-date r = self.admin_misp_connector.update_taxonomies() self.assertEqual(r['name'], 'All taxonomy libraries are up to date already.') # Get list taxonomies = self.admin_misp_connector.get_taxonomies_list() self.assertTrue(isinstance(taxonomies, list)) list_name_test = 'tlp' for tax in taxonomies: if tax['Taxonomy']['namespace'] == list_name_test: break if not travis_run: r = self.admin_misp_connector.get_taxonomy(tax['Taxonomy']['id']) self.assertEqual(r['Taxonomy']['namespace'], list_name_test) self.assertTrue('enabled' in r['Taxonomy']) r = self.admin_misp_connector.enable_taxonomy(tax['Taxonomy']['id']) self.assertEqual(r['message'], 'Taxonomy enabled') r = self.admin_misp_connector.disable_taxonomy(tax['Taxonomy']['id']) self.assertEqual(r['message'], 'Taxonomy disabled') def test_warninglists(self): # Make sure we're up-to-date r = self.admin_misp_connector.update_warninglists() self.assertTrue('name' in r, msg=r) try: self.assertEqual(r['name'], 'All warninglists are up to date already.', msg=r) except Exception: print(r) # Get list r = self.admin_misp_connector.get_warninglists() # FIXME It returns Warninglists object instead of a list of warning lists directly. This is inconsistent. warninglists = r['Warninglists'] self.assertTrue(isinstance(warninglists, list)) list_name_test = 'List of known hashes with common false-positives (based on Florian Roth input list)' for wl in warninglists: if wl['Warninglist']['name'] == list_name_test: break testwl = wl['Warninglist'] r = self.admin_misp_connector.get_warninglist(testwl['id']) self.assertEqual(r['Warninglist']['name'], list_name_test) self.assertTrue('WarninglistEntry' in r['Warninglist']) r = self.admin_misp_connector.enable_warninglist(testwl['id']) self.assertEqual(r['success'], '1 warninglist(s) enabled') r = self.admin_misp_connector.disable_warninglist(testwl['id']) self.assertEqual(r['success'], '1 warninglist(s) disabled') def test_noticelists(self): # Make sure we're up-to-date r = self.admin_misp_connector.update_noticelists() self.assertEqual(r['name'], 'All noticelists are up to date already.') # Get list noticelists = self.admin_misp_connector.get_noticelists() self.assertTrue(isinstance(noticelists, list)) list_name_test = 'gdpr' for nl in noticelists: if nl['Noticelist']['name'] == list_name_test: break testnl = nl r = self.admin_misp_connector.get_noticelist(testnl['Noticelist']['id']) self.assertEqual(r['Noticelist']['name'], list_name_test) self.assertTrue('NoticelistEntry' in r['Noticelist']) r = self.admin_misp_connector.enable_noticelist(testnl['Noticelist']['id']) self.assertTrue(r['Noticelist']['enabled']) r = self.admin_misp_connector.disable_noticelist(testnl['Noticelist']['id']) self.assertFalse(r['Noticelist']['enabled']) def test_galaxies(self): if not travis_run: # Make sure we're up-to-date r = self.admin_misp_connector.update_galaxies() self.assertEqual(r['name'], 'Galaxies updated.') # Get list galaxies = self.admin_misp_connector.get_galaxies() self.assertTrue(isinstance(galaxies, list)) list_name_test = 'Mobile Attack - Attack Pattern' for galaxy in galaxies: if galaxy['Galaxy']['name'] == list_name_test: break r = self.admin_misp_connector.get_galaxy(galaxy['Galaxy']['id']) self.assertEqual(r['Galaxy']['name'], list_name_test) self.assertTrue('GalaxyCluster' in r) def test_zmq(self): first = self.create_simple_event() try: first = self.user_misp_connector.add_event(first) if not travis_run: r = self.admin_misp_connector.pushEventToZMQ(first.id) self.assertEqual(r['message'], 'Event published to ZMQ') finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_csv_loader(self): csv1 = CSVLoader(template_name='file', csv_path=Path('tests/csv_testfiles/valid_fieldnames.csv')) event = MISPEvent() event.info = 'Test event from CSV loader' for o in csv1.load(): event.add_object(**o) csv2 = CSVLoader(template_name='file', csv_path=Path('tests/csv_testfiles/invalid_fieldnames.csv'), fieldnames=['SHA1', 'fileName', 'size-in-bytes'], has_fieldnames=True) try: first = self.user_misp_connector.add_event(event) for o in csv2.load(): new_object = self.user_misp_connector.add_object(first.id, o) self.assertEqual(len(new_object.attributes), 3) finally: # Delete event self.admin_misp_connector.delete_event(first.id) def test_user(self): user = self.user_misp_connector.get_user() self.assertEqual(user.authkey, self.test_usr.authkey) def test_attribute(self): first = self.create_simple_event() try: first = self.user_misp_connector.add_event(first) # Get attribute attribute = self.user_misp_connector.get_attribute(first.attributes[0].id) self.assertEqual(first.attributes[0].uuid, attribute.uuid) # Add attribute new_attribute = MISPAttribute() new_attribute.value = '1.2.3.4' new_attribute.type = 'ip-dst' new_attribute = self.user_misp_connector.add_attribute(first.id, new_attribute) self.assertEqual(new_attribute.value, '1.2.3.4') # Add attribute as proposal new_proposal = MISPAttribute() new_proposal.value = '5.2.3.4' new_proposal.type = 'ip-dst' new_proposal.category = 'Network activity' new_proposal = self.user_misp_connector.add_attribute_proposal(first.id, new_proposal) self.assertEqual(new_proposal.value, '5.2.3.4') # Update attribute new_attribute.value = '5.6.3.4' new_attribute = self.user_misp_connector.update_attribute(new_attribute) self.assertEqual(new_attribute.value, '5.6.3.4') # Update attribute as proposal new_proposal_update = self.user_misp_connector.update_attribute_proposal(new_attribute.id, {'to_ids': False}) self.assertEqual(new_proposal_update.to_ids, False) # Get attribute proposal temp_new_proposal = self.user_misp_connector.get_attribute_proposal(new_proposal.id) self.assertEqual(temp_new_proposal.uuid, new_proposal.uuid) # Accept attribute proposal - New attribute self.user_misp_connector.accept_attribute_proposal(new_proposal.id) first = self.user_misp_connector.get_event(first.id) self.assertEqual(first.attributes[-1].value, '5.2.3.4') # Accept attribute proposal - Attribute update response = self.user_misp_connector.accept_attribute_proposal(new_proposal_update.id) self.assertEqual(response['message'], 'Proposed change accepted.') attribute = self.user_misp_connector.get_attribute(new_attribute.id) self.assertEqual(attribute.to_ids, False) # Discard attribute proposal new_proposal_update = self.user_misp_connector.update_attribute_proposal(new_attribute.id, {'to_ids': True}) response = self.user_misp_connector.discard_attribute_proposal(new_proposal_update.id) self.assertEqual(response['message'], 'Proposal discarded.') attribute = self.user_misp_connector.get_attribute(new_attribute.id) self.assertEqual(attribute.to_ids, False) finally: # Delete event self.admin_misp_connector.delete_event(first.id) @unittest.skip("Currently failing") def test_search_type_event_csv(self): try: first, second, third = self.environment() # Search as admin events = self.admin_misp_connector.search(return_format='csv', timestamp=first.timestamp.timestamp()) print(events) attributes_types_search = self.admin_misp_connector.build_complex_query(or_parameters=['ip-src', 'ip-dst']) events = self.admin_misp_connector.search(return_format='csv', timestamp=first.timestamp.timestamp(), type_attribute=attributes_types_search) print(events) finally: # Delete event self.admin_misp_connector.delete_event(first.id) self.admin_misp_connector.delete_event(second.id) self.admin_misp_connector.delete_event(third.id) def test_live_acl(self): missing_acls = self.admin_misp_connector.get_live_query_acl() self.assertEqual(missing_acls, [], msg=missing_acls) def test_roles(self): role = self.admin_misp_connector.set_default_role(4) self.assertEqual(role['message'], 'Default role set.') self.admin_misp_connector.set_default_role(3) if __name__ == '__main__': unittest.main()