Add WorkflowDefinitionsStatus endpoint and corresponding method in Workflows class

Author: dweissbacher

.github/wordlist.txt

@@ -1283,6 +1283,7 @@ shubham
 WorkflowUpdateHumanInputV
 WorkflowGetHumanInputV
 WorkflowDefinitionsCreate
+WorkflowDefinitionsStatus
 WorkflowDefinitionsUpdate
 WorkflowDefinitionsImport
 WorkflowDefinitionsExport

src/falconpy/_constant/__init__.py

@@ -57,7 +57,7 @@
     "setContentUpdatePoliciesPrecedence", "setDeviceControlPoliciesPrecedence",
     "setFirewallPoliciesPrecedence", "setPreventionPoliciesPrecedence", "signalChangesExternal",
     "setRTResponsePoliciesPrecedence", "setSensorUpdatePoliciesPrecedence", "GetDeviceDetails",
-    "CreateSavedSearchesDeployV1", "cancel-scans", "get-rules-get"
+    "CreateSavedSearchesDeployV1", "cancel-scans", "get-rules-get", "WorkflowDefinitionsStatus"
 ]
 MOCK_OPERATIONS: List[str] = [
     "GetImageAssessmentReport", "DeleteImageDetails", "ImageMatchesPolicy"

src/falconpy/_endpoint/_workflows.py

@@ -198,6 +198,31 @@
       }
     ]
   ],
+  [
+    "WorkflowDefinitionsStatus",
+    "POST",
+    "/workflows/entities/definition-actions/v1",
+    "Enable or disable a workflow definition, or stop all executions for a definition."
+    "When a definition is disabled it will not execute against any new trigger events.",
+    "workflows",
+    [
+      {
+        "type": "string",
+        "description": "Specify one of these actions:\n  enable - enable the workflow(s) specified in ids\n  "
+        "disable - disable the workflow(s) specified in ids.\n cancle - cancel all in-flight executions for "
+        "the workflow specified in ids",
+        "name": "action_name",
+        "in": "query",
+        "required": True
+      },
+      {
+        "description": "IDs of workflow definitions",
+        "name": "body",
+        "in": "body",
+        "required": True
+      }
+    ]
+  ],
   [
     "WorkflowDefinitionsExport",
     "GET",

src/falconpy/workflows.py

@@ -42,7 +42,8 @@
     force_default,
     process_service_request,
     handle_single_argument,
-    generate_error_result
+    generate_error_result,
+    args_to_params
     )
 from ._payload import (
     simple_action_parameter,
@@ -840,6 +841,61 @@ def provision(self: object, body: dict = None, **kwargs) -> Union[Dict[str, Unio
             body=body
             )
 
+    @force_default(defaults=["parameters", "body"], default_types=["dict"])
+    def set_workflow_definition_status(self: object,
+                                       body: dict = None,
+                                       parameters: dict = None,
+                                       **kwargs
+                                       ) -> Union[Dict[str, Union[int, dict]], Result]:
+        """Enable or disable a workflow definition, or stop all executions for a definition.
+
+        When a definition is disabled it will not execute against any new trigger events.
+
+        Keyword arguments:
+        action_name -- action to perform, 'enable', 'disable', or 'cancel'.
+        body -- full body payload, not required if ids are provided as keyword.
+                You must use body if you are going to specify action_parameters.
+                {
+                    "ids": [
+                        "string"
+                    ]
+                }
+        ids -- IDs of workflow definitions to perform the action against. String or list of strings.
+        parameters - full parameters payload, not required if action_name is provide as a keyword.
+
+        This method only supports keywords for providing arguments.
+
+        Returns: dict object containing API response.
+
+        HTTP Method: POST
+
+        Swagger URL
+        https://assets.falcon.crowdstrike.com/support/api/swagger.html#/workflows/WorkflowDefinitionsAction
+        """
+        if not body:
+            body = generic_payload_list(submitted_keywords=kwargs, payload_value="ids")
+
+        _allowed_actions = ['enable', 'disable', 'cancel']
+        operation_id = "WorkflowDefinitionsStatus"
+        parameter_payload = args_to_params(parameters, kwargs, Endpoints, operation_id)
+        action_name = parameter_payload.get("action_name", "Not Specified")
+        # Only process allowed actions
+        if action_name.lower() in _allowed_actions:
+            returned = process_service_request(
+                calling_object=self,
+                endpoints=Endpoints,
+                operation_id=operation_id,
+                body=body,
+                keywords=kwargs,
+                params=parameters,
+                body_validator={"ids": list} if self.validate_payloads else None,
+                body_required=["ids"] if self.validate_payloads else None
+                )
+        else:
+            returned = generate_error_result("Invalid value specified for action_name parameter.")
+
+        return returned
+
     # These method names align to the operation IDs in the API but
     # do not conform to snake_case / PEP8 and are defined here for
     # backwards compatibility / ease of use purposes
@@ -851,6 +907,7 @@ def provision(self: object, body: dict = None, **kwargs) -> Union[Dict[str, Unio
     WorkflowDefinitionsExport = export_definition
     WorkflowDefinitionsImport = import_definition
     WorkflowDefinitionsUpdate = update_definition
+    WorkflowDefinitionsStatus = set_workflow_definition_status
     WorkflowExecute = execute
     WorkflowExecuteInternal = execute_internal
     WorkflowMockExecute = mock_execute

tests/test_workflows.py

@@ -43,6 +43,7 @@ def run_all_tests(self):
             "WorkflowDefinitionsImport4": falcon.import_definition(validate_only=True, data_file="tests/test.yml", name="workflow_name"),
             "WorkflowDefinitionsImport4": falcon.import_definition(validate_only=True, data_file=binary_example, name="workflow_name"),
             "WorkflowDefinitionsUpdate": falcon.update_definition(change_log="testing"),
+            "WorkflowDefinitionsStatus": falcon.set_workflow_definition_status(ids="1234567", action_name="enable"),
             "WorkflowGetHumanInputV1": falcon.get_human_input(ids="1234567"),
             "WorkflowUpdateHumanInputV1": falcon.update_human_input(input="whatever", note="whatever"),
             "WorkflowActivitiesContentCombined": falcon.search_activities_content(limit=1)