[DEPLOY] Version 0.4.4 Update - Sensor Download Service Class, Fixes

.github/labeler.yml

@@ -15,7 +15,8 @@ pipeline:
 - .github/ISSUE_TEMPLATE/*
 
 unit testing:
-- any: ['tests/*', 'util/*']
+- tests/*
+- util/*
 
 code samples:
 - samples/*.py

.github/workflows/bandit.yml

@@ -1,10 +1,14 @@
 name: Bandit
 on:
   push:
+    paths:
+      - '**.py'
     branches: 
       - main
       - 'ver_*'
   pull_request:
+    paths:
+      - '**.py'
     branches: 
       - main
       - 'ver_*'

.github/workflows/dev-deploy.yml

@@ -2,6 +2,8 @@ name: Test Package Build and Deploy
 
 on:
   push:
+    paths:
+      - '**.py'
     branches:
       - main
 

.github/workflows/linting.yml

@@ -1,10 +1,14 @@
 name: Flake8
 on:
   push:
+    paths:
+      - '**.py'
     branches:
       - main
       - 'ver_*'
   pull_request:
+    paths:
+      - '**.py'
     branches: 
       - main
       - 'ver_*'

.github/workflows/unit_testing.yml

@@ -1,8 +1,12 @@
 name: Python package
 on:
   push:
+    paths:
+      - '**.py'
     branches: [ main ]
   pull_request:
+    paths:
+      - '**.py'
     branches: 
       - main
       - 'ver_*'

.gitignore

@@ -8,5 +8,8 @@ env/
 .vscode/
 build/
 dist
-falconpy.egg-info/
-__pycache__/
+*.egg-info/
+sensor_downloads/
+__pycache__/
+.env
+.coverage

README.md

@@ -57,7 +57,7 @@ $ python3 -m pip uninstall crowdstrike-falconpy
 | [CrowdStrike Real Time Response (RTR) APIs](https://falcon.crowdstrike.com/support/documentation/90/real-time-response-apis) | [real_time_response.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/real_time_response.py) |
 | [CrowdStrike Realtime Response (RTR) Administration API](https://falcon.crowdstrike.com/support/documentation/90/real-time-response-apis) | [real_time_response_admin.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/real_time_response_admin.py) |
 | CrowdStrike Sample Uploads API | [sample_uploads.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/sample_uploads.py) |
-| [CrowdStrike Sensor Download APIs](https://falcon.crowdstrike.com/support/documentation/109/sensor-download-apis) | *Coming Soon* |
+| [CrowdStrike Sensor Download APIs](https://falcon.crowdstrike.com/support/documentation/109/sensor-download-apis) | [sensor_download.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/sensor_download.py) |
 | [CrowdStrike Spotlight APIs](https://falcon.crowdstrike.com/support/documentation/98/spotlight-apis) | [spotlight_vulnerabilities.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/spotlight_vulnerabilities.py) |
 | [CrowdStrike User and Roles API](https://falcon.crowdstrike.com/support/documentation/87/users-and-roles-apis) | [user_management.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/user_management.py) | 
 | [Falcon Discover for Cloud and Containers - AWS Accounts APIs](https://falcon.crowdstrike.com/support/documentation/91/discover-for-aws-apis) | [cloud_connect_aws.py](https://github.com/CrowdStrike/falconpy/blob/main/src/falconpy/cloud_connect_aws.py) |

samples/real_time_response/quarantine_hosts.py

@@ -83,9 +83,6 @@
         print(f"\n[+] Lifting Containment: {hostname}\n")
 
     # Perform the requested action
-    # TODO: Get rid of action_name="contain" once bug is resolved
-    # BUG: https://github.com/CrowdStrike/falconpy/issues/114
-    response = falcon.PerformActionV2(parameters=PARAMS, body=BODY,
-                                      action_name="contain")
+    response = falcon.PerformActionV2(parameters=PARAMS, body=BODY)
     # Output the result
     print(json.dumps(response, indent=4))

src/falconpy/README.md

@@ -22,6 +22,7 @@ This folder contains the FalconPy project, a Python 3 interface handler for the
 + `real_time_response_admin.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/real-time-response-admin
 + `real_time_response.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/real-time-response
 + `sample_uploads.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/sample-uploads
++ `sensor_download.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/sensor-download
 + `sensor_update_policy.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/sensor-update-policies
 + `spotlight_vulnerabilities.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/spotlight-vulnerabilities
 + `user_management.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/user-management
@@ -35,7 +36,6 @@ This folder contains the FalconPy project, a Python 3 interface handler for the
 + `ml_exclusions.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/ml-exclusions
 + `sensor_visibility_exclusions.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/sensor-visibility-exclusions
 + `quick_scan.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/quick-scan
-+ `sensor_download.py` https://assets.falcon.crowdstrike.com/support/api/swagger.html#/sensor-download
 
 ## The Uber Class
 ### A single class to interface with the entire API

src/falconpy/_util.py

@@ -190,3 +190,7 @@ def perform_request(method: str = "", endpoint: str = "", headers: dict = None,
 def generate_error_result(message: str = "An error has occurred. Check your payloads and try again.", code: int = 500) -> dict:
     """ Normalized error messaging handler. """
     return Result()(status_code=code, headers={}, body={"errors": [{"message": f"{message}"}], "resources": []})
+
+
+def generate_ok_result(message: str = "Request returned with success", code: int = 200) -> dict:
+    return Result()(status_code=code, headers={}, body={"message": message, "resources": []})

src/falconpy/_version.py

@@ -36,7 +36,7 @@
 
 For more information, please refer to <https://unlicense.org>
 """
-_version = '0.4.3'
+_version = '0.4.4'
 _maintainer = 'Joshua Hiller'
 _author = 'CrowdStrike'
 _author_email = '[email protected]'

src/falconpy/device_control_policies.py

@@ -80,12 +80,14 @@ def queryCombinedDeviceControlPolicies(self: object, parameters: dict = {}) -> d
                                    )
         return returned
 
-    def performDeviceControlPoliciesAction(self: object, parameters: dict, body: dict, action_name: str) -> dict:
+    def performDeviceControlPoliciesAction(self: object, parameters: dict, body: dict, action_name: str = None) -> dict:
         """ Search for Device Control Policies in your environment by providing an FQL filter
             and paging details. Returns a set of Device Control Policies which match the filter criteria.
         """
         # [POST] https://assets.falcon.crowdstrike.com/support/api/swagger.html#
         #        ...    /device-control-policies/performDeviceControlPoliciesAction
+        if "action_name" in parameters:
+            action_name = parameters["action_name"].lower()
         ALLOWED_ACTIONS = ['add-host-group', 'disable', 'enable', 'remove-host-group']
         if action_name.lower() in ALLOWED_ACTIONS:
             FULL_URL = self.base_url+'/policy/combined/device-control/v1'

src/falconpy/firewall_policies.py

@@ -79,10 +79,12 @@ def queryCombinedFirewallPolicies(self: object, parameters: dict = {}) -> dict:
                                    )
         return returned
 
-    def performFirewallPoliciesAction(self: object, parameters: dict, body: dict, action_name: str) -> dict:
+    def performFirewallPoliciesAction(self: object, parameters: dict, body: dict, action_name: str = None) -> dict:
         """ Perform the specified action on the Firewall Policies specified in the request. """
         # [POST] https://assets.falcon.crowdstrike.com/support/api/swagger.html#
         #        ...    /firewall-policies/performFirewallPoliciesAction
+        if "action_name" in parameters:
+            action_name = parameters["action_name"].lower()
         ALLOWED_ACTIONS = ['add-host-group', 'disable', 'enable', 'remove-host-group']
         if action_name.lower() in ALLOWED_ACTIONS:
             FULL_URL = self.base_url+'/policy/entities/firewall-actions/v1'

src/falconpy/host_group.py

@@ -78,9 +78,11 @@ def queryCombinedHostGroups(self: object, parameters: dict = {}) -> dict:
                                    )
         return returned
 
-    def performGroupAction(self: object, parameters: dict, body: dict, action_name: str) -> dict:
+    def performGroupAction(self: object, parameters: dict, body: dict, action_name: str = None) -> dict:
         """ Perform the specified action on the Host Groups specified in the request. """
         # [POST] https://assets.falcon.crowdstrike.com/support/api/swagger.html#/host-group/performGroupAction
+        if "action_name" in parameters:
+            action_name = parameters["action_name"].lower()
         ALLOWED_ACTIONS = ['add-hosts', 'remove-hosts']
         if action_name.lower() in ALLOWED_ACTIONS:
             FULL_URL = self.base_url+'/devices/entities/host-group-actions/v1'

src/falconpy/hosts.py

@@ -44,10 +44,12 @@ class Hosts(ServiceClass):
     """ The only requirement to instantiate an instance of this class
         is a valid token provided by the Falcon API SDK OAuth2 class.
     """
-    def PerformActionV2(self: object, parameters: dict, body: dict, action_name: str) -> dict:
+    def PerformActionV2(self: object, parameters: dict, body: dict, action_name: str = None) -> dict:
         """ Take various actions on the hosts in your environment.
             Contain or lift containment on a host. Delete or restore a host.
         """
+        if "action_name" in parameters:
+            action_name = parameters["action_name"].lower()
         # [POST] https://assets.falcon.crowdstrike.com/support/api/swagger.html#/hosts/PerformActionV2
         ALLOWED_ACTIONS = ['contain', 'lift_containment', 'hide_host', 'unhide_host']
         if action_name.lower() in ALLOWED_ACTIONS:

src/falconpy/prevention_policy.py

@@ -80,10 +80,12 @@ def queryCombinedPreventionPolicies(self: object, parameters: dict = {}) -> dict
                                    )
         return returned
 
-    def performPreventionPoliciesAction(self: object, parameters: dict, body: dict, action_name: str) -> dict:
+    def performPreventionPoliciesAction(self: object, parameters: dict, body: dict, action_name: str = None) -> dict:
         """ Perform the specified action on the Prevention Policies specified in the request. """
         # [POST] https://assets.falcon.crowdstrike.com/support/api/swagger.html#
         #        ...    /prevention-policies/performPreventionPoliciesAction
+        if "action_name" in parameters:
+            action_name = parameters["action_name"].lower()
         ALLOWED_ACTIONS = ['add-host-group', 'disable', 'enable', 'remove-host-group']
         if action_name.lower() in ALLOWED_ACTIONS:
             FULL_URL = self.base_url+'/policy/entities/prevention-actions/v1'

src/falconpy/sensor_download.py

@@ -0,0 +1,96 @@
+from ._util import service_request, parse_id_list, generate_ok_result
+from ._service_class import ServiceClass
+
+import os
+
+
+class Sensor_Download(ServiceClass):
+
+    def GetCombinedSensorInstallersByQuery(self: object, parameters: dict = {}) -> dict:
+        """
+        retrieve all metadata for installers from provided query
+        """
+        FULL_URL = self.base_url+'/sensors/combined/installers/v1'
+        HEADERS = self.headers
+        PARAMS = parameters
+        returned = service_request(caller=self,
+                                   method="GET",
+                                   endpoint=FULL_URL,
+                                   params=PARAMS,
+                                   headers=HEADERS,
+                                   verify=self.ssl_verify
+                                   )
+        return returned
+
+    def DownloadSensorInstallerById(self: object,
+                                    parameters: dict,
+                                    file_name: str = None,
+                                    download_path: str = None
+                                    ) -> object:
+        """
+        download the sensor by the sha256 into the specified directory.
+        the path will be created for the user if it does not already exist
+        """
+        FULL_URL = self.base_url+"/sensors/entities/download-installer/v1"
+        HEADERS = self.headers
+        PARAMS = parameters
+        returned = service_request(caller=self,
+                                   method="GET",
+                                   endpoint=FULL_URL,
+                                   headers=HEADERS,
+                                   params=PARAMS,
+                                   verify=self.ssl_verify
+                                   )
+        if file_name and download_path and isinstance(returned, bytes):
+            os.makedirs(download_path, exist_ok=True)
+            # write the newly downloaded sensor into the aforementioned directory with provided file name
+            with open(os.path.join(download_path, file_name), "wb") as sensor:
+                sensor.write(returned)
+            returned = generate_ok_result(message="Download successful")
+        return returned
+
+    def GetSensorInstallersEntities(self: object, ids: list or str) -> object:
+        """
+        For a given list of SHA256's, retrieve the metadata for each installer
+        such as the release_date and version among other fields
+        """
+        ID_LIST = str(parse_id_list(ids)).replace(",", "&ids=")
+        FULL_URL = self.base_url+'/sensors/entities/installers/v1?ids={}'.format(ID_LIST)
+        HEADERS = self.headers
+        returned = service_request(caller=self,
+                                   method="GET",
+                                   endpoint=FULL_URL,
+                                   headers=HEADERS,
+                                   verify=self.ssl_verify
+                                   )
+        return returned
+
+    def GetSensorInstallersCCIDByQuery(self: object) -> dict:
+        """
+        retrieve the CID for the current oauth environment
+        """
+        FULL_URL = self.base_url+'/sensors/queries/installers/ccid/v1'
+        HEADERS = self.headers
+        returned = service_request(caller=self,
+                                   method="GET",
+                                   endpoint=FULL_URL,
+                                   headers=HEADERS,
+                                   verify=self.ssl_verify
+                                   )
+        return returned
+
+    def GetSensorInstallersByQuery(self: object, parameters: dict = {}) -> dict:
+        """
+        retrieve a list of SHA256 for installers based on the filter
+        """
+        FULL_URL = self.base_url+'/sensors/queries/installers/v1'
+        HEADERS = self.headers
+        PARAMS = parameters
+        returned = service_request(caller=self,
+                                   method="GET",
+                                   endpoint=FULL_URL,
+                                   params=PARAMS,
+                                   headers=HEADERS,
+                                   verify=self.ssl_verify
+                                   )
+        return returned

src/falconpy/sensor_update_policy.py

@@ -132,13 +132,15 @@ def queryCombinedSensorUpdatePoliciesV2(self: object, parameters: dict = {}) ->
                                    )
         return returned
 
-    def performSensorUpdatePoliciesAction(self: object, parameters: dict, body: dict, action_name: str) -> dict:
+    def performSensorUpdatePoliciesAction(self: object, parameters: dict, body: dict, action_name: str = None) -> dict:
         """ Perform the specified action on the Sensor Update Policies specified in the request. """
         # [POST] https://assets.falcon.crowdstrike.com/support/api/swagger.html#
         #        ...    /sensor-update-policies/performSensorUpdatePoliciesAction
+        if "action_name" in parameters:
+            action_name = parameters["action_name"].lower()
         ALLOWED_ACTIONS = ['add-host-group', 'disable', 'enable', 'remove-host-group']
         if action_name.lower() in ALLOWED_ACTIONS:
-            FULL_URL = self.base_url+'/policy/entities/sensor-update-actions/v1?action_name={}'.format(action_name.lower())
+            FULL_URL = self.base_url+'/policy/entities/sensor-update-actions/v1'
             HEADERS = self.headers
             BODY = body
             PARAMS = parameters

tests/test_device_control_policies.py

@@ -57,7 +57,7 @@ def serviceDeviceControlPolicies_GenerateErrors(self):
             ["queryCombinedDeviceControlPolicyMembers", ""],
             ["queryCombinedDeviceControlPolicies", ""],
             ["performDeviceControlPoliciesAction", "body={}, parameters={}, action_name='enable'"],
-            ["performDeviceControlPoliciesAction", "body={}, parameters={}, action_name='PooF'"],
+            ["performDeviceControlPoliciesAction", "body={}, parameters={'action_name':'PooF'}"],
             ["setDeviceControlPoliciesPrecedence", "body={}"],
             ["getDeviceControlPolicies", "ids='12345678'"],
             ["createDeviceControlPolicies", "body={}"],

tests/test_firewall_policies.py

@@ -34,7 +34,7 @@ def serviceFirewall_GenerateErrors(self):
             ["queryCombinedFirewallPolicyMembers", ""],
             ["queryCombinedFirewallPolicies", ""],
             ["performFirewallPoliciesAction", "action_name='enable', body={}, parameters={}"],
-            ["performFirewallPoliciesAction", "action_name='make-it-go-boom', body={}, parameters={}"],
+            ["performFirewallPoliciesAction", "body={}, parameters={'action_name':'PooF'}"],
             ["setFirewallPoliciesPrecedence", "body={}"],
             ["getFirewallPolicies", "ids='12345678'"],
             ["createFirewallPolicies", "body={}"],

tests/test_host_group.py

@@ -57,7 +57,7 @@ def serviceHostGroup_GenerateErrors(self):
             ["queryCombinedGroupMembers", ""],
             ["queryCombinedHostGroups", ""],
             ["performGroupAction", "action_name='add-hosts', body={}, parameters={}"],
-            ["performGroupAction", "action_name='iLikeErrorMessages', body={}, parameters={}"],
+            ["performGroupAction", "body={}, parameters={'action_name':'PooF'}"],
             ["getHostGroups", "ids='12345678'"],
             ["createHostGroups", "body={}"],
             ["deleteHostGroups", "ids='12345678'"],

tests/test_hosts.py

@@ -108,7 +108,7 @@ def serviceHosts_GenerateErrors(self):
         errorChecks = True
         commandList = [
             ["PerformActionV2","body={}, action_name='unhide_host', parameters={}"],
-            ["PerformActionV2","body={}, action_name='KErrrPOW', parameters={}"],
+            ["PerformActionV2","body={}, parameters={'action_name':'PooF'}"],
             ["GetDeviceDetails", "ids='12345678'"],
             ["QueryHiddenDevices", ""],
             ["QueryDevicesByFilterScroll", ""],

tests/test_prevention_policy.py

@@ -57,7 +57,7 @@ def servicePrevent_GenerateErrors(self):
             ["queryCombinedPreventionPolicyMembers", ""],
             ["queryCombinedPreventionPolicies", ""],
             ["performPreventionPoliciesAction", "body={}, action_name='enable', parameters={}"],
-            ["performPreventionPoliciesAction", "body={}, action_name='kaB00M', parameters={}"],
+            ["performPreventionPoliciesAction", "body={}, parameters={'action_name':'PooF'}"],
             ["setPreventionPoliciesPrecedence", "body={}"],
             ["getPreventionPolicies", "ids='12345678'"],
             ["createPreventionPolicies", "body={}"],