Extend add_object for new server-side feature breakOnDuplicate

[chrisinmtown]

Extend the add_object() method for the duplicate-detection feature supported by the server. MISP was extended to allow detection of duplicate object-attribute collections - see https://github.com/MISP/MISP/issues/2826 - but today there appears to be no way to invoke that feature from PyMISP.

I'll be glad to propose a change to api.py, but it's not obvious how to pass the parameter that's read here:

https://github.com/MISP/MISP/blob/ca5043a184b8eec53fb4093377fc4a727d4345cf/app/Controller/ObjectsController.php#L224

Here's client code you might use to test the feature:

#!/usr/bin/env python3

import sys
import urllib3
from pymisp import PyMISP, MISPAttribute, MISPEvent, MISPObject
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def add_phishing_obj(misp_client, event):
    """Create an object on an event with a single attribute"""
    obj = MISPObject(name='phishing', strict=True)
    obj.add_attribute(object_relation='url', simple_value='https://simple.tld/')
    response = misp_client.add_object(event.id, obj, pythonify=True)  # *NEED NEW PARAMETER HERE*
    print('added obj on evt: {}'.format(obj.to_json(sort_keys=True, indent=4)))
    return obj

misp_client = PyMISP(url='https://misp:48752', key='keykeykey', ssl=False)
if not misp_client:
    print('PyMISP() failed\n', file=sys.stderr)
    sys.exit(1)

event = MISPEvent()
event.distribution = 0  # internal
event.published = False
event.info = 'Test event for dupes'
event = misp_client.add_event(event, pythonify=True)
print('created event:\n{}'.format(event.to_json(sort_keys=True, indent=4)))

obj1 = add_phishing_obj(misp_client, event)
obj2 = add_phishing_obj(misp_client, event)

[chrisinmtown]

Here's my best guess at the required change to api.py. Still need an automated test case tho.

diff --git a/pymisp/api.py b/pymisp/api.py
index a7811a4..67a3ff6 100644
--- a/pymisp/api.py
+++ b/pymisp/api.py
@@ -415,15 +415,17 @@ class PyMISP:
         r = self._prepare_request('HEAD', f'objects/view/{object_id}')
         return self._check_head_response(r)

-    def add_object(self, event: Union[MISPEvent, int, str, UUID], misp_object: MISPObject, pythonify: bool = False) -> Union[Dict, MISPObject]:
+    def add_object(self, event: Union[MISPEvent, int, str, UUID], misp_object: MISPObject, break_on_duplicate:bool = False, pythonify: bool = False) -> Union[Dict, MISPObject]:
         """Add a MISP Object to an existing MISP event

         :param event: event to extend
         :param misp_object: object to add
+        :param break_on_duplicate: if True, check and reject if this object's attributes match an existing object's attributes; may require much time
         :param pythonify: Returns a PyMISP Object instead of the plain json output
         """
         event_id = get_uuid_or_id_from_abstract_misp(event)
-        r = self._prepare_request('POST', f'objects/add/{event_id}', data=misp_object)
+        params = { 'breakOnDuplicate': True } if break_on_duplicate else {}
+        r = self._prepare_request('POST', f'objects/add/{event_id}', data=misp_object, params=params)
         new_object = self._check_json_response(r)
         if not (self.global_pythonify or pythonify) or 'errors' in new_object:
             return new_object

[github-germ]

I discovered issues with the server side MISPObject deduplication within an Event and have a proposed fix. Please see Proposed fix for deduplicating MISPObjects within an Event via breakOnDuplicate in https://github.com/MISP/MISP/issues/6838

[chrisinmtown]

Is a similar extension needed to other PyMISP methods such as update_object() ?

[github-germ]

The fix I've proposed for the server side object dedup worked for PyMISP 2.4.133 add_event, update_event, add_object without any changes to PyMISP, i.e. setting MISPObject.breakOnDuplicate was passed in the POST request and seen within the server.

[chrisinmtown]

Glad to hear, but personally I would like the PyMISP API to advertise its behaviors clearly via parameters and docstring entries, instead of requiring me to use semi-secret back-door approaches like setting an undocumented attribute on an object.

[chrisinmtown]

I see the server side is being changed, for example https://github.com/MISP/MISP/commit/0a062215b8c5a50b8e1baebf9e37472d89d1b567

I'm not sure what change to propose for PyMISP but if you give me a hint I will try.

[Rafiot]

Fixed, thank you for the proposal: https://github.com/MISP/PyMISP/commit/c3d6c3cc732f1c15bc1faabdb76ff965cd09b060

[chrisinmtown]

Glad to see this was helpful @Rafiot but please comment about method update_object() -- it needs a similar extension, right?

[Rafiot]

If you pass a key breakOnDuplicate directly in the object, it will automatically trigger the feature: https://github.com/MISP/PyMISP/commit/5b97b7d0158906cd0f646a7273a3ca5b1828cd15

This test case calls add_object, but the same code is used on update, so it will work the same. Not sure how to document that properly, but maybe directly in the MISPObject class?

[chrisinmtown]

Thanks for confirming @Rafiot but I feel that asking users to force a key not in the data model into their object just to trigger a server-side behavior is inappropriate. It feels secret, hidden, like an Easter egg feature. And as a user, I would never think of reading the MISPObject class to search for ways to influence the add_object and update_object methods.

I'll put my money where my mouth is, my next post here will be a proposal for extending the update_object() method similarly to what you accepted for add_object. I'll take a stab at the test case too.

Are there any other PyMISP methods that honor this check-for-dupes request?

