feat: added scan and status methods to Inventory (#111)

* fix: modified create policy wrapper to use type bool instead of int for 'enabled' field

* feat: added `scan` and `status` methods to `Inventory`

* minor linting and pytest changes for inventory

Author: iancrichardson

examples/example_inventory.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+"""
+Example script showing how to use the LaceworkClient class.
+"""
+
+import logging
+
+from dotenv import load_dotenv
+from laceworksdk import LaceworkClient
+
+logging.basicConfig(level=logging.DEBUG)
+
+load_dotenv()
+
+if __name__ == "__main__":
+
+    # Instantiate a LaceworkClient instance
+    lacework_client = LaceworkClient()
+
+    # Inventory API
+
+    # Scan each CSP (cloud service provider)
+
+    for csp in ["AWS", "GCP", "Azure"]:
+        inventory_scan = lacework_client.inventory.scan(csp=csp)
+
+        inventory_scan_status = lacework_client.inventory.status(csp=csp)

examples/example_query_policy.py

@@ -29,7 +29,7 @@
     query_response = lacework_client.queries.create(
         evaluator_id="Cloudtrail",
         query_id=QUERY_ID,
-        query_text=f"""{{
+        query_text="""{{
             source {{CloudTrailRawEvents e}}
             filter {{EVENT_SOURCE = 'iam.amazonaws.com' AND
                      EVENT:userIdentity.name::String NOT LIKE 'Terraform-Service-Acct'}}

examples/example_syscall_query_policy.py

@@ -29,7 +29,7 @@
     # Create a Query
     query_response = lacework_client.queries.create(
         query_id=QUERY_ID,
-        query_text=f"""{{
+        query_text="""{{
           source {{
                 LW_HA_SYSCALLS_FILE
           }}

laceworksdk/api/v2/inventory.py

@@ -18,3 +18,42 @@ def __init__(self, session):
         """
 
         super().__init__(session, "Inventory")
+
+    def scan(self,
+             csp,
+             **request_params):
+        """
+        A method to issue Resource Inventory scans.
+
+        :param csp: A string representing the cloud service provider to run the scan on (i.e. AWS, Azure, GCP).
+        :param request_params: Additional request parameters.
+            (provides support for parameters that may be added in the future)
+
+        :return response json
+        """
+
+        params = self.build_dict_from_items(
+            csp=csp
+        )
+
+        response = self._session.post(self.build_url(action="scan"), params=params)
+
+        return response.json()
+
+    def status(self,
+               csp):
+        """
+        A method to get the status of a Resource Inventory scan.
+
+        :param csp: A string representing the cloud service provider retrieve the scan status from (i.e. AWS, Azure, GCP).
+
+        :return response json
+        """
+
+        params = self.build_dict_from_items(
+            csp=csp
+        )
+
+        response = self._session.get(self.build_url(action="scan"), params=params)
+
+        return response.json()

laceworksdk/api/v2/policies.py

@@ -71,7 +71,7 @@ def create(self,
         return super().create(
             policy_type=policy_type,
             query_id=query_id,
-            enabled=int(bool(enabled)),
+            enabled=enabled,
             title=title,
             description=description,
             remediation=remediation,

tests/api/v2/test_inventory.py

@@ -31,3 +31,13 @@ def test_api_search_by_date(self, api_object, csp):
     def test_api_search_by_date_deprecated(self, api_object, dataset):
 
         return super().test_api_search_by_date(api_object=api_object, filters={"dataset": dataset})
+
+    def test_inventory_scan(self, api_object, request):
+        response = api_object.scan(csp="AWS")
+
+        assert "data" in response.keys()
+
+    def test_inventory_status(self, api_object, request):
+        response = api_object.scan(csp="AWS")
+
+        assert "data" in response.keys()