PyMISP/tests/testlive_comprehensive.py

2131 lines
112 KiB
Python
Raw Normal View History

2018-08-10 19:04:02 +02:00
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
2018-08-10 19:04:02 +02:00
import unittest
from pymisp.tools import make_binary_objects
from datetime import datetime, timedelta, date, timezone
from io import BytesIO
2019-02-01 11:07:42 +01:00
import json
from pathlib import Path
2018-08-10 19:04:02 +02:00
2020-01-23 10:27:40 +01:00
import urllib3 # type: ignore
2018-08-10 19:04:02 +02:00
import time
2018-12-10 14:08:23 +01:00
from uuid import uuid4
2018-08-10 19:04:02 +02:00
2019-08-30 11:45:56 +02:00
import email
2019-09-11 14:20:55 +02:00
from collections import defaultdict
2019-02-07 15:53:36 +01:00
import logging
logging.disable(logging.CRITICAL)
2019-11-29 16:33:17 +01:00
logger = logging.getLogger('pymisp')
2019-02-07 15:53:36 +01:00
try:
2019-10-16 17:22:19 +02:00
from pymisp import ExpandedPyMISP, MISPEvent, MISPOrganisation, MISPUser, Distribution, ThreatLevel, Analysis, MISPObject, MISPAttribute, MISPSighting, MISPShadowAttribute, MISPTag, MISPSharingGroup, MISPFeed, MISPServer, MISPUserSetting
2019-07-12 17:35:02 +02:00
from pymisp.tools import CSVLoader, DomainIPObject, ASNObject, GenericObjectGenerator
2019-09-11 14:20:55 +02:00
from pymisp.exceptions import MISPServerError
except ImportError:
if sys.version_info < (3, 6):
print('This test suite requires Python 3.6+, breaking.')
sys.exit(0)
else:
raise
2018-08-10 19:04:02 +02:00
try:
2020-01-23 10:27:40 +01:00
from keys import url, key # type: ignore
2018-12-10 14:08:23 +01:00
verifycert = False
2018-08-12 01:21:49 +02:00
except ImportError as e:
print(e)
2019-07-12 17:35:02 +02:00
url = 'https://localhost:8443'
2019-08-05 15:59:24 +02:00
key = 'd6OmdDFvU3Seau3UjwvHS1y3tFQbaRNhJhDX0tjh'
verifycert = False
2018-08-10 19:04:02 +02:00
2019-07-12 17:35:02 +02:00
urllib3.disable_warnings()
fast_mode = False
2019-08-20 15:34:21 +02:00
2019-07-12 17:35:02 +02:00
2018-08-10 19:04:02 +02:00
class TestComprehensive(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.maxDiff = None
# Connect as admin
cls.admin_misp_connector = ExpandedPyMISP(url, key, verifycert, debug=False)
2019-08-20 15:34:21 +02:00
if not fast_mode:
r = cls.admin_misp_connector.update_misp()
print(r)
2018-08-10 19:04:02 +02:00
# Creates an org
organisation = MISPOrganisation()
organisation.name = 'Test Org'
cls.test_org = cls.admin_misp_connector.add_organisation(organisation, pythonify=True)
2019-08-20 15:34:21 +02:00
# Create an org to delegate to
organisation = MISPOrganisation()
organisation.name = 'Test Org - delegate'
cls.test_org_delegate = cls.admin_misp_connector.add_organisation(organisation, pythonify=True)
2019-04-04 16:39:17 +02:00
# Set the refault role (id 3 on the VM)
cls.admin_misp_connector.set_default_role(3)
2018-08-10 19:04:02 +02:00
# Creates a user
user = MISPUser()
user.email = 'testusr@user.local'
user.org_id = cls.test_org.id
2019-07-12 17:35:02 +02:00
cls.test_usr = cls.admin_misp_connector.add_user(user, pythonify=True)
2019-11-29 16:33:17 +01:00
cls.user_misp_connector = ExpandedPyMISP(url, cls.test_usr.authkey, verifycert, debug=True)
cls.user_misp_connector.toggle_global_pythonify()
2018-08-10 19:04:02 +02:00
# Creates a publisher
user = MISPUser()
user.email = 'testpub@user.local'
user.org_id = cls.test_org.id
user.role_id = 4
2019-07-12 17:35:02 +02:00
cls.test_pub = cls.admin_misp_connector.add_user(user, pythonify=True)
cls.pub_misp_connector = ExpandedPyMISP(url, cls.test_pub.authkey, verifycert)
# Creates a user that can accept a delegation request
user = MISPUser()
user.email = 'testusr@delegate.recipient.local'
user.org_id = cls.test_org_delegate.id
user.role_id = 2
cls.test_usr_delegate = cls.admin_misp_connector.add_user(user, pythonify=True)
cls.delegate_user_misp_connector = ExpandedPyMISP(url, cls.test_usr_delegate.authkey, verifycert, debug=False)
cls.delegate_user_misp_connector.toggle_global_pythonify()
2019-08-20 15:34:21 +02:00
if not fast_mode:
# 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()
2018-08-10 19:04:02 +02:00
@classmethod
def tearDownClass(cls):
# Delete publisher
2019-08-16 10:48:06 +02:00
cls.admin_misp_connector.delete_user(cls.test_pub)
2018-08-10 19:04:02 +02:00
# Delete user
2019-08-16 10:48:06 +02:00
cls.admin_misp_connector.delete_user(cls.test_usr)
cls.admin_misp_connector.delete_user(cls.test_usr_delegate)
2018-08-10 19:04:02 +02:00
# Delete org
2019-08-16 10:48:06 +02:00
cls.admin_misp_connector.delete_organisation(cls.test_org)
2019-08-20 15:34:21 +02:00
cls.admin_misp_connector.delete_organisation(cls.test_org_delegate)
2018-08-10 19:04:02 +02:00
2018-08-21 00:32:27 +02:00
def create_simple_event(self, force_timestamps=False):
2018-08-10 19:04:02 +02:00
mispevent = MISPEvent(force_timestamps=force_timestamps)
2018-08-21 00:32:27 +02:00
mispevent.info = 'This is a super simple test'
2018-08-10 19:04:02 +02:00
mispevent.distribution = Distribution.your_organisation_only
mispevent.threat_level_id = ThreatLevel.low
mispevent.analysis = Analysis.completed
mispevent.add_attribute('text', str(uuid4()))
return mispevent
2018-08-20 18:27:06 +02:00
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', 'FIRST_EVENT' + str(uuid4()))
2018-08-21 11:16:51 +02:00
first_event.attributes[0].add_tag('admin_only')
first_event.attributes[0].add_tag('tlp:white___test')
2018-08-21 14:34:29 +02:00
first_event.add_attribute('text', str(uuid4()))
first_event.attributes[1].add_tag('unique___test')
2018-08-20 18:27:06 +02:00
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', 'SECOND_EVENT' + str(uuid4()))
2018-08-20 18:27:06 +02:00
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')
2018-08-20 18:27:06 +02:00
# 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', 'THIRD_EVENT' + str(uuid4()))
2018-08-20 18:27:06 +02:00
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, pythonify=True)
third = self.admin_misp_connector.add_event(third_event, pythonify=True)
2018-08-20 18:27:06 +02:00
# Create second event as user
second = self.user_misp_connector.add_event(second_event)
return first, second, third
2019-08-07 18:21:46 +02:00
def test_server_settings(self):
settings = self.admin_misp_connector.server_settings()
for final_setting in settings['finalSettings']:
if final_setting['setting'] == 'MISP.max_correlations_per_event':
self.assertEqual(final_setting['value'], 5000)
break
2019-08-16 11:28:07 +02:00
r = self.admin_misp_connector.set_server_setting('MISP.max_correlations_per_event', 10)
self.assertEqual(r['message'], 'Field updated', r)
2019-08-16 10:48:06 +02:00
setting = self.admin_misp_connector.get_server_setting('MISP.max_correlations_per_event')
self.assertEqual(setting['value'], 10)
2019-08-16 11:28:07 +02:00
r = self.admin_misp_connector.set_server_setting('MISP.max_correlations_per_event', 5000)
self.assertEqual(r['message'], 'Field updated', r)
2019-08-07 18:21:46 +02:00
2019-08-16 10:48:06 +02:00
setting = self.admin_misp_connector.get_server_setting('MISP.live')
self.assertTrue(setting['value'])
2019-08-16 11:28:07 +02:00
r = self.admin_misp_connector.set_server_setting('MISP.live', False, force=True)
self.assertEqual(r['message'], 'Field updated', r)
2019-08-16 10:48:06 +02:00
setting = self.admin_misp_connector.get_server_setting('MISP.live')
self.assertFalse(setting['value'])
2019-08-16 11:28:07 +02:00
r = self.admin_misp_connector.set_server_setting('MISP.live', True, force=True)
self.assertEqual(r['message'], 'Field updated', r)
2019-08-16 10:48:06 +02:00
setting = self.admin_misp_connector.get_server_setting('MISP.live')
self.assertTrue(setting['value'])
2019-08-07 18:21:46 +02:00
2018-08-10 19:04:02 +02:00
def test_search_value_event(self):
2018-08-20 18:27:06 +02:00
'''Search a value on the event controller
* Test ACL admin user vs normal user in an other org
* Make sure we have one match
'''
2018-08-10 19:04:02 +02:00
try:
2018-08-20 18:27:06 +02:00
first, second, third = self.environment()
2018-08-10 19:04:02 +02:00
# Search as admin
events = self.admin_misp_connector.search(value=first.attributes[0].value, pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 2)
for e in events:
self.assertIn(e.id, [first.id, second.id])
2018-08-20 18:27:06 +02:00
# Search as user
events = self.user_misp_connector.search(value=first.attributes[0].value)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
for e in events:
self.assertIn(e.id, [second.id])
2018-08-20 18:27:06 +02:00
# Non-existing value
events = self.user_misp_connector.search(value=str(uuid4()))
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
2018-08-20 18:27:06 +02:00
finally:
# Delete events
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-20 18:27:06 +02:00
def test_search_value_attribute(self):
'''Search value in attributes controller'''
2018-08-20 18:27:06 +02:00
try:
first, second, third = self.environment()
# Search as admin
attributes = self.admin_misp_connector.search(controller='attributes', value=first.attributes[0].value, pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(attributes), 2)
for a in attributes:
self.assertIn(a.event_id, [first.id, second.id])
2018-08-10 19:04:02 +02:00
# Search as user
attributes = self.user_misp_connector.search(controller='attributes', value=first.attributes[0].value)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(attributes), 1)
for a in attributes:
self.assertIn(a.event_id, [second.id])
2018-08-20 18:27:06 +02:00
# Non-existing value
attributes = self.user_misp_connector.search(controller='attributes', value=str(uuid4()))
2018-08-21 11:16:51 +02:00
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)
self.assertEqual(attributes[1].Event.uuid, second.uuid)
self.assertTrue(isinstance(attributes[0].Sighting[0], MISPSighting))
2018-08-10 19:04:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-10 19:04:02 +02:00
2018-08-20 18:27:06 +02:00
def test_search_type_event(self):
'''Search multiple events, search events containing attributes with specific types'''
2018-08-19 14:35:32 +02:00
try:
2018-08-20 18:27:06 +02:00
first, second, third = self.environment()
2018-08-19 14:35:32 +02:00
# Search as admin
events = self.admin_misp_connector.search(timestamp=first.timestamp.timestamp(), pythonify=True)
2018-08-21 11:16:51 +02:00
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)
2018-08-21 11:35:36 +02:00
self.assertEqual(len(events), 2)
2018-08-21 11:16:51 +02:00
for e in events:
2018-08-21 11:35:36 +02:00
self.assertIn(e.id, [second.id, third.id])
2018-08-19 14:35:32 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-19 14:35:32 +02:00
2018-08-20 18:27:06 +02:00
def test_search_type_attribute(self):
'''Search multiple attributes, search attributes with specific types'''
2018-08-19 14:35:32 +02:00
try:
2018-08-20 18:27:06 +02:00
first, second, third = self.environment()
2018-08-19 14:35:32 +02:00
# Search as admin
2018-08-21 11:16:51 +02:00
attributes = self.admin_misp_connector.search(controller='attributes',
timestamp=first.timestamp.timestamp(), pythonify=True)
2018-08-21 14:34:29 +02:00
self.assertEqual(len(attributes), 8)
2018-08-21 11:16:51 +02:00
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)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(attributes), 3)
for a in attributes:
self.assertIn(a.event_id, [second.id, third.id])
2018-08-19 14:35:32 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-19 14:35:32 +02:00
2018-08-20 18:27:06 +02:00
def test_search_tag_event(self):
'''Search Tags at events level'''
2018-08-17 15:09:17 +02:00
try:
2018-08-20 18:27:06 +02:00
first, second, third = self.environment()
2018-08-17 15:09:17 +02:00
# Search as admin
events = self.admin_misp_connector.search(tags='tlp:white___test', pythonify=True)
2018-08-21 11:16:51 +02:00
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)
2018-08-21 11:16:51 +02:00
for e in events:
self.assertIn(e.id, [second.id, third.id])
events = self.admin_misp_connector.search(tags='admin_only', pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
for e in events:
self.assertIn(e.id, [first.id])
2018-08-17 15:09:17 +02:00
# Search as user
events = self.user_misp_connector.search(tags='tlp:white___test')
2018-08-21 11:16:51 +02:00
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')
self.assertEqual(len(events), 2)
2018-08-21 11:16:51 +02:00
for e in events:
self.assertIn(e.id, [second.id, third.id])
events = self.user_misp_connector.search(tags='admin_only')
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
2018-08-17 15:09:17 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-10 19:04:02 +02:00
2018-08-20 18:27:06 +02:00
def test_search_tag_attribute(self):
'''Search Tags at attributes level'''
2018-08-17 15:09:17 +02:00
try:
2018-08-20 18:27:06 +02:00
first, second, third = self.environment()
2018-08-17 15:09:17 +02:00
# Search as admin
attributes = self.admin_misp_connector.search(controller='attributes', tags='tlp:white___test', pythonify=True)
2018-08-21 11:16:51 +02:00
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)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(attributes), 1)
2018-08-17 15:09:17 +02:00
# Search as user
attributes = self.user_misp_connector.search(controller='attributes', tags='tlp:white___test')
2018-08-21 11:16:51 +02:00
self.assertEqual(len(attributes), 4)
attributes = self.user_misp_connector.search(controller='attributes', tags='tlp:amber___test')
self.assertEqual(len(attributes), 3)
attributes = self.user_misp_connector.search(tags='admin_only')
2018-08-21 11:16:51 +02:00
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)
self.assertEqual(len(attributes), 1)
2018-08-17 15:09:17 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-10 19:04:02 +02:00
2018-08-20 18:27:06 +02:00
def test_search_tag_advanced_event(self):
'''Advanced search Tags at events level'''
2018-08-20 18:27:06 +02:00
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)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 3)
2018-08-20 18:27:06 +02:00
for e in events:
2018-08-21 11:16:51 +02:00
self.assertIn(e.id, [first.id, second.id, third.id])
2018-08-20 18:27:06 +02:00
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'], [])
2018-08-21 11:16:51 +02:00
2018-08-21 14:34:29 +02:00
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)
2018-08-21 14:34:29 +02:00
self.assertEqual(len(events), 1)
2018-08-21 11:16:51 +02:00
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'], [])
2018-08-20 18:27:06 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-20 18:27:06 +02:00
def test_search_tag_advanced_attributes(self):
'''Advanced search Tags at attributes level'''
2018-08-20 18:27:06 +02:00
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)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(attributes), 3)
2018-08-20 18:27:06 +02:00
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
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2018-08-20 18:27:06 +02:00
def test_search_timestamp_event(self):
'''Search specific update timestamps at events level'''
2018-08-10 19:04:02 +02:00
# Creating event 1 - timestamp 5 min ago
2018-08-21 00:32:27 +02:00
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')
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
self.assertEqual(events[0].timestamp.timestamp(), int(event_creation_timestamp_second.timestamp()))
2018-08-21 00:32:27 +02:00
# # Test timestamp of 2nd event
events = self.user_misp_connector.search(timestamp=event_creation_timestamp_second.timestamp())
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
self.assertEqual(events[0].timestamp.timestamp(), int(event_creation_timestamp_second.timestamp()))
2018-08-21 00:32:27 +02:00
# # Test interval -6 min -> -4 min
events = self.user_misp_connector.search(timestamp=['6m', '4m'])
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, first.id)
self.assertEqual(events[0].timestamp.timestamp(), int(event_creation_timestamp_first.timestamp()))
2018-08-21 00:32:27 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
2018-08-21 00:32:27 +02:00
2018-08-21 11:16:51 +02:00
def test_search_timestamp_attribute(self):
'''Search specific update timestamps at attributes level'''
2018-08-21 00:32:27 +02:00
# Creating event 1 - timestamp 5 min ago
first = self.create_simple_event(force_timestamps=True)
2018-08-10 19:04:02 +02:00
event_creation_timestamp_first = datetime.now() - timedelta(minutes=5)
first.timestamp = event_creation_timestamp_first
2018-08-21 00:32:27 +02:00
first.attributes[0].timestamp = event_creation_timestamp_first
2018-08-10 19:04:02 +02:00
# Creating event 2 - timestamp 2 min ago
2018-08-21 00:32:27 +02:00
second = self.create_simple_event(force_timestamps=True)
2018-08-10 19:04:02 +02:00
event_creation_timestamp_second = datetime.now() - timedelta(minutes=2)
second.timestamp = event_creation_timestamp_second
2018-08-21 00:32:27 +02:00
second.attributes[0].timestamp = event_creation_timestamp_second
2018-08-10 19:04:02 +02:00
try:
2018-08-21 00:32:27 +02:00
first = self.user_misp_connector.add_event(first)
second = self.user_misp_connector.add_event(second)
2018-08-10 19:04:02 +02:00
# Search as user
# # Test - last 4 min
attributes = self.user_misp_connector.search(controller='attributes', timestamp='4m')
2018-08-21 11:16:51 +02:00
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()))
2018-08-10 19:04:02 +02:00
2018-08-19 14:35:32 +02:00
# # Test timestamp of 2nd event
attributes = self.user_misp_connector.search(controller='attributes', timestamp=event_creation_timestamp_second.timestamp())
2018-08-21 11:16:51 +02:00
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()))
2018-08-10 19:04:02 +02:00
# # Test interval -6 min -> -4 min
attributes = self.user_misp_connector.search(controller='attributes', timestamp=['6m', '4m'])
2018-08-21 11:16:51 +02:00
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()))
2018-08-10 19:04:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
2018-08-10 19:04:02 +02:00
def test_user_perms(self):
'''Test publish rights'''
2018-08-10 19:04:02 +02:00
try:
2018-08-21 11:16:51 +02:00
first = self.create_simple_event()
2018-08-21 00:32:27 +02:00
first.publish()
2018-08-10 19:04:02 +02:00
# Add event as user, no publish rights
2018-08-21 00:32:27 +02:00
first = self.user_misp_connector.add_event(first)
self.assertFalse(first.published)
2018-08-10 19:04:02 +02:00
# Add event as publisher
2018-08-21 00:32:27 +02:00
first.publish()
first = self.pub_misp_connector.update_event(first, pythonify=True)
2018-08-21 00:32:27 +02:00
self.assertTrue(first.published)
2018-08-10 19:04:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-08-09 17:58:55 +02:00
def test_delete_by_uuid(self):
try:
first = self.create_simple_event()
obj = MISPObject('file')
obj.add_attribute('filename', 'foo')
first.add_object(obj)
first = self.user_misp_connector.add_event(first)
2019-08-27 10:43:58 +02:00
r = self.user_misp_connector.delete_attribute(first.attributes[0].uuid)
self.assertEqual(r['message'], 'Attribute deleted.')
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.delete_object(first.objects[0].uuid)
self.assertEqual(r['message'], 'Object deleted')
r = self.user_misp_connector.delete_event(first.uuid)
self.assertEqual(r['message'], 'Event deleted.')
2019-08-09 17:58:55 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2018-08-10 19:04:02 +02:00
2018-08-21 00:32:27 +02:00
def test_search_publish_timestamp(self):
'''Search for a specific publication timestamp, an interval, and invalid values.'''
2018-08-10 19:04:02 +02:00
# Creating event 1
2018-08-21 00:32:27 +02:00
first = self.create_simple_event()
2018-08-10 19:04:02 +02:00
first.publish()
# Creating event 2
2018-08-21 00:32:27 +02:00
second = self.create_simple_event()
2018-08-10 19:04:02 +02:00
second.publish()
try:
first = self.pub_misp_connector.add_event(first, pythonify=True)
2018-08-21 00:32:27 +02:00
time.sleep(10)
second = self.pub_misp_connector.add_event(second, pythonify=True)
2018-08-10 19:04:02 +02:00
# Test invalid query
events = self.pub_misp_connector.search(publish_timestamp='5x', pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
events = self.pub_misp_connector.search(publish_timestamp='ad', pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
events = self.pub_misp_connector.search(publish_timestamp='aaad', pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
2018-08-21 00:32:27 +02:00
# Test - last 4 min
events = self.pub_misp_connector.search(publish_timestamp='5s', pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
2018-08-10 19:04:02 +02:00
2018-08-21 00:32:27 +02:00
# Test 5 sec before timestamp of 2nd event
events = self.pub_misp_connector.search(publish_timestamp=(second.publish_timestamp.timestamp()), pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
2018-08-10 19:04:02 +02:00
2018-08-21 00:32:27 +02:00
# Test interval -6 min -> -4 min
2018-08-21 11:16:51 +02:00
events = self.pub_misp_connector.search(publish_timestamp=[first.publish_timestamp.timestamp() - 5,
second.publish_timestamp.timestamp() - 5], pythonify=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, first.id)
2018-08-10 19:04:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
2018-08-21 00:32:27 +02:00
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)'''
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
2019-08-16 10:48:06 +02:00
attribute = self.user_misp_connector.add_attribute(first, {'type': 'comment', 'value': 'bar'})
2019-07-12 17:35:02 +02:00
self.assertEqual(attribute.value, 'bar', attribute.to_json())
self.assertEqual(attribute.distribution, Distribution.inherit.value, attribute.to_json())
# Object - add
o = MISPObject('file')
o.add_attribute('filename', value='blah.exe')
2019-08-16 10:48:06 +02:00
new_obj = self.user_misp_connector.add_object(first, o)
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
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2018-08-21 00:32:27 +02:00
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
2018-09-20 02:22:37 +02:00
warning list
'''
2018-08-21 00:32:27 +02:00
first = self.create_simple_event()
first.info = 'foo bar blah'
2018-09-20 02:22:37 +02:00
# First has one text attribute
second = self.create_simple_event()
second.info = 'foo blah'
2019-09-11 14:20:55 +02:00
second.add_tag('tlp:amber___test')
second.set_date('2018-09-01')
second.add_attribute('ip-src', '8.8.8.8')
2018-09-20 02:22:37 +02:00
# second has two attributes: text and ip-src
2018-08-10 19:04:02 +02:00
try:
2018-08-21 00:32:27 +02:00
first = self.user_misp_connector.add_event(first)
second = self.user_misp_connector.add_event(second)
2018-08-21 00:32:27 +02:00
timeframe = [first.timestamp.timestamp() - 5, first.timestamp.timestamp() + 5]
2018-08-21 11:16:51 +02:00
# 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)
self.assertEqual(len(events), 2)
2018-08-21 11:16:51 +02:00
self.assertEqual(events[0].id, first.id)
self.assertEqual(events[1].id, second.id)
events = self.user_misp_connector.search(timestamp=timeframe, value='nothere')
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
events = self.user_misp_connector.search(timestamp=timeframe, value=first.attributes[0].value)
2018-08-21 11:16:51 +02:00
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)
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
2018-08-10 19:04:02 +02:00
# Test return content
events = self.user_misp_connector.search(timestamp=timeframe, metadata=False)
self.assertEqual(len(events), 2)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events[0].attributes), 1)
self.assertEqual(len(events[1].attributes), 2)
events = self.user_misp_connector.search(timestamp=timeframe, metadata=True)
self.assertEqual(len(events), 2)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events[0].attributes), 0)
self.assertEqual(len(events[1].attributes), 0)
2018-08-10 19:04:02 +02:00
# other things
events = self.user_misp_connector.search(timestamp=timeframe, published=True)
2018-08-21 11:16:51 +02:00
self.assertEqual(events, [])
events = self.user_misp_connector.search(timestamp=timeframe, published=False)
self.assertEqual(len(events), 2)
events = self.user_misp_connector.search(eventid=first.id)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, first.id)
events = self.user_misp_connector.search(uuid=first.uuid)
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, first.id)
events = self.user_misp_connector.search(org=first.orgc_id)
self.assertEqual(len(events), 2)
# test like search
events = self.user_misp_connector.search(timestamp=timeframe, value='%{}%'.format(first.attributes[0].value.split('-')[2]))
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, first.id)
events = self.user_misp_connector.search(timestamp=timeframe, eventinfo='%bar blah%')
2018-09-28 16:36:56 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, first.id)
# quickfilter
events = self.user_misp_connector.search(timestamp=timeframe,
2018-11-29 18:14:10 +01:00
quickfilter='%foo blah%', pythonify=True)
# FIXME: should return one event
# print(events)
# self.assertEqual(len(events), 1)
# self.assertEqual(events[0].id, second.id)
2018-09-20 02:22:37 +02:00
# date_from / date_to
events = self.user_misp_connector.search(timestamp=timeframe, date_from=date.today().isoformat())
2018-08-21 11:16:51 +02:00
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')
self.assertEqual(len(events), 2)
events = self.user_misp_connector.search(timestamp=timeframe, date_from='2018-09-01', date_to='2018-09-02')
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
# Category
events = self.user_misp_connector.search(timestamp=timeframe, category='Network activity')
2018-08-21 11:16:51 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
# toids
events = self.user_misp_connector.search(timestamp=timeframe, to_ids='0')
self.assertEqual(len(events), 2)
events = self.user_misp_connector.search(timestamp=timeframe, to_ids='1')
2018-09-28 16:36:56 +02:00
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)
self.assertEqual(len(events[0].attributes), 1)
events = self.user_misp_connector.search(eventid=second.id, deleted=True)
self.assertEqual(len(events[0].attributes), 1)
# include_event_uuid
attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, include_event_uuid=True)
2018-09-21 22:33:22 +02:00
self.assertEqual(attributes[0].event_uuid, second.uuid)
2019-09-11 14:20:55 +02:00
# include_event_tags
attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, include_event_tags=True)
self.assertEqual(attributes[0].tags[0].name, 'tlp:amber___test')
# event_timestamp
2018-11-02 09:01:29 +01:00
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())
2018-09-28 16:36:56 +02:00
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)
self.assertEqual(len(events), 1)
# warninglist
2018-09-25 15:32:17 +02:00
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')
2018-09-25 15:32:17 +02:00
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)
2018-09-25 15:32:17 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
self.assertEqual(len(events[0].attributes), 5)
2018-09-25 15:32:17 +02:00
events = self.user_misp_connector.search(eventid=second.id, enforce_warninglist=False)
2018-09-25 15:32:17 +02:00
self.assertEqual(len(events), 1)
self.assertEqual(events[0].id, second.id)
self.assertEqual(len(events[0].attributes), 5)
2018-09-25 15:32:17 +02:00
events = self.user_misp_connector.search(eventid=second.id, enforce_warninglist=True)
2019-07-19 17:26:59 +02:00
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'})
2018-10-22 00:49:38 +02:00
# Page / limit
attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, page=1, limit=3)
2018-10-22 00:49:38 +02:00
self.assertEqual(len(attributes), 3)
attributes = self.user_misp_connector.search(controller='attributes', eventid=second.id, page=2, limit=3)
self.assertEqual(len(attributes), 2)
2018-10-22 00:49:38 +02:00
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)
2018-09-20 02:22:37 +02:00
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)
2018-08-10 19:04:02 +02:00
2019-07-12 17:35:02 +02:00
# Search index
events = self.user_misp_connector.search_index(timestamp=first.timestamp.timestamp(),
pythonify=True)
self.assertEqual(len(events), 1)
self.assertEqual(events[0].info, 'foo bar blah')
self.assertEqual(events[0].attributes, [])
2019-08-28 16:02:20 +02:00
# Contact reporter
r = self.user_misp_connector.contact_event_reporter(events[0].id, 'This is a test')
self.assertEqual(r['message'], 'Email sent to the reporter.')
2018-08-10 19:04:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
2018-08-10 19:04:02 +02:00
2018-08-17 15:09:17 +02:00
def test_edit_attribute(self):
2018-08-21 00:32:27 +02:00
first = self.create_simple_event()
2018-08-17 15:09:17 +02:00
try:
first.attributes[0].comment = 'This is the original comment'
2018-08-21 00:32:27 +02:00
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])
2019-08-16 10:48:06 +02:00
self.assertTrue(isinstance(attribute, MISPAttribute), attribute)
2018-08-21 00:32:27 +02:00
self.assertEqual(attribute.comment, 'This is the modified comment')
2019-08-16 10:48:06 +02:00
attribute = self.user_misp_connector.update_attribute({'comment': 'This is the modified comment, again'}, attribute)
self.assertTrue(isinstance(attribute, MISPAttribute), attribute)
self.assertEqual(attribute.comment, 'This is the modified comment, again', attribute)
attribute = self.user_misp_connector.update_attribute({'disable_correlation': True}, attribute)
self.assertTrue(attribute.disable_correlation, attribute)
attribute = self.user_misp_connector.update_attribute({'disable_correlation': False}, attribute)
self.assertFalse(attribute.disable_correlation, attribute)
2018-08-17 15:09:17 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2018-08-17 15:09:17 +02:00
def test_sightings(self):
first = self.create_simple_event()
2018-09-20 02:22:37 +02:00
second = self.create_simple_event()
try:
first = self.user_misp_connector.add_event(first)
2018-09-20 02:22:37 +02:00
second = self.user_misp_connector.add_event(second)
current_ts = int(time.time())
time.sleep(5)
2019-07-12 17:35:02 +02:00
r = self.user_misp_connector.add_sighting({'value': first.attributes[0].value})
2019-08-20 15:34:21 +02:00
self.assertEqual(int(r.attribute_id), first.attributes[0].id)
2019-07-12 17:35:02 +02:00
s = MISPSighting()
s.value = second.attributes[0].value
s.source = 'Testcases'
s.type = '1'
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.add_sighting(s, second.attributes[0])
2019-08-20 15:34:21 +02:00
self.assertEqual(r.source, 'Testcases')
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)
2019-07-12 17:35:02 +02:00
self.assertEqual(s[0]['event'].id, second.id, s)
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))
2019-07-12 17:35:02 +02:00
# Get sightings from event/attribute / org
s = self.user_misp_connector.sightings(first)
2019-07-12 17:35:02 +02:00
self.assertTrue(isinstance(s, list))
self.assertEqual(int(s[0].attribute_id), first.attributes[0].id)
2019-08-20 15:34:21 +02:00
self.admin_misp_connector.add_sighting(s, second.attributes[0])
s = self.user_misp_connector.sightings(second.attributes[0])
2019-07-12 17:35:02 +02:00
self.assertEqual(len(s), 2)
2019-08-16 10:48:06 +02:00
s = self.user_misp_connector.sightings(second.attributes[0], self.test_org)
2019-07-12 17:35:02 +02:00
self.assertEqual(len(s), 1)
self.assertEqual(s[0].org_id, self.test_org.id)
# Delete sighting
r = self.user_misp_connector.delete_sighting(s[0])
self.assertEqual(r['message'], 'Sighting successfuly deleted.')
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
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)
2019-08-16 10:48:06 +02:00
response = self.user_misp_connector.publish(first, alert=False)
2018-12-26 17:38:19 +01:00
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)
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.publish(first, alert=False)
csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp())
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)
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')
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')
self.assertEqual(len(csv), 0)
# type_attribute
csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), type_attribute='text')
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')
self.assertEqual(len(csv), 0)
# context
csv = self.user_misp_connector.search(return_format='csv', publish_timestamp=first.timestamp.timestamp(), include_context=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())
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')
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)
# 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)
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)
self.assertEqual(len(csv[0].keys()), 2)
for k in columns:
self.assertTrue(k in csv[0])
finally:
2019-08-05 16:53:32 +02:00
# Mostly solved -> https://github.com/MISP/MISP/issues/4886
time.sleep(5)
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
2019-02-01 11:07:42 +01:00
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)
2019-07-19 17:26:59 +02:00
stix = self.user_misp_connector.search(return_format='stix', eventid=first.id)
self.assertTrue(stix['related_packages'][0]['package']['incidents'][0]['related_indicators']['indicators'][0]['indicator']['observable']['object']['properties']['address_value']['value'], '8.8.8.8')
2019-07-19 17:26:59 +02:00
stix2 = self.user_misp_connector.search(return_format='stix2', eventid=first.id)
2020-01-17 11:34:40 +01:00
print(json.dumps(stix2, indent=2))
2019-07-19 17:26:59 +02:00
self.assertEqual(stix2['objects'][-1]['pattern'], "[network-traffic:src_ref.type = 'ipv4-addr' AND network-traffic:src_ref.value = '8.8.8.8']")
2019-02-01 11:07:42 +01:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-02-01 11:07:42 +01:00
2019-03-11 09:39:08 +01:00
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
2019-03-11 09:39:08 +01:00
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')
2019-03-11 09:39:08 +01:00
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')
2019-07-12 17:35:02 +02:00
ip_dom.add_attribute('domain', value='google.fr', disable_correlation=True)
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)
2019-07-12 17:35:02 +02:00
self.assertTrue(second.objects[0].attributes[0].disable_correlation)
self.assertFalse(second.objects[0].attributes[1].to_ids)
self.assertTrue(second.objects[0].attributes[2].to_ids)
2019-07-12 17:35:02 +02:00
# Test generic Tag methods
r = self.admin_misp_connector.tag(second, 'generic_tag_test')
self.assertTrue(r['message'].endswith(f'successfully attached to Event({second.id}).'), r['message'])
r = self.admin_misp_connector.untag(second, 'generic_tag_test')
self.assertTrue(r['message'].endswith(f'successfully removed from Event({second.id}).'), r['message'])
# NOTE: object tagging not supported yet
# r = self.admin_misp_connector.tag(second.objects[0].uuid, 'generic_tag_test')
# self.assertTrue(r['message'].endswith(f'successfully attached to Object({second.objects[0].id}).'), r['message'])
# r = self.admin_misp_connector.untag(second.objects[0].uuid, 'generic_tag_test')
# self.assertTrue(r['message'].endswith(f'successfully removed from Object({second.objects[0].id}).'), r['message'])
r = self.admin_misp_connector.tag(second.objects[0].attributes[0].uuid, 'generic_tag_test')
self.assertTrue(r['message'].endswith(f'successfully attached to Attribute({second.objects[0].attributes[0].id}).'), r['message'])
r = self.admin_misp_connector.untag(second.objects[0].attributes[0].uuid, 'generic_tag_test')
self.assertTrue(r['message'].endswith(f'successfully removed from Attribute({second.objects[0].attributes[0].id}).'), r['message'])
# Delete tag to avoid polluting the db
tags = self.admin_misp_connector.tags(pythonify=True)
for t in tags:
if t.name == 'generic_tag_test':
2019-08-16 10:48:06 +02:00
response = self.admin_misp_connector.delete_tag(t)
2019-07-12 17:35:02 +02:00
self.assertEqual(response['message'], 'Tag deleted.')
# Test delete object
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.delete_object(second.objects[0])
2019-07-12 17:35:02 +02:00
self.assertEqual(r['message'], 'Object deleted')
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
def test_custom_template(self):
first = self.create_simple_event()
try:
with open('tests/viper-test-files/test_files/whoami.exe', 'rb') as f:
first.add_attribute('malware-sample', value='whoami.exe', data=BytesIO(f.read()), expand='binary')
first.run_expansions()
first = self.admin_misp_connector.add_event(first, pythonify=True)
self.assertEqual(len(first.objects), 7)
file_object = first.get_objects_by_name('file')[0]
file_object.force_misp_objects_path_custom('tests/mispevent_testfiles', 'overwrite_file')
file_object.add_attribute('test_overwrite', 'blah')
2019-12-16 13:50:29 +01:00
obj_json = self.admin_misp_connector.update_object(file_object)
self.assertTrue('Object' in obj_json, obj_json)
self.assertTrue('name' in obj_json['Object'], obj_json)
obj = MISPObject(obj_json['Object']['name'])
obj.from_dict(**obj_json)
self.assertEqual(obj.get_attributes_by_relation('test_overwrite')[0].value, 'blah')
2019-12-16 15:24:04 +01:00
# FULL object add & update with custom template
new_object = MISPObject('overwrite_file', misp_objects_path_custom='tests/mispevent_testfiles')
new_object.add_attribute('test_overwrite', 'barbaz')
new_object.add_attribute('filename', 'barbaz.exe')
new_object = self.admin_misp_connector.add_object(first, new_object, pythonify=True)
self.assertEqual(new_object.get_attributes_by_relation('test_overwrite')[0].value, 'barbaz', new_object)
new_object.force_misp_objects_path_custom('tests/mispevent_testfiles', 'overwrite_file')
new_object.add_attribute('filename', 'foobar.exe')
new_object = self.admin_misp_connector.update_object(new_object, pythonify=True)
self.assertEqual(new_object.get_attributes_by_relation('filename')[1].value, 'foobar.exe', new_object)
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-07-12 17:35:02 +02:00
def test_unknown_template(self):
first = self.create_simple_event()
attributeAsDict = [{'MyCoolAttribute': {'value': 'critical thing', 'type': 'text'}},
{'MyCoolerAttribute': {'value': 'even worse', 'type': 'text', 'disable_correlation': True}}]
misp_object = GenericObjectGenerator('my-cool-template')
misp_object.generate_attributes(attributeAsDict)
first.add_object(misp_object)
blah_object = MISPObject('BLAH_TEST')
blah_object.add_reference(misp_object.uuid, "test relation")
blah_object.add_attribute('transaction-number', value='foo', type="text", disable_correlation=True)
first.add_object(blah_object)
try:
first = self.user_misp_connector.add_event(first)
self.assertEqual(len(first.objects[0].attributes), 2)
self.assertFalse(first.objects[0].attributes[0].disable_correlation)
self.assertTrue(first.objects[0].attributes[1].disable_correlation)
self.assertTrue(first.objects[1].attributes[0].disable_correlation)
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-07-12 17:35:02 +02:00
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)
2019-03-11 09:39:08 +01:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-03-11 09:39:08 +01:00
2019-04-11 09:47:57 +02:00
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
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-04-11 09:47:57 +02:00
def test_object_template(self):
2018-09-29 21:44:02 +02:00
r = self.admin_misp_connector.update_object_templates()
self.assertEqual(type(r), list)
2019-07-12 17:35:02 +02:00
object_templates = self.admin_misp_connector.object_templates(pythonify=True)
self.assertTrue(isinstance(object_templates, list))
for object_template in object_templates:
if object_template.name == 'file':
break
template = self.admin_misp_connector.get_object_template(object_template.uuid, pythonify=True)
self.assertEqual(template.name, 'file')
2018-10-05 17:45:12 +02:00
def test_tags(self):
# Get list
2019-07-12 17:35:02 +02:00
tags = self.admin_misp_connector.tags(pythonify=True)
2018-10-05 17:45:12 +02:00
self.assertTrue(isinstance(tags, list))
# Get tag
for tag in tags:
2019-07-12 17:35:02 +02:00
if not tag.hide_tag:
2018-10-05 17:45:12 +02:00
break
2019-08-16 10:48:06 +02:00
tag = self.admin_misp_connector.get_tag(tag, pythonify=True)
2018-10-05 17:45:12 +02:00
self.assertTrue('name' in tag)
2019-07-12 17:35:02 +02:00
# Enable by MISPTag
tag = self.admin_misp_connector.disable_tag(tag, pythonify=True)
self.assertTrue(tag.hide_tag)
tag = self.admin_misp_connector.enable_tag(tag, pythonify=True)
self.assertFalse(tag.hide_tag)
# Add tag
tag = MISPTag()
tag.name = 'this is a test tag'
new_tag = self.admin_misp_connector.add_tag(tag, pythonify=True)
self.assertEqual(new_tag.name, tag.name)
2019-07-19 10:57:15 +02:00
# Add non-exportable tag
tag = MISPTag()
tag.name = 'non-exportable tag'
tag.exportable = False
non_exportable_tag = self.admin_misp_connector.add_tag(tag, pythonify=True)
self.assertFalse(non_exportable_tag.exportable)
first = self.create_simple_event()
first.attributes[0].add_tag('non-exportable tag')
# Add tag restricted to an org
tag = MISPTag()
tag.name = f'restricted to org {self.test_org.id}'
tag.org_id = self.test_org.id
tag_org_restricted = self.admin_misp_connector.add_tag(tag, pythonify=True)
self.assertEqual(tag_org_restricted.org_id, tag.org_id)
# Add tag restricted to a user
tag.name = f'restricted to user {self.test_usr.id}'
tag.user_id = self.test_usr.id
tag_user_restricted = self.admin_misp_connector.add_tag(tag, pythonify=True)
self.assertEqual(tag_user_restricted.user_id, tag.user_id)
2019-07-19 10:57:15 +02:00
try:
first = self.user_misp_connector.add_event(first)
2019-07-19 10:57:15 +02:00
self.assertFalse(first.attributes[0].tags)
first = self.admin_misp_connector.get_event(first, pythonify=True)
# Reference: https://github.com/MISP/MISP/issues/1394
self.assertFalse(first.attributes[0].tags)
# Reference: https://github.com/MISP/PyMISP/issues/483
r = self.delegate_user_misp_connector.tag(first, tag_org_restricted)
2019-11-29 16:33:17 +01:00
# FIXME: The error message changed and is unhelpful.
# self.assertEqual(r['errors'][1]['message'], 'Invalid Tag. This tag can only be set by a fixed organisation.')
self.assertEqual(r['errors'][1]['message'], 'Invalid Target.')
r = self.user_misp_connector.tag(first, tag_org_restricted)
self.assertEqual(r['name'], f'Global tag {tag_org_restricted.name}({tag_org_restricted.id}) successfully attached to Event({first.id}).')
r = self.pub_misp_connector.tag(first.attributes[0], tag_user_restricted)
self.assertEqual(r['errors'][1]['message'], 'Invalid Tag. This tag can only be set by a fixed user.')
r = self.user_misp_connector.tag(first.attributes[0], tag_user_restricted)
self.assertEqual(r['name'], f'Global tag {tag_user_restricted.name}({tag_user_restricted.id}) successfully attached to Attribute({first.attributes[0].id}).')
2019-07-19 10:57:15 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-07-19 10:57:15 +02:00
2019-07-12 17:35:02 +02:00
# Delete tag
2019-08-16 10:48:06 +02:00
response = self.admin_misp_connector.delete_tag(new_tag)
2019-07-12 17:35:02 +02:00
self.assertEqual(response['message'], 'Tag deleted.')
2019-08-16 10:48:06 +02:00
response = self.admin_misp_connector.delete_tag(non_exportable_tag)
2019-07-19 10:57:15 +02:00
self.assertEqual(response['message'], 'Tag deleted.')
response = self.admin_misp_connector.delete_tag(tag_org_restricted)
response = self.admin_misp_connector.delete_tag(tag_user_restricted)
2018-10-05 17:45:12 +02:00
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:
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.add_object(first, s)
self.assertEqual(r.name, 'pe-section', r)
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.add_object(first, peo)
self.assertEqual(r.name, 'pe', r)
for ref in peo.ObjectReference:
r = self.user_misp_connector.add_object_reference(ref)
2019-08-05 16:53:32 +02:00
self.assertEqual(r.object_uuid, peo.uuid, r.to_json())
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.add_object(first, fo)
2019-05-08 12:16:43 +02:00
obj_attrs = r.get_attributes_by_relation('ssdeep')
self.assertEqual(len(obj_attrs), 1, obj_attrs)
self.assertEqual(r.name, 'file', r)
r = self.user_misp_connector.add_object_reference(fo.ObjectReference[0])
2019-08-05 16:53:32 +02:00
self.assertEqual(r.object_uuid, fo.uuid, r.to_json())
2019-07-12 17:35:02 +02:00
self.assertEqual(r.referenced_uuid, peo.uuid, r.to_json())
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.delete_object_reference(r)
2019-07-12 17:35:02 +02:00
self.assertEqual(r['message'], 'ObjectReference deleted')
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
def test_add_event_with_attachment(self):
first = self.create_simple_event()
try:
first = self.user_misp_connector.add_event(first)
2018-12-26 19:41:59 +01:00
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, 'includes')
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, 'includes')
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2018-10-05 17:45:12 +02:00
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
2019-07-12 17:35:02 +02:00
taxonomies = self.admin_misp_connector.taxonomies(pythonify=True)
2018-10-05 17:45:12 +02:00
self.assertTrue(isinstance(taxonomies, list))
list_name_test = 'tlp'
for tax in taxonomies:
2019-07-12 17:35:02 +02:00
if tax.namespace == list_name_test:
2018-10-05 17:45:12 +02:00
break
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.get_taxonomy(tax, pythonify=True)
2019-07-19 17:26:59 +02:00
self.assertEqual(r.namespace, list_name_test)
self.assertTrue('enabled' in r)
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.enable_taxonomy(tax)
2018-10-05 17:45:12 +02:00
self.assertEqual(r['message'], 'Taxonomy enabled')
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.enable_taxonomy_tags(tax)
2019-08-01 13:19:21 +02:00
self.assertEqual(r['name'], 'The tag(s) has been saved.')
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.disable_taxonomy(tax)
2018-10-05 17:45:12 +02:00
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)
2018-10-05 17:45:12 +02:00
# Get list
2019-07-12 17:35:02 +02:00
warninglists = self.admin_misp_connector.warninglists(pythonify=True)
2018-10-05 17:45:12 +02:00
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:
2019-07-12 17:35:02 +02:00
if wl.name == list_name_test:
2018-10-05 17:45:12 +02:00
break
2019-07-12 17:35:02 +02:00
testwl = wl
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.get_warninglist(testwl, pythonify=True)
2019-07-12 17:35:02 +02:00
self.assertEqual(r.name, list_name_test)
self.assertTrue('WarninglistEntry' in r)
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.enable_warninglist(testwl)
2018-10-05 17:45:12 +02:00
self.assertEqual(r['success'], '1 warninglist(s) enabled')
2019-07-12 17:35:02 +02:00
# Check if a value is in a warning list
md5_empty_file = 'd41d8cd98f00b204e9800998ecf8427e'
r = self.user_misp_connector.values_in_warninglist([md5_empty_file])
self.assertEqual(r[md5_empty_file][0]['name'], list_name_test)
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.disable_warninglist(testwl)
2018-10-05 17:45:12 +02:00
self.assertEqual(r['success'], '1 warninglist(s) disabled')
def test_noticelists(self):
# Make sure we're up-to-date
2018-09-29 21:44:02 +02:00
r = self.admin_misp_connector.update_noticelists()
self.assertEqual(r['name'], 'All noticelists are up to date already.')
2018-10-05 17:45:12 +02:00
# Get list
2019-07-12 17:35:02 +02:00
noticelists = self.admin_misp_connector.noticelists(pythonify=True)
2018-10-05 17:45:12 +02:00
self.assertTrue(isinstance(noticelists, list))
list_name_test = 'gdpr'
for nl in noticelists:
2019-07-12 17:35:02 +02:00
if nl.name == list_name_test:
2018-10-05 17:45:12 +02:00
break
testnl = nl
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.get_noticelist(testnl, pythonify=True)
2019-07-12 17:35:02 +02:00
self.assertEqual(r.name, list_name_test)
# FIXME: https://github.com/MISP/MISP/issues/4856
self.assertTrue('NoticelistEntry' in r)
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.enable_noticelist(testnl)
2019-07-12 17:35:02 +02:00
self.assertTrue(r['Noticelist']['enabled'], r)
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.disable_noticelist(testnl)
2019-07-12 17:35:02 +02:00
self.assertFalse(r['Noticelist']['enabled'], r)
2018-10-05 17:45:12 +02:00
def test_galaxies(self):
2019-07-19 17:26:59 +02:00
# 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.galaxies(pythonify=True)
self.assertTrue(isinstance(galaxies, list))
list_name_test = 'Mobile Attack - Attack Pattern'
for galaxy in galaxies:
if galaxy.name == list_name_test:
break
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.get_galaxy(galaxy, pythonify=True)
2019-07-19 17:26:59 +02:00
self.assertEqual(r.name, list_name_test)
# FIXME: Fails due to https://github.com/MISP/MISP/issues/4855
# self.assertTrue('GalaxyCluster' in r)
2018-09-29 21:44:02 +02:00
def test_zmq(self):
first = self.create_simple_event()
try:
first = self.user_misp_connector.add_event(first)
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.push_event_to_ZMQ(first)
2019-07-19 17:26:59 +02:00
self.assertEqual(r['message'], 'Event published to ZMQ')
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
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():
2019-08-16 10:48:06 +02:00
new_object = self.user_misp_connector.add_object(first, o)
self.assertEqual(len(new_object.attributes), 3)
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
def test_user(self):
2019-07-12 17:35:02 +02:00
# Get list
users = self.admin_misp_connector.users(pythonify=True)
self.assertTrue(isinstance(users, list))
users_email = 'testusr@user.local'
for user in users:
if user.email == users_email:
break
2019-09-11 14:20:55 +02:00
else:
raise Exception('Unable to find that user')
2019-07-12 17:35:02 +02:00
self.assertEqual(user.email, users_email)
# get user
user = self.user_misp_connector.get_user(pythonify=True)
self.assertEqual(user.authkey, self.test_usr.authkey)
2019-07-12 17:35:02 +02:00
# Update user
user.email = 'foo@bar.de'
user = self.admin_misp_connector.update_user(user, pythonify=True)
self.assertEqual(user.email, 'foo@bar.de')
def test_organisation(self):
# Get list
orgs = self.admin_misp_connector.organisations(pythonify=True)
self.assertTrue(isinstance(orgs, list))
org_name = 'ORGNAME'
for org in orgs:
if org.name == org_name:
break
self.assertEqual(org.name, org_name)
# Get org
organisation = self.user_misp_connector.get_organisation(self.test_usr.org_id)
self.assertEqual(organisation.name, 'Test Org')
# Update org
organisation.name = 'blah'
organisation = self.admin_misp_connector.update_organisation(organisation, pythonify=True)
self.assertEqual(organisation.name, 'blah', organisation)
def test_attribute(self):
first = self.create_simple_event()
2019-07-12 17:35:02 +02:00
second = self.create_simple_event()
a = second.add_attribute('ip-src', '11.11.11.11')
a.add_tag('testtag_admin_created')
2019-07-12 17:35:02 +02:00
second.distribution = Distribution.all_communities
try:
first = self.user_misp_connector.add_event(first)
2019-08-01 13:19:21 +02:00
second = self.admin_misp_connector.add_event(second, pythonify=True)
# Get attribute
2019-08-16 10:48:06 +02:00
attribute = self.user_misp_connector.get_attribute(first.attributes[0])
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'
2019-08-16 10:48:06 +02:00
new_attribute = self.user_misp_connector.add_attribute(first, new_attribute)
self.assertTrue(isinstance(new_attribute, MISPAttribute), new_attribute)
self.assertEqual(new_attribute.value, '1.2.3.4', new_attribute)
# Test attribute already in event
# new_attribute.uuid = str(uuid4())
2019-08-16 10:48:06 +02:00
# new_attribute = self.user_misp_connector.add_attribute(first, new_attribute)
new_similar = MISPAttribute()
new_similar.value = '1.2.3.4'
new_similar.type = 'ip-dst'
2019-08-16 10:48:06 +02:00
similar_error = self.user_misp_connector.add_attribute(first, new_similar)
self.assertEqual(similar_error['errors'][1]['errors']['value'][0], 'A similar attribute already exists for this event.')
# Test add multiple attributes at once
attr1 = MISPAttribute()
attr1.value = '1.2.3.4'
attr1.type = 'ip-dst'
attr2 = MISPAttribute()
attr2.value = '1.2.3.5'
attr2.type = 'ip-dst'
attr3 = MISPAttribute()
attr3.value = first.attributes[0].value
attr3.type = first.attributes[0].type
attr4 = MISPAttribute()
attr4.value = '1.2.3.6'
attr4.type = 'ip-dst'
attr4.add_tag('tlp:amber___test_unique_not_created')
attr4.add_tag('testtag_admin_created')
2019-08-16 10:48:06 +02:00
response = self.user_misp_connector.add_attribute(first, [attr1, attr2, attr3, attr4])
time.sleep(5)
2019-08-19 11:58:37 +02:00
self.assertTrue(isinstance(response['attributes'], list), response['attributes'])
2019-08-19 11:30:53 +02:00
self.assertEqual(response['attributes'][0].value, '1.2.3.5')
self.assertEqual(response['attributes'][1].value, '1.2.3.6')
2019-08-19 11:58:37 +02:00
self.assertTrue(isinstance(response['attributes'][1].tags, list), response['attributes'][1].to_json())
2019-08-19 12:11:54 +02:00
self.assertTrue(len(response['attributes'][1].tags), response['attributes'][1].to_json())
self.assertEqual(response['attributes'][1].tags[0].name, 'testtag_admin_created')
2019-08-19 11:30:53 +02:00
self.assertEqual(response['errors']['attribute_0']['value'][0], 'A similar attribute already exists for this event.')
self.assertEqual(response['errors']['attribute_2']['value'][0], 'A similar attribute already exists for this event.')
# 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)
2019-07-12 17:35:02 +02:00
# Delete attribute as proposal
2019-08-16 10:48:06 +02:00
proposal_delete = self.user_misp_connector.delete_attribute_proposal(new_attribute)
2019-07-12 17:35:02 +02:00
self.assertTrue(proposal_delete['saved'])
# Get attribute proposal
2019-08-16 10:48:06 +02:00
temp_new_proposal = self.user_misp_connector.get_attribute_proposal(new_proposal)
self.assertEqual(temp_new_proposal.uuid, new_proposal.uuid)
2019-08-01 13:19:21 +02:00
# Get attribute proposal*S*
proposals = self.user_misp_connector.attribute_proposals()
self.assertTrue(isinstance(proposals, list))
self.assertEqual(len(proposals), 3)
self.assertEqual(proposals[0].value, '5.2.3.4')
# Get proposals on a specific event
self.admin_misp_connector.add_attribute_proposal(second.id, {'type': 'ip-src', 'value': '123.123.123.1'})
proposals = self.admin_misp_connector.attribute_proposals(pythonify=True)
self.assertTrue(isinstance(proposals, list))
self.assertEqual(len(proposals), 4)
proposals = self.admin_misp_connector.attribute_proposals(second, pythonify=True)
self.assertTrue(isinstance(proposals, list))
self.assertEqual(len(proposals), 1)
self.assertEqual(proposals[0].value, '123.123.123.1')
# Accept attribute proposal - New attribute
2019-08-16 10:48:06 +02:00
self.user_misp_connector.accept_attribute_proposal(new_proposal)
first = self.user_misp_connector.get_event(first)
self.assertEqual(first.attributes[-1].value, '5.2.3.4')
# Accept attribute proposal - Attribute update
2019-08-16 10:48:06 +02:00
response = self.user_misp_connector.accept_attribute_proposal(new_proposal_update)
self.assertEqual(response['message'], 'Proposed change accepted.')
2019-08-16 10:48:06 +02:00
attribute = self.user_misp_connector.get_attribute(new_attribute)
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})
2019-08-16 10:48:06 +02:00
response = self.user_misp_connector.discard_attribute_proposal(new_proposal_update)
self.assertEqual(response['message'], 'Proposal discarded.')
2019-08-16 10:48:06 +02:00
attribute = self.user_misp_connector.get_attribute(new_attribute)
self.assertEqual(attribute.to_ids, False)
2019-07-12 17:35:02 +02:00
# Test fallback to proposal if the user doesn't own the event
prop_attr = MISPAttribute()
prop_attr.from_dict(**{'type': 'ip-dst', 'value': '123.43.32.21'})
# Add attribute on event owned by someone else
2019-07-12 17:35:02 +02:00
attribute = self.user_misp_connector.add_attribute(second.id, prop_attr)
2019-08-16 10:48:06 +02:00
self.assertTrue(isinstance(attribute, MISPShadowAttribute), attribute)
2019-08-01 13:19:21 +02:00
# Test if add proposal without category works - https://github.com/MISP/MISP/issues/4868
attribute = self.user_misp_connector.add_attribute(second.id, {'type': 'ip-dst', 'value': '123.43.32.22'})
self.assertTrue(isinstance(attribute, MISPShadowAttribute))
# Add attribute with the same value as an existing proposal
prop_attr.uuid = str(uuid4())
2019-08-16 10:48:06 +02:00
attribute = self.admin_misp_connector.add_attribute(second, prop_attr, pythonify=True)
prop_attr.uuid = str(uuid4())
# Add a duplicate attribute (same value)
2019-08-16 10:48:06 +02:00
attribute = self.admin_misp_connector.add_attribute(second, prop_attr, pythonify=True)
self.assertTrue('errors' in attribute)
# Update attribute owned by someone else
2019-07-12 17:35:02 +02:00
attribute = self.user_misp_connector.update_attribute({'comment': 'blah'}, second.attributes[0].id)
2019-08-16 10:48:06 +02:00
self.assertTrue(isinstance(attribute, MISPShadowAttribute), attribute)
2019-07-12 17:35:02 +02:00
self.assertEqual(attribute.value, second.attributes[0].value)
second = self.admin_misp_connector.get_event(second, pythonify=True)
self.assertEqual(len(second.attributes), 3)
# Delete attribute owned by someone else
2019-08-27 10:43:58 +02:00
response = self.user_misp_connector.delete_attribute(second.attributes[1])
self.assertTrue(response['success'])
# Delete attribute owned by user
2019-08-16 10:48:06 +02:00
response = self.admin_misp_connector.delete_attribute(second.attributes[1])
2019-07-12 17:35:02 +02:00
self.assertEqual(response['message'], 'Attribute deleted.')
# Hard delete
response = self.admin_misp_connector.delete_attribute(second.attributes[0], hard=True)
self.assertEqual(response['message'], 'Attribute deleted.')
new_second = self.admin_misp_connector.get_event(second, deleted=[0, 1], pythonify=True)
self.assertEqual(len(new_second.attributes), 2)
2019-08-01 13:19:21 +02:00
# Test attribute*S*
attributes = self.admin_misp_connector.attributes()
self.assertEqual(len(attributes), 6)
2019-08-01 13:19:21 +02:00
# attributes = self.user_misp_connector.attributes()
# self.assertEqual(len(attributes), 5)
# Test event*S*
events = self.admin_misp_connector.events()
self.assertEqual(len(events), 2)
events = self.user_misp_connector.events()
self.assertEqual(len(events), 2)
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
def test_search_type_event_csv(self):
try:
first, second, third = self.environment()
# Search as admin
2019-07-12 17:35:02 +02:00
events = self.admin_misp_connector.search(return_format='csv', timestamp=first.timestamp.timestamp(), pythonify=True)
self.assertTrue(isinstance(events, list))
self.assertEqual(len(events), 8)
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(),
2019-07-12 17:35:02 +02:00
type_attribute=attributes_types_search, pythonify=True)
self.assertTrue(isinstance(events, list))
self.assertEqual(len(events), 6)
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2019-07-12 17:35:02 +02:00
def test_search_logs(self):
# FIXME: https://github.com/MISP/MISP/issues/4872
r = self.admin_misp_connector.update_user({'email': 'testusr-changed@user.local'}, self.test_usr)
2019-07-12 17:35:02 +02:00
r = self.admin_misp_connector.search_logs(model='User', created=date.today(), pythonify=True)
for entry in r[-1:]:
self.assertEqual(entry.action, 'edit')
r = self.admin_misp_connector.search_logs(email='admin@admin.test', created=date.today(), pythonify=True)
for entry in r[-1:]:
self.assertEqual(entry.action, 'edit')
r = self.admin_misp_connector.update_user({'email': 'testusr@user.local'}, self.test_usr)
2019-07-12 17:35:02 +02:00
def test_db_schema(self):
diag = self.admin_misp_connector.db_schema_diagnostic()
self.assertEqual(diag['actual_db_version'], diag['expected_db_version'], diag)
2019-03-25 15:41:11 +01:00
def test_live_acl(self):
missing_acls = self.admin_misp_connector.remote_acl()
2019-03-25 16:12:01 +01:00
self.assertEqual(missing_acls, [], msg=missing_acls)
2019-03-25 15:41:11 +01:00
2019-04-04 16:39:17 +02:00
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)
2019-07-12 17:35:02 +02:00
roles = self.admin_misp_connector.roles(pythonify=True)
self.assertTrue(isinstance(roles, list))
def test_describe_types(self):
remote = self.admin_misp_connector.describe_types_remote
remote_types = remote.pop('types')
remote_categories = remote.pop('categories')
remote_category_type_mappings = remote.pop('category_type_mappings')
local = dict(self.admin_misp_connector.describe_types_local)
local_types = local.pop('types')
local_categories = local.pop('categories')
local_category_type_mappings = local.pop('category_type_mappings')
2019-07-12 17:35:02 +02:00
self.assertDictEqual(remote, local)
self.assertEqual(sorted(remote_types), sorted(local_types))
self.assertEqual(sorted(remote_categories), sorted(local_categories))
for category, mapping in remote_category_type_mappings.items():
self.assertEqual(sorted(local_category_type_mappings[category]), sorted(mapping))
2019-07-12 17:35:02 +02:00
def test_versions(self):
self.assertEqual(self.user_misp_connector.version, self.user_misp_connector.pymisp_version_master)
self.assertEqual(self.user_misp_connector.misp_instance_version['version'],
self.user_misp_connector.misp_instance_version_master['version'])
def test_statistics(self):
try:
# Attributes
first, second, third = self.environment()
expected_attr_stats = {'ip-dst': '2', 'ip-src': '1', 'text': '5'}
attr_stats = self.admin_misp_connector.attributes_statistics()
self.assertDictEqual(attr_stats, expected_attr_stats)
expected_attr_stats_percent = {'ip-dst': '25%', 'ip-src': '12.5%', 'text': '62.5%'}
attr_stats = self.admin_misp_connector.attributes_statistics(percentage=True)
self.assertDictEqual(attr_stats, expected_attr_stats_percent)
expected_attr_stats_category_percent = {'Network activity': '37.5%', 'Other': '62.5%'}
attr_stats = self.admin_misp_connector.attributes_statistics(context='category', percentage=True)
self.assertDictEqual(attr_stats, expected_attr_stats_category_percent)
# Tags
to_test = {'tags': {'tlp:white___test': '1'}, 'taxonomies': []}
2019-07-12 17:35:02 +02:00
tags_stats = self.admin_misp_connector.tags_statistics()
self.assertDictEqual(tags_stats, to_test)
to_test = {'tags': {'tlp:white___test': '100%'}, 'taxonomies': []}
2019-07-12 17:35:02 +02:00
tags_stats = self.admin_misp_connector.tags_statistics(percentage=True, name_sort=True)
self.assertDictEqual(tags_stats, to_test)
# Users
users_stats = self.admin_misp_connector.users_statistics(context='data')
2019-07-21 02:08:33 +02:00
self.assertTrue('stats' in users_stats)
2019-07-12 17:35:02 +02:00
users_stats = self.admin_misp_connector.users_statistics(context='orgs')
self.assertTrue('ORGNAME' in list(users_stats.keys()))
users_stats = self.admin_misp_connector.users_statistics(context='users')
self.assertEqual(list(users_stats.keys()), ['user', 'org_local', 'org_external'])
users_stats = self.admin_misp_connector.users_statistics(context='tags')
self.assertEqual(list(users_stats.keys()), ['flatData', 'treemap'])
2019-08-01 13:19:21 +02:00
users_stats = self.admin_misp_connector.users_statistics(context='attributehistogram')
2019-08-27 10:43:58 +02:00
self.assertTrue(isinstance(users_stats, list), users_stats)
2019-07-12 17:35:02 +02:00
self.user_misp_connector.add_sighting({'value': first.attributes[0].value})
users_stats = self.user_misp_connector.users_statistics(context='sightings')
self.assertEqual(list(users_stats.keys()), ['toplist', 'eventids'])
# FIXME this one fails on travis.
# users_stats = self.admin_misp_connector.users_statistics(context='galaxyMatrix')
# self.assertTrue('matrix' in users_stats)
2019-07-12 17:35:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
self.admin_misp_connector.delete_event(third)
2019-07-12 17:35:02 +02:00
def test_direct(self):
try:
r = self.user_misp_connector.direct_call('events/add', data={'info': 'foo'})
event = MISPEvent()
event.from_dict(**r)
r = self.user_misp_connector.direct_call(f'events/view/{event.id}')
event_get = MISPEvent()
event_get.from_dict(**r)
self.assertDictEqual(event.to_dict(), event_get.to_dict())
finally:
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(event)
2019-07-12 17:35:02 +02:00
def test_freetext(self):
first = self.create_simple_event()
try:
self.admin_misp_connector.toggle_warninglist(warninglist_name='%dns resolv%', force_enable=True)
first = self.user_misp_connector.add_event(first)
2019-08-01 13:19:21 +02:00
# disable_background_processing => returns the parsed data, before insertion
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.freetext(first, '1.1.1.1 foo@bar.de', adhereToWarninglists=False,
2019-08-01 13:19:21 +02:00
distribution=2, returnMetaAttributes=False, pythonify=True,
kw_params={'disable_background_processing': 1})
2019-07-12 17:35:02 +02:00
self.assertTrue(isinstance(r, list))
self.assertEqual(r[0].value, '1.1.1.1')
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.freetext(first, '9.9.9.9 foo@bar.com', adhereToWarninglists='soft',
2019-08-01 13:19:21 +02:00
distribution=2, returnMetaAttributes=False, pythonify=True,
kw_params={'disable_background_processing': 1})
self.assertTrue(isinstance(r, list))
self.assertEqual(r[0].value, '9.9.9.9')
2019-08-16 10:48:06 +02:00
event = self.user_misp_connector.get_event(first, pythonify=True)
2019-08-01 13:19:21 +02:00
self.assertEqual(event.attributes[3].value, '9.9.9.9')
self.assertFalse(event.attributes[3].to_ids)
2019-08-16 10:48:06 +02:00
r_wl = self.user_misp_connector.freetext(first, '8.8.8.8 foo@bar.de', adhereToWarninglists=True,
2019-08-01 13:19:21 +02:00
distribution=2, returnMetaAttributes=False,
kw_params={'disable_background_processing': 0})
self.assertEqual(r_wl[0].value, '8.8.8.8')
2019-08-16 10:48:06 +02:00
event = self.user_misp_connector.get_event(first, pythonify=True)
2019-08-01 13:19:21 +02:00
for attribute in event.attributes:
self.assertFalse(attribute.value == '8.8.8.8')
2019-08-16 10:48:06 +02:00
r = self.user_misp_connector.freetext(first, '1.1.1.1 foo@bar.de', adhereToWarninglists=True,
2019-07-12 17:35:02 +02:00
distribution=2, returnMetaAttributes=True)
self.assertTrue(isinstance(r, list))
self.assertTrue(isinstance(r[0]['types'], dict))
2019-08-01 13:19:21 +02:00
finally:
2019-08-05 16:53:32 +02:00
# Mostly solved https://github.com/MISP/MISP/issues/4886
2019-07-12 17:35:02 +02:00
time.sleep(10)
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-07-12 17:35:02 +02:00
def test_sharing_groups(self):
# add
sg = MISPSharingGroup()
sg.name = 'Testcases SG'
sg.releasability = 'Testing'
sharing_group = self.admin_misp_connector.add_sharing_group(sg, pythonify=True)
self.assertEqual(sharing_group.name, 'Testcases SG')
self.assertEqual(sharing_group.releasability, 'Testing')
# add org
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.add_org_to_sharing_group(sharing_group,
self.test_org, extend=True)
2019-08-01 13:19:21 +02:00
self.assertEqual(r['name'], 'Organisation added to the sharing group.')
2019-07-12 17:35:02 +02:00
# delete org
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.remove_org_from_sharing_group(sharing_group,
self.test_org)
2019-08-01 15:10:48 +02:00
self.assertEqual(r['name'], 'Organisation removed from the sharing group.', r)
2019-07-12 17:35:02 +02:00
# Get list
sharing_groups = self.admin_misp_connector.sharing_groups(pythonify=True)
self.assertTrue(isinstance(sharing_groups, list))
self.assertEqual(sharing_groups[0].name, 'Testcases SG')
# Use the SG
first = self.create_simple_event()
2019-08-01 15:10:48 +02:00
o = first.add_object(name='file')
o.add_attribute('filename', value='foo2.exe')
2019-07-12 17:35:02 +02:00
try:
first = self.user_misp_connector.add_event(first)
first = self.admin_misp_connector.change_sharing_group_on_entity(first, sharing_group.id, pythonify=True)
2019-07-12 17:35:02 +02:00
self.assertEqual(first.SharingGroup['name'], 'Testcases SG')
2019-08-01 15:10:48 +02:00
first_object = self.admin_misp_connector.change_sharing_group_on_entity(first.objects[0], sharing_group.id, pythonify=True)
self.assertEqual(first_object.sharing_group_id, sharing_group.id)
2019-08-05 15:59:24 +02:00
first_attribute = self.admin_misp_connector.change_sharing_group_on_entity(first.attributes[0], sharing_group.id, pythonify=True)
self.assertEqual(first_attribute.distribution, 4)
self.assertEqual(first_attribute.sharing_group_id, int(sharing_group.id))
2019-07-12 17:35:02 +02:00
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-08-01 15:10:48 +02:00
# Delete sharing group
r = self.admin_misp_connector.delete_sharing_group(sharing_group.id)
self.assertEqual(r['message'], 'SharingGroup deleted')
2019-07-12 17:35:02 +02:00
def test_feeds(self):
# Add
feed = MISPFeed()
feed.name = 'TestFeed'
feed.provider = 'TestFeed - Provider'
feed.url = 'http://example.com'
feed = self.admin_misp_connector.add_feed(feed, pythonify=True)
self.assertEqual(feed.name, 'TestFeed')
self.assertEqual(feed.url, 'http://example.com')
# Update
feed.name = 'TestFeed - Update'
feed = self.admin_misp_connector.update_feed(feed, pythonify=True)
self.assertEqual(feed.name, 'TestFeed - Update')
# Delete
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.delete_feed(feed)
2019-07-12 17:35:02 +02:00
self.assertEqual(r['message'], 'Feed deleted.')
# List
feeds = self.admin_misp_connector.feeds(pythonify=True)
self.assertTrue(isinstance(feeds, list))
for feed in feeds:
if feed.name == 'The Botvrij.eu Data':
break
# Get
2019-08-16 10:48:06 +02:00
botvrij = self.admin_misp_connector.get_feed(feed, pythonify=True)
self.assertEqual(botvrij.url, "https://www.botvrij.eu/data/feed-osint")
2019-07-12 17:35:02 +02:00
# Enable
# MISP OSINT
feed = self.admin_misp_connector.enable_feed(feeds[0].id, pythonify=True)
self.assertTrue(feed.enabled)
feed = self.admin_misp_connector.enable_feed_cache(feeds[0].id, pythonify=True)
self.assertTrue(feed.caching_enabled)
# Botvrij.eu
feed = self.admin_misp_connector.enable_feed(botvrij.id, pythonify=True)
self.assertTrue(feed.enabled)
feed = self.admin_misp_connector.enable_feed_cache(botvrij.id, pythonify=True)
self.assertTrue(feed.caching_enabled)
# Cache
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.cache_feed(botvrij)
2019-07-12 17:35:02 +02:00
self.assertEqual(r['message'], 'Feed caching job initiated.')
# Fetch
# Cannot test that, it fetches all the events.
2019-08-16 10:48:06 +02:00
# r = self.admin_misp_connector.fetch_feed(botvrij)
2019-07-12 17:35:02 +02:00
# FIXME https://github.com/MISP/MISP/issues/4834#issuecomment-511889274
# self.assertEqual(r['message'], 'Feed caching job initiated.')
# Cache all enabled feeds
r = self.admin_misp_connector.cache_all_feeds()
self.assertEqual(r['message'], 'Feed caching job initiated.')
# Compare all enabled feeds
r = self.admin_misp_connector.compare_feeds()
# FIXME: https://github.com/MISP/MISP/issues/4834#issuecomment-511890466
# self.assertEqual(r['message'], 'Feed caching job initiated.')
time.sleep(30)
# Disable both feeds
feed = self.admin_misp_connector.disable_feed(feeds[0].id, pythonify=True)
self.assertFalse(feed.enabled)
feed = self.admin_misp_connector.disable_feed(botvrij.id, pythonify=True)
self.assertFalse(feed.enabled)
feed = self.admin_misp_connector.disable_feed_cache(feeds[0].id, pythonify=True)
self.assertFalse(feed.enabled)
feed = self.admin_misp_connector.disable_feed_cache(botvrij.id, pythonify=True)
self.assertFalse(feed.enabled)
def test_servers(self):
# add
server = MISPServer()
server.name = 'Test Server'
server.url = 'https://127.0.0.1'
server.remote_org_id = 1
server.authkey = key
server = self.admin_misp_connector.add_server(server, pythonify=True)
self.assertEqual(server.name, 'Test Server')
# Update
server.name = 'Updated name'
server = self.admin_misp_connector.update_server(server, pythonify=True)
self.assertEqual(server.name, 'Updated name')
# List
servers = self.admin_misp_connector.servers(pythonify=True)
self.assertEqual(servers[0].name, 'Updated name')
# Delete
2019-08-16 10:48:06 +02:00
r = self.admin_misp_connector.delete_server(server)
2019-08-01 13:19:21 +02:00
self.assertEqual(r['name'], 'Server deleted')
2019-07-12 17:35:02 +02:00
def test_roles_expanded(self):
'''Test all possible things regarding roles
1. Use existing roles (ID in test VM):
* Read only (6): Can only connect via API and see events visible by its organisation
* User (3): Same as readonly + create event, tag (using existing tags), add sighting
* Publisher (4): Same as User + publish (also on zmq and kafka), and delegate
* Org Admin (2): Same as publisher + admin org, audit, create tags, templates, sharing groups
* Sync user (5): Same as publisher + sync, create tag, sharing group
* admin (1): Same as Org admin and sync user + site admin, edit regexes, edit object templates
2. Create roles:
* No Auth key access
* Auth key (=> Read only)
* + tagger
* + sightings creator (=> User)
* +
'''
# 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')
self.assertEqual(r['errors'][1]['message'], 'You do not have permission to use this functionality.', r)
r = test_roles_user_connector.add_sighting({'name': 'foo'}, e.attributes[1])
self.assertEqual(r['errors'][1]['message'], 'You do not have permission to use this functionality.', r)
self.user_misp_connector.add_sighting({'source': 'blah'}, e.attributes[0])
sightings = test_roles_user_connector.sightings(e.attributes[0])
self.assertEqual(sightings[0].source, 'blah')
e = test_roles_user_connector.get_event(e)
self.assertEqual(e.attributes[0].sightings[0].source, 'blah')
# FIXME: http://github.com/MISP/MISP/issues/5022
# a = test_roles_user_connector.get_attribute(e.attributes[0])
# self.assertEqual(a.sightings[0].source, 'blah')
# ===== User (the capabilities were tested just before, only testing the publisher capabilities)
self.admin_misp_connector.update_user({'role_id': 3}, test_roles_user)
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.delegate_event(e, self.test_org_delegate)
self.assertEqual(r['errors'][1]['message'], 'You do not have permission to use this functionality.', r)
# ===== Publisher
# Make sure the delegation is enabled
r = self.admin_misp_connector.set_server_setting('MISP.delegation', True, force=True)
self.assertEqual(r['message'], 'Field updated', r)
setting = self.admin_misp_connector.get_server_setting('MISP.delegation')
self.assertTrue(setting['value'])
# ======
self.admin_misp_connector.update_user({'role_id': 4}, test_roles_user)
r = test_roles_user_connector.publish(e)
self.assertEqual(r['message'], 'Job queued', r)
delegation = test_roles_user_connector.delegate_event(e, self.test_org_delegate)
self.assertEqual(delegation.org_id, self.test_org_delegate.id)
self.assertEqual(delegation.requester_org_id, self.test_org.id)
r = test_roles_user_connector.accept_event_delegation(delegation.id)
self.assertEqual(r['errors'][1]['message'], 'You are not authorised to do that.', r)
# Test delegation
delegations = self.delegate_user_misp_connector.event_delegations()
self.assertEqual(delegations[0].id, delegation.id)
2019-08-30 14:19:52 +02:00
r = self.delegate_user_misp_connector.accept_event_delegation(delegation)
self.assertEqual(r['message'], 'Event ownership transferred.')
e = self.delegate_user_misp_connector.get_event(e)
self.assertTrue(isinstance(e, MISPEvent), e)
self.assertEqual(e.info, 'Test Roles')
self.assertEqual(e.org.name, 'Test Org - delegate')
r = self.delegate_user_misp_connector.delete_event(e)
self.assertEqual(r['message'], 'Event deleted.', r)
# Change base_event UUID do we can add it
base_event.uuid = str(uuid4())
e = test_roles_user_connector.add_event(base_event)
delegation = test_roles_user_connector.delegate_event(e, self.test_org_delegate)
2019-08-30 14:19:52 +02:00
r = test_roles_user_connector.discard_event_delegation(delegation.id)
self.assertEqual(r['message'], 'Delegation request deleted.')
e = test_roles_user_connector.get_event(e)
self.assertTrue(isinstance(e, MISPEvent), e)
self.assertEqual(e.info, 'Test Roles')
self.assertEqual(e.org_id, int(self.test_org.id))
finally:
self.user_misp_connector.delete_event(e)
# Publisher
self.admin_misp_connector.update_user({'role_id': 4}, test_roles_user)
# Org Admin
self.admin_misp_connector.update_user({'role_id': 2}, test_roles_user)
# Sync User
self.admin_misp_connector.update_user({'role_id': 5}, test_roles_user)
# Admin
self.admin_misp_connector.update_user({'role_id': 1}, test_roles_user)
finally:
self.admin_misp_connector.delete_user(test_roles_user)
self.admin_misp_connector.delete_tag(test_tag)
def test_expansion(self):
first = self.create_simple_event()
try:
with open('tests/viper-test-files/test_files/whoami.exe', 'rb') as f:
first.add_attribute('malware-sample', value='whoami.exe', data=BytesIO(f.read()), expand='binary')
first.run_expansions()
first = self.admin_misp_connector.add_event(first, pythonify=True)
self.assertEqual(len(first.objects), 7)
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-10-16 17:22:19 +02:00
def test_user_settings(self):
first = self.create_simple_event()
first.distribution = 3
first.add_tag('test_publish_filter')
first.add_tag('test_publish_filter_not')
second = self.create_simple_event()
second.distribution = 3
try:
# Set
setting = self.admin_misp_connector.set_user_setting('dashboard_access', 1, pythonify=True)
setting_value = {'Tag.name': 'test_publish_filter'}
setting = self.admin_misp_connector.set_user_setting('publish_alert_filter', setting_value, pythonify=True)
self.assertTrue(isinstance(setting, MISPUserSetting))
self.assertEqual(setting.value, setting_value)
# Get
# FIXME: https://github.com/MISP/MISP/issues/5297
# setting = self.admin_misp_connector.get_user_setting('dashboard_access', pythonify=True)
# Get All
user_settings = self.admin_misp_connector.user_settings(pythonify=True)
# TODO: Make that one better
self.assertTrue(isinstance(user_settings, list))
# Test if publish_alert_filter works
first = self.admin_misp_connector.add_event(first, pythonify=True)
second = self.admin_misp_connector.add_event(second, pythonify=True)
r = self.user_misp_connector.change_user_password('Password1234')
self.assertEqual(r['message'], 'Password Changed.')
self.test_usr.autoalert = True
self.test_usr.termsaccepted = True
user = self.user_misp_connector.update_user(self.test_usr, pythonify=True)
self.assertTrue(user.autoalert)
self.admin_misp_connector.publish(first, alert=True)
self.admin_misp_connector.publish(second, alert=True)
time.sleep(10)
# FIXME https://github.com/MISP/MISP/issues/4872
# mail_logs = self.admin_misp_connector.search_logs(model='User', action='email', limit=2, pythonify=True)
mail_logs = self.admin_misp_connector.search_logs(model='User', action='email', created=datetime.now() - timedelta(seconds=30), pythonify=True)
if mail_logs:
# FIXME: On travis, the mails aren't working, so we stik that.
self.assertEqual(len(mail_logs), 3)
self.assertTrue(mail_logs[0].title.startswith(f'Email to {self.admin_misp_connector._current_user.email}'), mail_logs[0].title)
self.assertTrue(mail_logs[1].title.startswith(f'Email to {self.user_misp_connector._current_user.email}'), mail_logs[1].title)
self.assertTrue(mail_logs[2].title.startswith(f'Email to {self.user_misp_connector._current_user.email}'), mail_logs[2].title)
2019-10-16 17:22:19 +02:00
# Delete
# FIXME: https://github.com/MISP/MISP/issues/5297
# response = self.admin_misp_connector.delete_user_setting('publish_alert_filter')
finally:
self.test_usr.autoalert = False
self.user_misp_connector.update_user(self.test_usr)
# Delete event
self.admin_misp_connector.delete_event(first)
self.admin_misp_connector.delete_event(second)
2019-08-29 18:08:53 +02:00
def test_communities(self):
communities = self.admin_misp_connector.communities(pythonify=True)
self.assertEqual(communities[0].name, 'CIRCL Private Sector Information Sharing Community - aka MISPPRIV')
community = self.admin_misp_connector.get_community(communities[1], pythonify=True)
self.assertEqual(community.name, 'CIRCL n/g CSIRT information sharing community - aka MISP')
# FIXME: Fails on travis for now due to GPG misconfigured
2019-08-30 12:03:52 +02:00
# r = self.admin_misp_connector.request_community_access(community, mock=False)
# self.assertTrue(r['message'], 'Request sent.')
# r = self.admin_misp_connector.request_community_access(community, mock=True)
# mail = email.message_from_string(r['headers'] + '\n' + r['message'])
# for k, v in mail.items():
# if k == 'To':
# self.assertEqual(v, 'info@circl.lu')
2019-08-29 18:08:53 +02:00
2019-07-12 17:35:02 +02:00
def test_upload_stix(self):
# FIXME https://github.com/MISP/MISP/issues/4892
pass
2019-04-04 16:39:17 +02:00
def test_toggle_global_pythonify(self):
first = self.create_simple_event()
second = self.create_simple_event()
try:
self.admin_misp_connector.toggle_global_pythonify()
first = self.admin_misp_connector.add_event(first)
self.assertTrue(isinstance(first, MISPEvent))
self.admin_misp_connector.toggle_global_pythonify()
second = self.admin_misp_connector.add_event(second)
self.assertTrue(isinstance(second, dict))
finally:
# Delete event
2019-08-16 10:48:06 +02:00
self.admin_misp_connector.delete_event(first)
2019-10-16 17:22:19 +02:00
self.admin_misp_connector.delete_event(second)
def test_first_last_seen(self):
local_tz = datetime.now(timezone.utc).astimezone().tzinfo
event = MISPEvent()
event.info = 'Test First Last seen'
event.add_attribute('ip-dst', '8.8.8.8', first_seen='2020-01-04', last_seen='2020-01-04T12:30:34.323242+0800')
obj = event.add_object(name='file', first_seen=1580147259.268763, last_seen=1580147300)
attr = obj.add_attribute('filename', 'blah.exe')
attr.first_seen = '2022-01-30'
attr.last_seen = '2022-02-23'
try:
first = self.admin_misp_connector.add_event(event, pythonify=True)
# Simple attribute
self.assertEqual(first.attributes[0].first_seen, datetime(2020, 1, 4, 0, 0, tzinfo=local_tz))
self.assertEqual(first.attributes[0].last_seen, datetime(2020, 1, 4, 4, 30, 34, 323242, tzinfo=timezone.utc))
# Object
self.assertEqual(first.objects[0].first_seen, datetime(2020, 1, 27, 17, 47, 39, 268763, tzinfo=timezone.utc))
self.assertEqual(first.objects[0].last_seen, datetime(2020, 1, 27, 17, 48, 20, tzinfo=timezone.utc))
# Object attribute
self.assertEqual(first.objects[0].attributes[0].first_seen, datetime(2022, 1, 30, 0, 0, tzinfo=local_tz))
self.assertEqual(first.objects[0].attributes[0].last_seen, datetime(2022, 2, 23, 0, 0, tzinfo=local_tz))
2020-01-28 16:01:14 +01:00
# Update values
# Attribute in full event
now = datetime.now().astimezone()
first.attributes[0].last_seen = now
first = self.admin_misp_connector.update_event(first, pythonify=True)
self.assertEqual(first.attributes[0].last_seen, now)
# Object only
now = datetime.now().astimezone()
obj = first.objects[0]
obj.last_seen = now
obj = self.admin_misp_connector.update_object(obj, pythonify=True)
self.assertEqual(obj.last_seen, now)
# Attribute in object only
now = datetime.now().astimezone()
attr = obj.attributes[0]
attr.last_seen = now
attr = self.admin_misp_connector.update_attribute(attr, pythonify=True)
self.assertEqual(attr.last_seen, now)
finally:
self.admin_misp_connector.delete_event(first)
if __name__ == '__main__':
unittest.main()