Ciac 8941/enhancement/cortex xdr flexible close reason mappings mirroring

Packs/ApiModules/Scripts/CoreIRApiModule/CoreIRApiModule.py

@@ -13,18 +13,18 @@
 
 XSOAR_RESOLVED_STATUS_TO_XDR = {
     'Other': 'resolved_other',
-    'Duplicate': 'resolved_duplicate',
+    'Duplicate': 'resolved_duplicate_incident',
     'False Positive': 'resolved_false_positive',
     'Resolved': 'resolved_true_positive',
-    'Resolved - Security Testing': 'resolved_security_testing',
+    'Security Testing': 'resolved_security_testing',
 }
 
 XDR_RESOLVED_STATUS_TO_XSOAR = {
     'resolved_known_issue': 'Other',
-    'resolved_duplicate': 'Duplicate',
+    'resolved_duplicate_incident': 'Duplicate',
     'resolved_false_positive': 'False Positive',
     'resolved_true_positive': 'Resolved',
-    'resolved_security_testing': 'Resolved - Security Testing',
+    'resolved_security_testing': 'Security Testing',
     'resolved_other': 'Other',
     'resolved_auto': 'Resolved'
 }
@@ -346,7 +346,7 @@ def get_endpoints(self,
         endpoints = reply.get('reply').get('endpoints', [])
         return endpoints
 
-    def set_endpoints_alias(self, filters: list[dict[str, str]], new_alias_name: str | None) -> dict:      # pragma: no cover
+    def set_endpoints_alias(self, filters: list[dict[str, str]], new_alias_name: str | None) -> dict:  # pragma: no cover
         """
         This func is used to set the alias name of an endpoint.
 
@@ -934,8 +934,7 @@ def get_endpoint_device_control_violations(self, endpoint_ids: list, type_of_vio
                                                ip_list: list, vendor: list, vendor_id: list, product: list,
                                                product_id: list,
                                                serial: list,
-                                               hostname: list, violation_ids: list, username: list) \
-            -> Dict[str, Any]:
+                                               hostname: list, violation_ids: list, username: list) -> Dict[str, Any]:
         arg_list = {'type': type_of_violation,
                     'endpoint_id_list': endpoint_ids,
                     'ip_list': ip_list,
@@ -1709,8 +1708,8 @@ def validate_args_scan_commands(args):
               'and without any other filters. This may cause performance issues.\n' \
               'To scan/abort scan some of the endpoints, please use the filter arguments.'
     if all_:
-        if endpoint_id_list or dist_name or gte_first_seen or gte_last_seen or lte_first_seen or lte_last_seen \
-                or ip_list or group_name or platform or alias or hostname:
+        if (endpoint_id_list or dist_name or gte_first_seen or gte_last_seen or lte_first_seen or lte_last_seen
+                or ip_list or group_name or platform or alias or hostname):
             raise Exception(err_msg)
     elif not endpoint_id_list and not dist_name and not gte_first_seen and not gte_last_seen \
             and not lte_first_seen and not lte_last_seen and not ip_list and not group_name and not platform \
@@ -2849,13 +2848,44 @@ def handle_outgoing_incident_owner_sync(update_args):
 
 def handle_user_unassignment(update_args):
     if ('assigned_user_mail' in update_args and update_args.get('assigned_user_mail') in ['None', 'null', '', None]) \
-            or ('assigned_user_pretty_name' in update_args
-                and update_args.get('assigned_user_pretty_name') in ['None', 'null', '', None]):
+        or ('assigned_user_pretty_name' in update_args
+            and update_args.get('assigned_user_pretty_name') in ['None', 'null', '', None]):
         update_args['unassign_user'] = 'true'
         update_args['assigned_user_mail'] = None
         update_args['assigned_user_pretty_name'] = None
 
 
+def resolve_xdr_close_reason(xsoar_close_reason: str) -> str:
+    """
+    Resolving XDR close reason from possible custom XSOAR->XDR close-reason mapping or default mapping.
+    :param xsoar_close_reason: XSOAR raw status/close reason e.g. 'False Positive'.
+    :return: XDR close-reason in snake_case format e.g. 'resolved_false_positive'.
+    """
+    # Initially setting the close reason according to the default mapping.
+    xdr_close_reason = XSOAR_RESOLVED_STATUS_TO_XDR.get(xsoar_close_reason, 'Other')
+    # Reading custom XSOAR->XDR close-reason mapping.
+    custom_xsoar_to_xdr_close_reason_mapping = comma_separated_mapping_to_dict(
+        demisto.params().get("custom_xsoar_to_xdr_close_reason_mapping")
+    )
+
+    # Overriding default close-reason mapping if there exists a custom one.
+    if xsoar_close_reason in custom_xsoar_to_xdr_close_reason_mapping:
+        xdr_close_reason_candidate = custom_xsoar_to_xdr_close_reason_mapping[xsoar_close_reason]
+        # Transforming resolved close-reason into snake_case format with known prefix to match XDR status format.
+        demisto.debug(
+            f"resolve_xdr_close_reason XSOAR->XDR custom close-reason exists, using {xsoar_close_reason}={xdr_close_reason}")
+        xdr_close_reason_candidate = "resolved_" + "_".join(xdr_close_reason_candidate.lower().split(" "))
+
+        if xdr_close_reason_candidate not in XDR_RESOLVED_STATUS_TO_XSOAR:
+            demisto.debug("Warning: Provided XDR close-reason does not exist. Using default XDR close-reason mapping. ")
+        else:
+            xdr_close_reason = xdr_close_reason_candidate
+    else:
+        demisto.debug(f"resolve_xdr_close_reason using default mapping {xsoar_close_reason}={xdr_close_reason}")
+
+    return xdr_close_reason
+
+
 def handle_outgoing_issue_closure(remote_args):
     incident_id = remote_args.remote_incident_id
     demisto.debug(f"handle_outgoing_issue_closure {incident_id=}")
@@ -2866,13 +2896,13 @@ def handle_outgoing_issue_closure(remote_args):
     # force closing remote incident only if:
     #   The XSOAR incident is closed
     #   and the remote incident isn't already closed
-    if remote_args.inc_status == 2 and \
-       current_remote_status not in XDR_RESOLVED_STATUS_TO_XSOAR and close_reason:
-
+    if remote_args.inc_status == 2 and current_remote_status not in XDR_RESOLVED_STATUS_TO_XSOAR and close_reason:
         if close_notes := update_args.get('closeNotes'):
             demisto.debug(f"handle_outgoing_issue_closure {incident_id=} {close_notes=}")
             update_args['resolve_comment'] = close_notes
-        update_args['status'] = XSOAR_RESOLVED_STATUS_TO_XDR.get(close_reason, 'Other')
+
+        xdr_close_reason = resolve_xdr_close_reason(close_reason)
+        update_args['status'] = xdr_close_reason
         demisto.debug(f"handle_outgoing_issue_closure Closing Remote incident {incident_id=} with status {update_args['status']}")
 
 
@@ -3148,7 +3178,6 @@ def get_script_code_command(client: CoreClient, args: Dict[str, str]) -> Tuple[s
     requires_polling_arg=False  # means it will always be default to poll, poll=true
 )
 def script_run_polling_command(args: dict, client: CoreClient) -> PollResult:
-
     if action_id := args.get('action_id'):
         response = client.get_script_execution_status(action_id)
         general_status = response.get('reply', {}).get('general_status') or ''
@@ -3740,7 +3769,6 @@ def create_request_filters(
 
 
 def args_to_request_filters(args):
-
     if set(args.keys()) & {  # check if any filter argument was provided
         'endpoint_id_list', 'dist_name', 'ip_list', 'group_name', 'platform', 'alias_name',
         'isolate', 'hostname', 'status', 'first_seen_gte', 'first_seen_lte', 'last_seen_gte', 'last_seen_lte'
@@ -3814,7 +3842,6 @@ def parse_risky_users_or_hosts(user_or_host_data: dict[str, Any],
                                score_header: str,
                                description_header: str
                                ) -> dict[str, Any]:
-
     reasons = user_or_host_data.get('reasons', [])
     return {
         id_header: user_or_host_data.get('id'),
@@ -4046,13 +4073,14 @@ def list_risky_users_or_host_command(client: CoreClient, command: str, args: dic
         ValueError: If the API connection fails.
 
     """
+
     def _warn_if_module_is_disabled(e: DemistoException) -> None:
         if (
-                e is not None
-                and e.res is not None
-                and e.res.status_code == 500
-                and 'No identity threat' in str(e)
-                and "An error occurred while processing XDR public API" in e.message
+            e is not None
+            and e.res is not None
+            and e.res.status_code == 500
+            and 'No identity threat' in str(e)
+            and "An error occurred while processing XDR public API" in e.message
         ):
             return_warning(f'Please confirm the XDR Identity Threat Module is enabled.\nFull error message: {e}', exit=True)
 

Packs/ApiModules/Scripts/CoreIRApiModule/CoreIRApiModule_test.py

@@ -9,7 +9,7 @@
 
 import demistomock as demisto
 from CommonServerPython import Common, tableToMarkdown, pascalToSpace, DemistoException
-from CoreIRApiModule import CoreClient, handle_outgoing_issue_closure
+from CoreIRApiModule import CoreClient, handle_outgoing_issue_closure, XSOAR_RESOLVED_STATUS_TO_XDR
 from CoreIRApiModule import add_tag_to_endpoints_command, remove_tag_from_endpoints_command, quarantine_files_command, \
     isolate_endpoint_command, list_user_groups_command, parse_user_groups, list_users_command, list_roles_command, \
     change_user_role_command, list_risky_users_or_host_command, enrich_error_message_id_group_role, get_incidents_command
@@ -18,7 +18,6 @@
     base_url='https://test_api.com/public_api/v1', headers={}
 )
 