[Rafiot]

It definitely deserves to be exposed on the update method.

The other place where it should be mentioned is the MISPEvent.add_object method: adding/updating a complete event with objects having that key will automatically filtered out the duplicate objects.

[chrisinmtown]

Here's the first trivial proposal, more soon.

diff --git a/pymisp/api.py b/pymisp/api.py
index 0824050..78df58b 100644
--- a/pymisp/api.py
+++ b/pymisp/api.py
@@ -435,18 +435,20 @@ class PyMISP:
         o.from_dict(**new_object)
         return o

-    def update_object(self, misp_object: MISPObject, object_id: Optional[int] = None, pythonify: bool = False) -> Union[Dict, MISPObject]:
+    def update_object(self, misp_object: MISPObject, object_id: Optional[int] = None, pythonify: bool = False, break_on_duplicate: bool = False) -> Union[Dict, MISPObject]:
         """Update an object on a MISP instance

         :param misp_object: object to update
         :param object_id: ID of object to update
         :param pythonify: Returns a PyMISP Object instead of the plain json output
+        :param break_on_duplicate: if True, check and reject if this object's attributes match an existing object's attributes; may require much time
         """
         if object_id is None:
             oid = get_uuid_or_id_from_abstract_misp(misp_object)
         else:
             oid = get_uuid_or_id_from_abstract_misp(object_id)
-        r = self._prepare_request('POST', f'objects/edit/{oid}', data=misp_object)
+        params = {'breakOnDuplicate': True} if break_on_duplicate else {}
+        r = self._prepare_request('POST', f'objects/edit/{oid}', data=misp_object, kw_params=params)
         updated_object = self._check_json_response(r)
         if not (self.global_pythonify or pythonify) or 'errors' in updated_object:
             return updated_object

[chrisinmtown]

Second docstring proposal:

diff --git a/pymisp/mispevent.py b/pymisp/mispevent.py
index 9363bd2..4266bda 100644
--- a/pymisp/mispevent.py
+++ b/pymisp/mispevent.py
@@ -1450,7 +1450,12 @@ class MISPEvent(AbstractMISP):
         return objects

     def add_object(self, obj: Union[MISPObject, dict, None] = None, **kwargs) -> MISPObject:
-        """Add an object to the Event, either by passing a MISPObject, or a dictionary"""
+        """
+        Add an object to the Event, either by passing a MISPObject or a dictionary. If the
+        object has key 'breakOnDuplicate' with any value, the server detects and drops
+        duplicate objects within the Event; i.e., objects that contain attributes with
+        matching object relation, category, type and value entries. May require much time.
+        """
         if isinstance(obj, MISPObject):
             misp_obj = obj
         elif isinstance(obj, dict):

[chrisinmtown]

I see that even tho I created this issue, I lack privs to reopen it. Probably it's dumb of me to keep posting to a closed issue. @Rafiot should I create a new one, or how do you suggest proceeding?

[chrisinmtown]

Last proposal for today, draft new tests for update_object, this is really a guess:

diff --git a/tests/testlive_comprehensive.py b/tests/testlive_comprehensive.py
index bcab2b4..7d1d2fe 100644
--- a/tests/testlive_comprehensive.py
+++ b/tests/testlive_comprehensive.py
@@ -1445,16 +1445,27 @@ class TestComprehensive(unittest.TestCase):
             self.assertEqual(len(obj_attrs), 1, obj_attrs)
             self.assertEqual(r.name, 'file', r)

-            # Test break_on_duplicate at object level
+            # Test break_on_duplicate at object level for add_object
             fo_dup, peo_dup, _ = make_binary_objects('tests/viper-test-files/test_files/whoami.exe')
             r = self.user_misp_connector.add_object(first, peo_dup, break_on_duplicate=True)
             self.assertTrue("Duplicate object found" in r['errors'][1]['errors'], r)

-            # Test break on duplicate with breakOnDuplicate key in object
+            # Test break on duplicate with breakOnDuplicate key in object for add_object
             fo_dup.breakOnDuplicate = True
             r = self.user_misp_connector.add_object(first, fo_dup)
             self.assertTrue("Duplicate object found" in r['errors'][1]['errors'], r)

+            # Test break_on_duplicate at object level for update_object
+            _, peo2, _ = make_binary_objects('tests/viper-test-files/test_files/cmd.exe')
+            r = self.user_misp_connector.add_object(first, peo2, pythonify=True)
+            self.assertEqual(r.name, 'pe', r)
+            r = self.user_misp_connector.update_object(fo, r.object_uuid, break_on_duplicate=True)
+            self.assertTrue("Duplicate object found" in r['errors'][1]['errors'], r)
+
+            # Test break on duplicate with breakOnDuplicate key in object for update_object
+            r = self.user_misp_connector.update_object(fo_dup, r.object_uuid)
+            self.assertTrue("Duplicate object found" in r['errors'][1]['errors'], r)
+
             # Test refs
             r = self.user_misp_connector.add_object_reference(fo.ObjectReference[0])
             self.assertEqual(r.object_uuid, fo.uuid, r.to_json())

[chrisinmtown]

Thanks @Rafiot for reopening this.

[chrisinmtown]

hi @Rafiot I know the MISP server team made some fixes to object de-duplication recently, would you please find time to fold in these little changes to PyMISP to keep up with that?

[Rafiot]

So break_on_duplicate at object level for update_object doesn't work for now.

[Rafiot]

WiP pull request: https://github.com/MISP/PyMISP/pull/709