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 MISP/MISP#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 MISP/MISP#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 MISP/MISP@0a06221

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: c3d6c3c

[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: 5b97b7d

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):