Adding Decommissioned operations mapping and error result

Author: alhumaw

src/falconpy/_endpoint/__init__.py

@@ -65,6 +65,7 @@
 from .deprecated import _user_management_deprecated
 from .deprecated import _deprecated_operation_mapping
 from .deprecated import _deprecated_class_mapping
+from .deprecated import _decommissioned_operations
 from .deprecated import _zero_trust_assessment_deprecated
 
 from ._admission_control_policies import _admission_control_policies_endpoints
@@ -346,6 +347,7 @@
 # Mapping of manually deprecated endpoints
 operation_deprecation_mapping = _deprecated_operation_mapping
 class_deprecation_mapping = _deprecated_class_mapping
+decommissioned_operations = _decommissioned_operations
 
 # api_endpoints contains all endpoints, production and deprecated
 api_endpoints.extend(deprecated_endpoints)

src/falconpy/_endpoint/deprecated/__init__.py

@@ -76,6 +76,7 @@
 from ._workflows import _workflows_endpoints
 from ._zero_trust_assessment import _zero_trust_assessment_endpoints
 from ._mapping import _deprecated_op_mapping, _deprecated_cls_mapping
+from ._mapping import _decommissioned_operations  # noqa: F401
 from ._certificate_based_exclusions import _certificate_based_exclusions_endpoints
 
 _admission_control_policies_deprecated = _admission_control_policies_endpoints

src/falconpy/_endpoint/deprecated/_mapping.py

@@ -72,3 +72,8 @@
     "Iocs": "IOC",
     "Detects": "Alerts"
 }
+
+_decommissioned_operations = [
+    "AggregateFCIncidents",
+    "QueryIncidentIdsByFilter",
+]

src/falconpy/_util/_functions.py

@@ -52,7 +52,7 @@
 import urllib3
 from urllib3.exceptions import InsecureRequestWarning
 from .._api_request import APIRequest
-from .._endpoint import operation_deprecation_mapping
+from .._endpoint import operation_deprecation_mapping, decommissioned_operations
 from .._enum import BaseURL, ContainerBaseURL
 from .._constant import (
     PREFER_NONETYPE,
@@ -203,6 +203,8 @@ def factory(*args, **kwargs):
                 # Should only receive this in pythonic mode
                 raise api_error
             except (SDKError, InvalidMethod) as bad_sdk_command:
+                if getattr(bad_sdk_command, 'code', None) == 410:
+                    raise bad_sdk_command
                 created = bad_sdk_command.result
             return created
         return factory
@@ -689,6 +691,13 @@ def process_service_request(calling_object: ServiceClass,  # pylint: disable=R09
     # Log the operation ID if we have logging enabled.
     if calling_object.log:
         calling_object.log.debug("OPERATION: %s", operation_id)
+    # Intercept decommissioned operations before making any API call.
+    if operation_id in decommissioned_operations:
+        _msg = (f"The {operation_id} operation has been decommissioned "
+                f"by CrowdStrike and is no longer available.")
+        if calling_object.log:
+            calling_object.log.warning(_msg)
+        return generate_error_result(message=_msg, code=410, caller=calling_object)
     # We have to create our headers dictionary first, as authentication happens here.
     # For scenarios where cloud region autodiscovery is leveraged, we cannot create
     # the target URL for our call to requests until we know our correct base_url.