#!/usr/bin/env python from __future__ import annotations import unittest import json from io import BytesIO import glob import hashlib from datetime import date, datetime from pymisp import (MISPAttribute, MISPEvent, MISPGalaxy, MISPObject, MISPOrganisation, MISPSighting, MISPTag) from pymisp.exceptions import InvalidMISPObject from pymisp.tools import GitVulnFinderObject class TestMISPEvent(unittest.TestCase): def setUp(self) -> None: self.maxDiff = None self.mispevent = MISPEvent() def init_event(self) -> None: self.mispevent.info = 'This is a test' self.mispevent.distribution = 1 self.mispevent.threat_level_id = 1 self.mispevent.analysis = 1 self.mispevent.set_date("2017-12-31") # test the set date method def test_simple(self) -> None: with open('tests/mispevent_testfiles/simple.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_event(self) -> None: self.init_event() self.mispevent.publish() with open('tests/mispevent_testfiles/event.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_loadfile(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/event.json') with open('tests/mispevent_testfiles/event.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_loadfile_validate(self) -> None: misp_event = MISPEvent() misp_event.load_file('tests/mispevent_testfiles/event.json', validate=True) def test_loadfile_validate_strict(self) -> None: misp_event = MISPEvent(strict_validation=True) misp_event.load_file('tests/mispevent_testfiles/event.json', validate=True) def test_event_tag(self) -> None: self.init_event() self.mispevent.add_tag('bar') self.mispevent.add_tag(name='baz') new_tag = MISPTag() new_tag.from_dict(name='foo') self.mispevent.add_tag(new_tag) with open('tests/mispevent_testfiles/event_tags.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_event_galaxy(self) -> None: self.init_event() with open('tests/mispevent_testfiles/galaxy.json') as f: galaxy = json.load(f) misp_galaxy = MISPGalaxy() misp_galaxy.from_dict(**galaxy) self.mispevent.add_galaxy(misp_galaxy) self.assertEqual(self.mispevent.galaxies[0].to_json(sort_keys=True, indent=2), json.dumps(galaxy, sort_keys=True, indent=2)) def test_attribute(self) -> None: self.init_event() a: MISPAttribute = self.mispevent.add_attribute('filename', 'bar.exe') # type: ignore[assignment] del a.uuid a = self.mispevent.add_attribute_tag('osint', 'bar.exe') # type: ignore[assignment] attr_tags = self.mispevent.get_attribute_tag('bar.exe') self.assertEqual(self.mispevent.attributes[0].tags[0].name, 'osint') self.assertEqual(attr_tags[0].name, 'osint') with open('tests/mispevent_testfiles/attribute.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) # Fake setting an attribute ID for testing self.mispevent.attributes[0].id = 42 self.mispevent.delete_attribute('42') with open('tests/mispevent_testfiles/attribute_del.json') as f: ref_json = json.load(f) self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_attribute_galaxy(self) -> None: self.init_event() with open('tests/mispevent_testfiles/galaxy.json') as f: galaxy = json.load(f) misp_galaxy = MISPGalaxy() misp_galaxy.from_dict(**galaxy) attribute = MISPAttribute() attribute.from_dict(**{'type': 'github-username', 'value': 'adulau'}) attribute.add_galaxy(misp_galaxy) self.mispevent.add_attribute(**attribute) self.assertEqual( self.mispevent.attributes[0].galaxies[0].to_json(sort_keys=True, indent=2), json.dumps(galaxy, sort_keys=True, indent=2) ) def test_to_dict_json_format(self) -> None: misp_event = MISPEvent() av_signature_object = MISPObject("av-signature") av_signature_object.add_attribute("signature", "EICAR") av_signature_object.add_attribute("software", "ClamAv") misp_event.add_object(av_signature_object) self.assertEqual(json.loads(misp_event.to_json()), misp_event.to_dict(json_format=True)) def test_object_tag(self) -> None: self.mispevent.add_object(name='file', strict=True) a: MISPAttribute = self.mispevent.objects[0].add_attribute('filename', value='') # type: ignore[assignment] self.assertEqual(a, None) a = self.mispevent.objects[0].add_attribute('filename', value=None) # type: ignore[assignment] self.assertEqual(a, None) a = self.mispevent.objects[0].add_attribute('filename', value='bar', Tag=[{'name': 'blah'}]) # type: ignore[assignment] del a.uuid self.assertEqual(self.mispevent.objects[0].attributes[0].tags[0].name, 'blah') self.assertTrue(self.mispevent.objects[0].has_attributes_by_relation(['filename'])) self.assertEqual(len(self.mispevent.objects[0].get_attributes_by_relation('filename')), 1) self.mispevent.add_object(name='url', strict=True) a = self.mispevent.objects[1].add_attribute('url', value='https://www.circl.lu') # type: ignore[assignment] del a.uuid self.mispevent.objects[0].uuid = 'a' self.mispevent.objects[1].uuid = 'b' reference = self.mispevent.objects[0].add_reference(self.mispevent.objects[1], 'baz', comment='foo') del reference.uuid self.assertEqual(self.mispevent.objects[0].references[0].relationship_type, 'baz') with open('tests/mispevent_testfiles/event_obj_attr_tag.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) @unittest.skip("Not supported on MISP: https://github.com/MISP/MISP/issues/2638 - https://github.com/MISP/PyMISP/issues/168") def test_object_level_tag(self) -> None: self.mispevent.add_object(name='file', strict=True) self.mispevent.objects[0].add_attribute('filename', value='bar') self.mispevent.objects[0].add_tag('osint') # type: ignore[attr-defined] self.mispevent.objects[0].uuid = 'a' with open('tests/mispevent_testfiles/event_obj_tag.json') as f: ref_json = json.load(f) self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_object_galaxy(self) -> None: self.init_event() misp_object = MISPObject('github-user') misp_object.add_attribute('username', 'adulau') misp_object.add_attribute('repository', 'cve-search') self.mispevent.add_object(misp_object) with open('tests/mispevent_testfiles/galaxy.json') as f: galaxy = json.load(f) misp_galaxy = MISPGalaxy() misp_galaxy.from_dict(**galaxy) self.mispevent.objects[0].attributes[0].add_galaxy(misp_galaxy) self.assertEqual( self.mispevent.objects[0].attributes[0].galaxies[0].to_json(sort_keys=True, indent=2), json.dumps(galaxy, sort_keys=True, indent=2) ) def test_malware(self) -> None: with open('tests/mispevent_testfiles/simple.json', 'rb') as f: pseudofile = BytesIO(f.read()) self.init_event() a: MISPAttribute = self.mispevent.add_attribute('malware-sample', 'bar.exe', data=pseudofile) # type: ignore[assignment] del a.uuid attribute = self.mispevent.attributes[0] self.assertEqual(attribute.malware_binary, pseudofile) with open('tests/mispevent_testfiles/malware.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_existing_malware(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/malware_exist.json') with open('tests/mispevent_testfiles/simple.json', 'rb') as f: pseudofile = BytesIO(f.read()) self.assertTrue(self.mispevent.objects[0].get_attributes_by_relation('malware-sample')[0].malware_binary) if _mb := self.mispevent.objects[0].get_attributes_by_relation('malware-sample')[0].malware_binary: self.assertEqual(_mb.read(), pseudofile.read()) def test_sighting(self) -> None: sighting = MISPSighting() sighting.from_dict(value='1', type='bar', timestamp=11111111) with open('tests/mispevent_testfiles/sighting.json') as f: ref_json = json.load(f) self.assertEqual(sighting.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_existing_event(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') with open('tests/mispevent_testfiles/existing_event.json') as f: ref_json = json.load(f) self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_shadow_attributes_existing(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/shadow.json') with open('tests/mispevent_testfiles/shadow.json') as f: ref_json = json.load(f) self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) @unittest.skip("Not supported on MISP.") def test_shadow_attributes(self) -> None: self.init_event() p = self.mispevent.add_proposal(type='filename', value='baz.jpg') del p.uuid a: MISPAttribute = self.mispevent.add_attribute('filename', 'bar.exe') # type: ignore[assignment] del a.uuid p = self.mispevent.attributes[0].add_proposal(type='filename', value='bar.pdf') del p.uuid with open('tests/mispevent_testfiles/proposals.json') as f: ref_json = json.load(f) self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_default_attributes(self) -> None: self.mispevent.add_object(name='file', strict=True) a: MISPAttribute = self.mispevent.objects[0].add_attribute('filename', value='bar', Tag=[{'name': 'blah'}]) # type: ignore[assignment] del a.uuid a = self.mispevent.objects[0].add_attribute('pattern-in-file', value='baz') # type: ignore[assignment] self.assertEqual(a.category, 'Artifacts dropped') del a.uuid self.mispevent.add_object(name='file', strict=False, default_attributes_parameters=self.mispevent.objects[0].attributes[0]) a = self.mispevent.objects[1].add_attribute('filename', value='baz') # type: ignore[assignment] del a.uuid self.mispevent.objects[0].uuid = 'a' self.mispevent.objects[1].uuid = 'b' with open('tests/mispevent_testfiles/event_obj_def_param.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_obj_default_values(self) -> None: self.init_event() self.mispevent.add_object(name='whois', strict=True) a: MISPAttribute = self.mispevent.objects[0].add_attribute('registrar', value='registar.example.com') # type: ignore[assignment] del a.uuid a = self.mispevent.objects[0].add_attribute('domain', value='domain.example.com') # type: ignore[assignment] del a.uuid a = self.mispevent.objects[0].add_attribute('nameserver', value='ns1.example.com') # type: ignore[assignment] del a.uuid a = self.mispevent.objects[0].add_attribute('nameserver', value='ns2.example.com', disable_correlation=False, to_ids=True, category='External analysis') # type: ignore[assignment] del a.uuid self.mispevent.objects[0].uuid = 'a' with open('tests/mispevent_testfiles/def_param.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_obj_references_export(self) -> None: self.init_event() obj1 = MISPObject(name="file") obj2 = MISPObject(name="url", standalone=False) obj1.add_reference(obj2, "downloads") obj2.add_reference(obj1, "downloaded-by") self.assertFalse("ObjectReference" in obj1.jsonable()) self.assertTrue("ObjectReference" in obj2.jsonable()) self.mispevent.add_object(obj1) obj2.standalone = True self.assertTrue("ObjectReference" in obj1.jsonable()) self.assertFalse("ObjectReference" in obj2.jsonable()) def test_event_not_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) def test_event_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.mispevent.info = 'blah' self.assertTrue(self.mispevent.edited) def test_event_tag_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) self.mispevent.add_tag('foo') self.assertTrue(self.mispevent.edited) def test_event_attribute_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.mispevent.attributes[0].value = 'blah' self.assertTrue(self.mispevent.attributes[0].edited) self.assertFalse(self.mispevent.attributes[1].edited) self.assertTrue(self.mispevent.edited) def test_event_attribute_tag_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) self.mispevent.attributes[0].tags[0].name = 'blah' self.assertTrue(self.mispevent.attributes[0].tags[0].edited) self.assertFalse(self.mispevent.attributes[0].tags[1].edited) self.assertTrue(self.mispevent.attributes[0].edited) self.assertTrue(self.mispevent.edited) def test_event_attribute_tag_edited_second(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) self.mispevent.attributes[0].add_tag(name='blah') self.assertTrue(self.mispevent.attributes[0].edited) self.assertTrue(self.mispevent.edited) def test_event_object_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) self.mispevent.objects[0].comment = 'blah' self.assertTrue(self.mispevent.objects[0].edited) self.assertFalse(self.mispevent.objects[1].edited) self.assertTrue(self.mispevent.edited) def test_event_object_attribute_edited(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) self.mispevent.objects[0].attributes[0].comment = 'blah' self.assertTrue(self.mispevent.objects[0].attributes[0].edited) self.assertTrue(self.mispevent.objects[0].edited) self.assertTrue(self.mispevent.edited) def test_event_object_attribute_edited_tag(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') self.assertFalse(self.mispevent.edited) self.mispevent.objects[0].attributes[0].add_tag('blah') self.assertTrue(self.mispevent.objects[0].attributes[0].edited) self.assertTrue(self.mispevent.objects[0].edited) self.assertTrue(self.mispevent.edited) with open('tests/mispevent_testfiles/existing_event_edited.json') as f: ref_json = json.load(f) self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_obj_by_id(self) -> None: self.mispevent.load_file('tests/mispevent_testfiles/existing_event.json') misp_obj = self.mispevent.get_object_by_id(1556) self.assertEqual(misp_obj.uuid, '5a3cd604-e11c-4de5-bbbf-c170950d210f') def test_userdefined_object_custom_template(self) -> None: self.init_event() with open('tests/mispevent_testfiles/test_object_template/definition.json') as f: template = json.load(f) self.mispevent.add_object(name='test_object_template', strict=True, misp_objects_template_custom=template) with self.assertRaises(InvalidMISPObject) as e: # Fail on required self.mispevent.to_json(sort_keys=True, indent=2) self.assertEqual(e.exception.message, '{\'member3\'} are required.') a: MISPAttribute = self.mispevent.objects[0].add_attribute('member3', value='foo') # type: ignore[assignment] del a.uuid with self.assertRaises(InvalidMISPObject) as e: # Fail on requiredOneOf self.mispevent.to_json(sort_keys=True, indent=2) self.assertEqual(e.exception.message, 'At least one of the following attributes is required: member1, member2') a = self.mispevent.objects[0].add_attribute('member1', value='bar') # type: ignore[assignment] del a.uuid a = self.mispevent.objects[0].add_attribute('member1', value='baz') # type: ignore[assignment] del a.uuid with self.assertRaises(InvalidMISPObject) as e: # member1 is not a multiple self.mispevent.to_json(sort_keys=True, indent=2) self.assertEqual(e.exception.message, 'Multiple occurrences of member1 is not allowed') self.mispevent.objects[0].attributes = self.mispevent.objects[0].attributes[:2] self.mispevent.objects[0].uuid = 'a' with open('tests/mispevent_testfiles/misp_custom_obj.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_userdefined_object_custom_dir(self) -> None: self.init_event() self.mispevent.add_object(name='test_object_template', strict=True, misp_objects_path_custom='tests/mispevent_testfiles') with self.assertRaises(InvalidMISPObject) as e: # Fail on required self.mispevent.to_json(sort_keys=True, indent=2) self.assertEqual(e.exception.message, '{\'member3\'} are required.') a: MISPAttribute = self.mispevent.objects[0].add_attribute('member3', value='foo') # type: ignore[assignment] del a.uuid with self.assertRaises(InvalidMISPObject) as e: # Fail on requiredOneOf self.mispevent.to_json(sort_keys=True, indent=2) self.assertEqual(e.exception.message, 'At least one of the following attributes is required: member1, member2') a = self.mispevent.objects[0].add_attribute('member1', value='bar') # type: ignore[assignment] del a.uuid a = self.mispevent.objects[0].add_attribute('member1', value='baz') # type: ignore[assignment] del a.uuid with self.assertRaises(InvalidMISPObject) as e: # member1 is not a multiple self.mispevent.to_json(sort_keys=True, indent=2) self.assertEqual(e.exception.message, 'Multiple occurrences of member1 is not allowed') self.mispevent.objects[0].attributes = self.mispevent.objects[0].attributes[:2] self.mispevent.objects[0].uuid = 'a' with open('tests/mispevent_testfiles/misp_custom_obj.json') as f: ref_json = json.load(f) del self.mispevent.uuid self.assertEqual(self.mispevent.to_json(sort_keys=True, indent=2), json.dumps(ref_json, sort_keys=True, indent=2)) def test_first_last_seen(self) -> None: me = MISPEvent() me.info = 'Test First and Last Seen' me.date = '2020.01.12' # type: ignore[assignment] self.assertEqual(me.date.day, 12) me.add_attribute('ip-dst', '8.8.8.8', first_seen='06-21-1998', last_seen=1580213607.469571) self.assertEqual(me.attributes[0].first_seen.year, 1998) self.assertEqual(me.attributes[0].last_seen.year, 2020) now = datetime.now().astimezone() me.attributes[0].last_seen = now today = date.today() me.attributes[0].first_seen = today # type: ignore[assignment] self.assertEqual(me.attributes[0].first_seen.year, today.year) self.assertEqual(me.attributes[0].last_seen, now) def test_feed(self) -> None: me = MISPEvent() me.info = 'Test feed' org = MISPOrganisation() org.name = 'TestOrg' org.uuid = '123478' me.Orgc = org me.add_attribute('ip-dst', '8.8.8.8') obj = me.add_object(name='file') obj.add_attributes('filename', *['foo.exe', 'bar.exe']) h = hashlib.new('md5') h.update(b'8.8.8.8') hash_attr_val = h.hexdigest() feed = me.to_feed(with_meta=True) self.assertEqual(feed['Event']['_hashes'][0], hash_attr_val) self.assertEqual(feed['Event']['_manifest'][me.uuid]['info'], 'Test feed') self.assertEqual(len(feed['Event']['Object'][0]['Attribute']), 2) def test_object_templates(self) -> None: me = MISPEvent() for template in glob.glob(str(me.misp_objects_path / '*' / 'definition.json')): with open(template) as f: t_json = json.load(f) if 'requiredOneOf' in t_json: obj_relations = set(t_json['attributes'].keys()) subset = set(t_json['requiredOneOf']).issubset(obj_relations) self.assertTrue(subset, f'{t_json["name"]}') if 'required' in t_json: obj_relations = set(t_json['attributes'].keys()) subset = set(t_json['required']).issubset(obj_relations) self.assertTrue(subset, f'{t_json["name"]}') for obj_relation, entry in t_json['attributes'].items(): self.assertTrue(entry['misp-attribute'] in me.describe_types['types'], f'Missing type: {entry["misp-attribute"]}') if 'categories' in entry: subset = set(entry['categories']).issubset(me.describe_types['categories']) self.assertTrue(subset, f'{t_json["name"]} - {obj_relation}') def test_git_vuln_finder(self) -> None: with open('tests/git-vuln-finder-quagga.json') as f: dump = json.load(f) for vuln in dump.values(): author = vuln['author'] vuln_finder = GitVulnFinderObject(vuln) self.assertEqual(vuln_finder.get_attributes_by_relation('author')[0].value, author) if __name__ == '__main__': unittest.main()