How to use pymisp - 10 common examples

def test_auth_error(self, m):
        self.initURI(m)
        pymisp = PyMISP(self.domain, self.key)
        error = pymisp.get(1)
        response = self.auth_error_msg
        response['errors'] = [response['message']]
        self.assertEqual(error, response)

def test_searchIndexByTagName(self, m):
        self.initURI(m)
        pymisp = PyMISP(self.domain, self.key)
        response = pymisp.search_index(tag='ecsirt:malicious-code="ransomware"')
        self.assertEqual(response['response'], self.search_index_result)

def test_change_disablecorrelation_invalid(self, m):
        self.initURI(m)
        pymisp = PyMISP(self.domain, self.key)
        try:
            pymisp.change_disablecorrelation(self.key, 42)
            self.assertFalse('Exception required for off domain value')
        except Exception:
            pass

* +
        '''
        # Creates a test user for roles
        user = MISPUser()
        user.email = 'testusr-roles@user.local'
        user.org_id = self.test_org.id
        tag = MISPTag()
        tag.name = 'tlp:white___test'
        try:
            test_roles_user = self.admin_misp_connector.add_user(user, pythonify=True)
            test_tag = self.admin_misp_connector.add_tag(tag, pythonify=True)
            test_roles_user_connector = ExpandedPyMISP(url, test_roles_user.authkey, verifycert, debug=False)
            test_roles_user_connector.toggle_global_pythonify()
            # ===== Read Only
            self.admin_misp_connector.update_user({'role_id': 6}, test_roles_user)
            base_event = MISPEvent()
            base_event.info = 'Test Roles'
            base_event.distribution = 0
            base_event.add_attribute('ip-dst', '8.8.8.8')
            base_event.add_attribute('ip-dst', '9.9.9.9')
            base_event.attributes[0].add_tag('tlp:white___test')
            r = test_roles_user_connector.add_event(base_event)
            self.assertTrue(isinstance(r['errors'], tuple), r['errors'])
            self.assertEqual(r['errors'][1]['message'], 'You do not have permission to use this functionality.', r)
            try:
                e = self.user_misp_connector.add_event(base_event, pythonify=True)
                e = test_roles_user_connector.get_event(e)
                self.assertEqual(e.info, 'Test Roles')
                self.assertEqual(e.attributes[0].tags[0].name, 'tlp:white___test')
                r = test_roles_user_connector.publish(e)
                self.assertEqual(r['errors'][1]['message'], 'You do not have permission to use this functionality.', r)
                r = test_roles_user_connector.tag(e.attributes[1], 'tlp:white___test')

attributes = self.user_misp_connector.search(controller='attributes', value=first.attributes[0].value)
            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()))
            self.assertEqual(attributes, [])

            # Include context - search as user (can only see one event)
            attributes = self.user_misp_connector.search(controller='attributes', value=first.attributes[0].value, include_context=True, pythonify=True)
            self.assertTrue(isinstance(attributes[0].Event, MISPEvent))
            self.assertEqual(attributes[0].Event.uuid, second.uuid)

            # Include context - search as admin (can see both event)
            attributes = self.admin_misp_connector.search(controller='attributes', value=first.attributes[0].value, include_context=True, pythonify=True)
            self.assertTrue(isinstance(attributes[0].Event, MISPEvent))
            self.assertEqual(attributes[0].Event.uuid, first.uuid)
            self.assertEqual(attributes[1].Event.uuid, second.uuid)

            # Include correlations - search as admin (can see both event)
            attributes = self.admin_misp_connector.search(controller='attributes', value=first.attributes[0].value, include_correlations=True, pythonify=True)
            self.assertTrue(isinstance(attributes[0].Event, MISPEvent))
            self.assertEqual(attributes[0].Event.uuid, first.uuid)
            self.assertEqual(attributes[1].Event.uuid, second.uuid)
            self.assertEqual(attributes[0].RelatedAttribute[0].Event.uuid, second.uuid)
            self.assertEqual(attributes[1].RelatedAttribute[0].Event.uuid, first.uuid)

            # Include sightings - search as admin (can see both event)
            self.admin_misp_connector.add_sighting({'value': first.attributes[0].value})
            attributes = self.admin_misp_connector.search(controller='attributes', value=first.attributes[0].value, include_sightings=True, pythonify=True)
            self.assertTrue(isinstance(attributes[0].Event, MISPEvent))
            self.assertEqual(attributes[0].Event.uuid, first.uuid)

def test_sync_all_communities(self):
        '''Simple event, all communities, enable automatic push on two sub-instances'''
        event = MISPEvent()
        event.info = 'Event created on first instance - test_sync_all_communities'
        event.distribution = Distribution.all_communities
        event.add_attribute('ip-src', '1.1.1.1')
        try:
            source = self.instances[0]
            server = source.site_admin_connector.update_server({'push': True}, source.sync_servers[0].id)
            self.assertTrue(server.push)
            middle = self.instances[1]
            middle.site_admin_connector.update_server({'push': True}, middle.sync_servers[1].id)  # Enable automatic push to 3rd instance
            last = self.instances[2]
            event = source.user_connector.add_event(event)
            source.org_admin_connector.publish(event)
            source.site_admin_connector.server_push(source.sync_servers[0])
            time.sleep(30)
            middle_event = middle.user_connector.get_event(event.uuid)
            self.assertEqual(event.attributes[0].value, middle_event.attributes[0].value)

def test_batch_image_events(self):
        # Test case ONLY for manual testing. Needs to download a full list of image events !

        if self.check_python_2():
            self.assertTrue(True)
        elif not manual_testing:
            self.assertTrue(True)
        else:
            self.init_event()

            file_nb = str(len(os.listdir(self.test_image_folder)))
            i = 0
            t = time.time()
            for curr_file in os.listdir(self.test_image_folder):
                self.mispevent = MISPEvent()
                file_path = self.test_image_folder + curr_file

                print("Current file : " + file_path + " " + str(i) + " over " + file_nb)
                i += 1

                self.mispevent.load_file(file_path)

                reportlab_generator.register_value_to_file(
                    reportlab_generator.convert_event_in_pdf_buffer(self.mispevent),
                    self.storage_image_folder + curr_file + ".pdf")
            print("Elapsed time : " + str(time.time() - t))
            # Local run : 73.061s for 102 files

def test_batch_OSINT_events(self):
        # Test case ONLY for manual testing. Needs to download a full list of OSINT events !

        if self.check_python_2():
            self.assertTrue(True)
        elif not manual_testing:
            self.assertTrue(True)
        else:
            self.init_event()

            file_nb = str(len(os.listdir(self.test_batch_folder)))
            i = 0
            t = time.time()
            for curr_file in os.listdir(self.test_batch_folder):
                self.mispevent = MISPEvent()
                file_path = self.test_batch_folder + curr_file

                print("Current file : " + file_path + " " + str(i) + " over " + file_nb)
                i += 1

                self.mispevent.load_file(file_path)

                reportlab_generator.register_value_to_file(
                    reportlab_generator.convert_event_in_pdf_buffer(self.mispevent),
                    self.storage_folder_OSINT + curr_file + ".pdf")
            print("Elapsed time : " + str(time.time() - t))
            # Local run : 1958.930s for 1064 files

def test_sync_community(self):
        '''Simple event, this community only, pull from member of the community'''
        event = MISPEvent()
        event.info = 'Event created on first instance - test_sync_community'
        event.distribution = Distribution.this_community_only
        event.add_attribute('ip-src', '1.1.1.1')
        try:
            source = self.instances[0]
            dest = self.instances[1]
            event = source.org_admin_connector.add_event(event)
            source.org_admin_connector.publish(event)
            dest.site_admin_connector.server_pull(dest.sync_servers[0])
            time.sleep(10)
            dest_event = dest.org_admin_connector.get_event(event.uuid)
            self.assertEqual(dest_event.distribution, 0)
        finally:
            source.org_admin_connector.delete_event(event)
            dest.site_admin_connector.delete_event(dest_event)

def __init__(self, params):
        self.initial_user_connector = ExpandedPyMISP(params['url'], params['key'], ssl=False, debug=False)
        # Git pull
        self.initial_user_connector.update_misp()
        # Set the default role (id 3 on the VM is normal user)
        self.initial_user_connector.set_default_role(3)
        # Restart workers
        self.initial_user_connector.restart_workers()
        if not fast_mode:
            # Load submodules
            self.initial_user_connector.update_object_templates()
            self.initial_user_connector.update_galaxies()
            self.initial_user_connector.update_noticelists()
            self.initial_user_connector.update_warninglists()
            self.initial_user_connector.update_taxonomies()

        self.initial_user_connector.toggle_global_pythonify()