PyMISP/tests/test_mispevent.py

#!/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()