fix!: get_user_falco_rules now returns the text and is updated with the new API endpoint (#151)

* fix: Custom rules are now correctly set and retrieved as string

* style: Change imports with concrete types

* fix: Update get_secure_user_falco_rules example to match the changes

* fix: Remove old outdated test

Author: tembleking

examples/get_secure_user_falco_rules.py

@@ -31,7 +31,7 @@
 # Return the result
 #
 if ok:
-    sys.stdout.write(res["userRulesFile"]["content"])
+    sys.stdout.write(res)
 else:
     print(res)
     sys.exit(1)

sdcclient/_secure.py

@@ -65,7 +65,16 @@ def get_user_falco_rules(self):
         **Example**
             `examples/get_secure_user_falco_rules.py <https://github.com/draios/python-sdc-client/blob/master/examples/get_secure_user_falco_rules.py>`_
         '''
-        return self._get_falco_rules("user")
+        ok, res = self._get_user_falco_rules()
+        return res if not ok else [True, res["customFalcoRulesFiles"]["files"][0]["variants"][0]["content"]]
+
+    def _get_user_falco_rules(self):
+        res = requests.get(self.url + '/api/settings/falco/customRulesFiles', headers=self.hdrs, verify=self.ssl_verify)
+
+        if not self._checkResponse(res):
+            return [False, self.lasterr]
+
+        return [True, (res.json())]
 
     def _set_falco_rules(self, kind, rules_content):
         payload = self._get_falco_rules(kind)
@@ -111,7 +120,21 @@ def set_user_falco_rules(self, rules_content):
             `examples/set_secure_user_falco_rules.py <https://github.com/draios/python-sdc-client/blob/master/examples/get_secure_user_falco_rules.py>`_
 
         '''
-        return self._set_falco_rules("user", rules_content)
+        ok, res = self._get_user_falco_rules()
+
+        if not ok:
+            return res
+
+        res["customFalcoRulesFiles"]["files"][0]["variants"][0]["content"] = rules_content
+
+        res = requests.put(self.url + '/api/settings/falco/customRulesFiles', headers=self.hdrs,
+                           data=json.dumps(res), verify=self.ssl_verify)
+
+        if not self._checkResponse(res):
+            return [False, self.lasterr]
+        res_json = res.json()
+        return [True, res_json["customFalcoRulesFiles"]["files"][0]["variants"][0]["content"]]
+
 
     # Only one kind for now called "default", but might add a "custom" kind later.
     def _get_falco_rules_files(self, kind):

specs/secure/custom_rules_spec.py

@@ -0,0 +1,77 @@
+import os
+
+from expects import equal, expect, start_with, contain
+from mamba import before, context, description, it
+
+from sdcclient import SdSecureClient
+from specs import be_successful_api_call
+
+with description("Custom Rules") as self:
+    with before.each:
+        self.client = SdSecureClient(sdc_url=os.getenv("SDC_SECURE_URL", "https://secure.sysdig.com"),
+                                     token=os.getenv("SDC_SECURE_TOKEN"))
+
+    with context("when the custom rules file exists"):
+        with it("can be retrieved"):
+            ok, res = self.client.get_user_falco_rules()
+
+            expect((ok, res)).to(be_successful_api_call)
+            expect(res).to(start_with("####################\n# Your custom rules!\n####################\n"))
+
+    with it("can push custom rules"):
+        _, previous_rules = self.client.get_user_falco_rules()
+        empty_rules = self.empty_falco_rules()
+        custom_rules = self.user_falco_rules()
+
+        ok, res = self.client.set_user_falco_rules(custom_rules)
+        expect((ok, res)).to(be_successful_api_call)
+        expect(res).to(equal(custom_rules))
+
+        ok, res = self.client.set_user_falco_rules(empty_rules)
+        expect((ok, res)).to(be_successful_api_call)
+        expect(res).to(equal(empty_rules))
+
+        ok, res = self.client.set_user_falco_rules(self.rules_without_header())
+        expect((ok, res)).to(be_successful_api_call)
+        # The endpoint automatically fills the header for the user.
+        expect(res).to(start_with("####################\n# Your custom rules!\n####################\n\n"))
+        expect(res).to(contain(self.rules_without_header()))
+
+        ok, res = self.client.set_user_falco_rules(previous_rules)
+        expect((ok, res)).to(be_successful_api_call)
+        expect(res).to(equal(previous_rules))
+
+
+    def user_falco_rules(self):
+        with open("fixtures/custom_rules.yaml", "r") as f:
+            return f.read()
+
+
+    def empty_falco_rules(self):
+        return """####################
+# Your custom rules!
+####################
+
+# Add new rules, like this one
+# - rule: A shell is run in a container
+#   desc: An event will trigger every time you run a shell in a container
+#   condition: evt.type = execve and evt.dir=< and container.id != host and proc.name = bash
+#   output: "Suspect shell run in container (user=%user.name %container.info shell=%proc.name parent=%proc.pname cmdline=%proc.cmdline)"
+#   priority: ERROR
+#   tags: [shell]
+
+# Or override any rule, macro, or list from the Default Rules
+"""
+
+
+    def rules_without_header(self):
+        return """\
+---
+- rule: "Testing rule"
+  desc: "Description"
+  condition: "always_true"
+  output: "Sample output"
+  priority: "WARNING"
+  tags: []
+  source: "syscall"
+"""

test/test_secure_apis.sh

@@ -35,7 +35,9 @@ EOF
 
 $SCRIPTDIR/../examples/set_secure_user_falco_rules.py $PYTHON_SDC_TEST_API_TOKEN /tmp/test_apis_user_rules.yaml
 $SCRIPTDIR/../examples/get_secure_user_falco_rules.py $PYTHON_SDC_TEST_API_TOKEN > /tmp/falco_rules.yaml
-diff /tmp/falco_rules.yaml /tmp/test_apis_user_rules.yaml
+# Removed comparison. The new endpoint automatically adds a header to the YAML file,
+# and this use case is already covered in the custom_rules_spec.py test file.
+# diff /tmp/falco_rules.yaml /tmp/test_apis_user_rules.yaml
 
 
 # Delete all policies and then get them. There should be none.