-
 Core_URL = 'https://api.xdrurl.com'
 
 ''' HELPER FUNCTIONS '''
@@ -544,7 +543,7 @@ def test_allowlist_files_command_with_more_than_one_file(requests_mock):
     test_data = load_test_data('test_data/blocklist_allowlist_files_success.json')
     expected_command_result = {
         'CoreApiModule.allowlist.added_hashes.fileHash(val.fileHash == obj.fileHash)':
-        test_data['multi_command_args']['hash_list']
+            test_data['multi_command_args']['hash_list']
     }
     requests_mock.post(f'{Core_URL}/public_api/v1/hash_exceptions/allowlist/', json=test_data['api_response'])
 
@@ -859,13 +858,14 @@ def test_handle_outgoing_issue_closure_close_reason(mocker):
     """
     from CoreIRApiModule import handle_outgoing_issue_closure
     from CommonServerPython import UpdateRemoteSystemArgs
-    remote_args = UpdateRemoteSystemArgs({'delta': {'assigned_user_mail': 'None', 'closeReason': 'Resolved - Security Testing'},
+    remote_args = UpdateRemoteSystemArgs({'delta': {'assigned_user_mail': 'None', 'closeReason': 'Security Testing'},
                                           'status': 2, 'inc_status': 2, 'data': {'status': 'other'}})
     request_data_log = mocker.patch.object(demisto, 'debug')
     handle_outgoing_issue_closure(remote_args)
 
-    assert "handle_outgoing_issue_closure Closing Remote incident incident_id=None with status resolved_security_testing" in request_data_log.call_args[  # noqa: E501
-        0][0]
+    assert "handle_outgoing_issue_closure Closing Remote incident incident_id=None with status resolved_security_testing" in \
+           request_data_log.call_args[  # noqa: E501
+               0][0]
 
 
 def test_get_update_args_close_incident():
@@ -3168,8 +3168,8 @@ def test_endpoint_alias_change_command__no_filters(mocker):
         },
         {
             "err_msg": "An error occurred while processing XDR public API - No endpoint "
-            "was found "
-            "for creating the requested action",
+                       "was found "
+                       "for creating the requested action",
             "status_code": 500,
         },
         False,
@@ -3444,7 +3444,7 @@ def test_parse_user_groups(data: dict[str, Any], expected_results: list[dict[str
     [
         ({"group_names": "test"}, "Error: Group test was not found. Full error message: Group 'test' was not found"),
         ({"group_names": "test, test2"}, "Error: Group test was not found. Note: If you sent more than one group name, "
-         "they may not exist either. Full error message: Group 'test' was not found")
+                                         "they may not exist either. Full error message: Group 'test' was not found")
     ]
 )
 def test_list_user_groups_command_raise_exception(mocker, test_data: dict[str, str], excepted_error: str):
@@ -3844,3 +3844,67 @@ def test_handle_outgoing_issue_closure(args, expected_delta):
     remote_args = UpdateRemoteSystemArgs(args)
     handle_outgoing_issue_closure(remote_args)
     assert remote_args.delta == expected_delta
+
+
+@pytest.mark.parametrize('custom_mapping, expected_resolved_status',
+                         [
+                             ("Other=Other,Duplicate=Other,False Positive=False Positive,Resolved=True Positive",
+                              ["resolved_other", "resolved_other", "resolved_false_positive", "resolved_true_positive",
+                               "resolved_security_testing"]),
+
+                             ("Other=True Positive,Duplicate=Other,False Positive=False Positive,Resolved=True Positive",
+                              ["resolved_true_positive", "resolved_other", "resolved_false_positive",
+                               "resolved_true_positive", "resolved_security_testing"]),
+
+                             ("Duplicate=Other", ["resolved_other", "resolved_other", "resolved_false_positive",
+                                                  "resolved_true_positive", "resolved_security_testing"]),
+
+                             # Expecting default mapping to be used when no mapping provided.
+                             ("", list(XSOAR_RESOLVED_STATUS_TO_XDR.values())),
+
+                             # Expecting default mapping to be used when improper mapping is provided.
+                             ("Duplicate=RANDOM1, Other=Random2", list(XSOAR_RESOLVED_STATUS_TO_XDR.values())),
+
+                             ("Random1=Duplicate Incident", list(XSOAR_RESOLVED_STATUS_TO_XDR.values())),
+
+                             # Expecting default mapping to be used when improper mapping *format* is provided.
+                             ("Duplicate=Other False Positive=Other", list(XSOAR_RESOLVED_STATUS_TO_XDR.values())),
+
+                             # Expecting default mapping to be used for when improper key-value pair *format* is provided.
+                             ("Duplicate=Other, False Positive=Other True Positive=Other, Other=True Positive",
+                              ["resolved_true_positive", "resolved_other", "resolved_false_positive",
+                               "resolved_true_positive", "resolved_security_testing"]),
+
+                         ],
+                         ids=["case-1", "case-2", "case-3", "empty-case", "improper-input-case-1", "improper-input-case-2",
+                              "improper-input-case-3", "improper-input-case-4"]
+                         )
+def test_xsoar_to_xdr_flexible_close_reason_mapping(capfd, mocker, custom_mapping, expected_resolved_status):
+    """
+    Given:
+        - A custom XSOAR->XDR close-reason mapping
+        - Expected resolved XDR status according to the custom mapping.
+    When
+        - Handling outgoing issue closure (handle_outgoing_issue_closure(...) executed).
+    Then
+        - The resolved XDR statuses match the expected statuses for all possible XSOAR close-reasons.
+    """
+    from CoreIRApiModule import handle_outgoing_issue_closure
+    from CommonServerPython import UpdateRemoteSystemArgs
+
+    mocker.patch.object(demisto, 'params', return_value={"mirror_direction": "Both",
+                                                         "custom_xsoar_to_xdr_close_reason_mapping": custom_mapping})
+
+    all_xsoar_close_reasons = XSOAR_RESOLVED_STATUS_TO_XDR.keys()
+    for i, close_reason in enumerate(all_xsoar_close_reasons):
+        remote_args = UpdateRemoteSystemArgs({'delta': {'closeReason': close_reason},
+                                              'status': 2,
+                                              'inc_status': 2,
+                                              'data': {'status': 'other'}
+                                              })
+        # Overcoming expected non-empty stderr test failures (Errors are submitted to stderr when improper mapping is provided).
+        with capfd.disabled():
+            handle_outgoing_issue_closure(remote_args)
+
+        assert remote_args.delta.get('status')
+        assert remote_args.delta['status'] == expected_resolved_status[i]

Packs/Base/ReleaseNotes/1_33_38.md

@@ -0,0 +1,6 @@
+
+#### Scripts
+
+##### CommonServerPython
+
+Added a utility function `comma_separated_mapping_to_dict` that gets a comma-separated mapping `key1=value1,key2=value2,...` and transforms it into a dictionary object.

Packs/Base/Scripts/CommonServerPython/CommonServerPython.py

@@ -11751,6 +11751,47 @@ def data_error_handler(res):
         demisto.updateModuleHealth({'{data_type}Pulled'.format(data_type=data_type): data_size})
 
 
+def comma_separated_mapping_to_dict(raw_text):
+    """
+     Transforming a textual comma-separated mapping into a dictionary object.
+
+    :type raw_text: ``str``
+    :param raw_text: Comma-separated mapping e.g ('key1=value1', 'key2=value2', ...)
+
+    :rtype: ``dict``
+    :return: Validated dictionary of the raw mapping e.g {'key1': 'value1', 'key2': 'value2', ...}
+    """
+    demisto.debug("comma_separated_mapping_to_dict "
+                  ">> Resolving comma-separated input mapping: {raw_text}".format(raw_text=raw_text))
+
+    mapping_dict = {}  # type: Dict[str, str]
+    # If a proper mapping was not provided, return an empty dict.
+    if not raw_text:
+        return mapping_dict
+
+    key_value_pairs = raw_text.split(',')
+
+    for pair in key_value_pairs:
+        # Trimming trailing whitespace
+        pair = pair.strip()
+
+        try:
+            key, value = pair.split('=')
+        except ValueError:
+            demisto.error("Error: Invalid mapping was provided. "
+                          "Expected comma-separated mapping of format `key1=value1, key2=value2, ...`")
+            key = value = ''
+
+        if key in mapping_dict:
+            demisto.debug(
+                "comma_separated_mapping_to_dict "
+                "Warning: duplicate key provided for {key}: using latter value: {value}".format(key=key, value=value)
+            )
+        mapping_dict[key] = value
+    demisto.debug("comma_separated_mapping_to_dict << Resolved mapping: {mapping_dict}".format(mapping_dict=mapping_dict))
+    return mapping_dict
+
+
 ###########################################
 #     DO NOT ADD LINES AFTER THIS ONE     #
 ###########################################

Packs/Base/pack_metadata.json

@@ -2,7 +2,7 @@
     "name": "Base",
     "description": "The base pack for Cortex XSOAR.",
     "support": "xsoar",
-    "currentVersion": "1.33.37",
+    "currentVersion": "1.33.38",
     "author": "Cortex XSOAR",
     "serverMinVersion": "6.0.0",
     "url": "https://www.paloaltonetworks.com/cortex",

Packs/Core/ReleaseNotes/3_0_22.md

@@ -0,0 +1,6 @@
+
+#### Integrations
+
+##### Investigation & Response
+
+Added support for flexible close-reason mapping in `handle_outgoing_issue_closure` in `CoreIRApiModule`.  Does not affect this module.

Packs/Core/pack_metadata.json

@@ -2,7 +2,7 @@
     "name": "Core - Investigation and Response",
     "description": "Automates incident response",
     "support": "xsoar",
-    "currentVersion": "3.0.21",
+    "currentVersion": "3.0.22",
     "author": "Cortex XSOAR",
     "url": "https://www.paloaltonetworks.com/cortex",
     "email": "",