Kubelet service process type rules to audits refactor (elastic#127)

* Initial commit of kubelet audits and refactoring to process

* Code review changes

* Refactor "with" keyword in tests

* Fixing duplicated "with" parameters

bundle/compliance/cis_eks/data_adapter.rego

@@ -1,7 +1,11 @@
 package compliance.cis_eks.data_adapter
 
-import data.compliance.lib.data_adapter
+is_aws_elb {
+	input.subType == "aws-elb"
+}
 
-process_args = result {
-	result = data_adapter.process_args(" ")
+is_aws_ecr {
+	input.subType == "aws-ecr"
 }
+
+process_args_seperator = " "

bundle/compliance/cis_eks/rules/cis_3_2_1/rule.rego

@@ -1,36 +1,18 @@
 package compliance.cis_eks.rules.cis_3_2_1
 
-import data.compliance.cis_eks
-import data.compliance.lib.assert
-import data.compliance.lib.common
-import data.compliance.lib.data_adapter
+import data.compliance.policy.process.ensure_arguments_and_config as audit
 
+# Ensure that the --anonymous-auth argument is set to false (Automated)
 default rule_evaluation = false
 
-process_args := cis_eks.data_adapter.process_args
-
 rule_evaluation {
-	common.contains_key_with_value(process_args, "--anonymous-auth", "false")
+	audit.process_contains_key_with_value("--anonymous-auth", "false")
 }
 
 # In case both flags and configuration file are specified, the executable argument takes precedence.
 # Checks that the entry for authentication:anonymous: enabled set to false.
 rule_evaluation {
-	not process_args["--anonymous-auth"]
-	assert.is_false(data_adapter.process_config.config.authentication.anonymous.enabled)
+	audit.not_process_arg_comparison("--anonymous-auth", ["authentication", "anonymous", "enabled"], false)
 }
 
-# Ensure that the --anonymous-auth argument is set to false (Automated)
-finding = result {
-	# filter
-	data_adapter.is_kubelet
-
-	# set result
-	result := {
-		"evaluation": common.calculate_result(rule_evaluation),
-		"evidence": {
-			"process_args": process_args,
-			"process_config": data_adapter.process_config,
-		},
-	}
-}
+finding = audit.finding(rule_evaluation)

bundle/compliance/cis_eks/rules/cis_3_2_1/test.rego

@@ -1,25 +1,34 @@
 package compliance.cis_eks.rules.cis_3_2_1
 
+import data.compliance.cis_eks.data_adapter
 import data.kubernetes_common.test_data
 import data.lib.test
 
-test_violation {
+violations {
 	test.assert_fail(finding) with input as rule_input("")
 	test.assert_fail(finding) with input as rule_input("--anonymous-auth true")
 	test.assert_fail(finding) with input as rule_input_with_external("--anonymous-auth true", create_process_config(true))
 	test.assert_fail(finding) with input as rule_input_with_external("--anonymous-auth true", create_process_config(false))
 	test.assert_fail(finding) with input as rule_input_with_external("", create_process_config(true))
 }
 
-test_pass {
+test_violations {
+	violations with data.benchmark_data_adapter as data_adapter
+}
+
+passes {
 	test.assert_pass(finding) with input as rule_input("--anonymous-auth false")
 	test.assert_pass(finding) with input as rule_input_with_external("--anonymous-auth false", create_process_config(true))
 	test.assert_pass(finding) with input as rule_input_with_external("--anonymous-auth false", create_process_config(false))
 	test.assert_pass(finding) with input as rule_input_with_external("", create_process_config(false))
 }
 
+test_pass {
+	passes with data.benchmark_data_adapter as data_adapter
+}
+
 test_not_evaluated {
-	not finding with input as test_data.process_input("some_process", [])
+	not finding with input as test_data.process_input("some_process", []) with data.benchmark_data_adapter as data_adapter
 }
 
 rule_input(argument) = test_data.process_input_with_external_data("kubelet", [argument], {})

bundle/compliance/cis_eks/rules/cis_3_2_10/rule.rego

@@ -1,35 +1,17 @@
 package compliance.cis_eks.rules.cis_3_2_10
 
-import data.compliance.cis_eks
-import data.compliance.lib.common
-import data.compliance.lib.data_adapter
+import data.compliance.policy.process.ensure_arguments_and_config as audit
 
 # Verify that the --rotate-certificates argument is not present, or is set to true.
-
 default rule_evaluation = true
 
-process_args := cis_eks.data_adapter.process_args
-
 rule_evaluation = false {
-	common.contains_key_with_value(process_args, "--rotate-certificates", "false")
+	audit.process_contains_key_with_value("--rotate-certificates", "false")
 }
 
 # In case both flags and configuration file are specified, the executable argument takes precedence.
 rule_evaluation = false {
-	not process_args["--rotate-certificates"]
-	data_adapter.process_config.config.rotateCertificates == false
+	audit.not_process_arg_comparison("--rotate-certificates", ["rotateCertificates"], false)
 }
 
-finding = result {
-	# filter
-	data_adapter.is_kubelet
-
-	# set result
-	result := {
-		"evaluation": common.calculate_result(rule_evaluation),
-		"evidence": {
-			"process_args": process_args,
-			"process_config": data_adapter.process_config,
-		},
-	}
-}
+finding = audit.finding(rule_evaluation)

bundle/compliance/cis_eks/rules/cis_3_2_10/test.rego

@@ -1,24 +1,33 @@
 package compliance.cis_eks.rules.cis_3_2_10
 
+import data.compliance.cis_eks.data_adapter
 import data.kubernetes_common.test_data
 import data.lib.test
 
-test_violation {
+violations {
 	test.assert_fail(finding) with input as rule_input("--rotate-certificates false")
 	test.assert_fail(finding) with input as rule_input_with_external("", create_process_config(false))
 	test.assert_fail(finding) with input as rule_input_with_external("--rotate-certificates false", create_process_config(true))
 }
 
-test_pass {
+test_violations {
+	violations with data.benchmark_data_adapter as data_adapter
+}
+
+passes {
 	test.assert_pass(finding) with input as rule_input("")
 	test.assert_pass(finding) with input as rule_input("--rotate-certificates true")
 	test.assert_pass(finding) with input as rule_input_with_external("--rotate-certificates true", create_process_config(false))
 	test.assert_pass(finding) with input as rule_input_with_external("--rotate-certificates true", create_process_config(true))
 	test.assert_pass(finding) with input as rule_input_with_external("", create_process_config(true))
 }
 
+test_pass {
+	passes with data.benchmark_data_adapter as data_adapter
+}
+
 test_not_evaluated {
-	not finding with input as test_data.process_input("some_process", [])
+	not finding with input as test_data.process_input("some_process", []) with data.benchmark_data_adapter as data_adapter
 }
 
 rule_input(argument) = test_data.process_input("kubelet", [argument])

bundle/compliance/cis_eks/rules/cis_3_2_11/rule.rego

@@ -1,48 +1,29 @@
 package compliance.cis_eks.rules.cis_3_2_11
 
-import data.compliance.cis_eks
-import data.compliance.lib.common
-import data.compliance.lib.data_adapter
+import data.compliance.policy.process.ensure_arguments_and_config as audit
 
 # Verify that the RotateKubeletServerCertificate argument is set to true
-
 default rule_evaluation = false
 
-process_args := cis_eks.data_adapter.process_args
-
 rule_evaluation {
-	common.contains_key_with_value(process_args, "--feature-gates", "RotateKubeletServerCertificate=true")
+	audit.process_contains_key_with_value("--feature-gates", "RotateKubeletServerCertificate=true")
 }
 
 # In case both flags and configuration file are specified, the executable argument takes precedence.
 rule_evaluation {
-	not process_args["--feature-gates"]
-	data_adapter.process_config.config.featureGates.RotateKubeletServerCertificate
+	audit.not_process_arg_variable("--feature-gates", ["featureGates", "RotateKubeletServerCertificate"])
 }
 
 rule_evaluation {
-	not contains(process_args["--feature-gates"], "RotateKubeletServerCertificate")
-	data_adapter.process_config.config.featureGates.RotateKubeletServerCertificate
+	audit.not_process_contains_variable("--feature-gates", "RotateKubeletServerCertificate", ["featureGates", "RotateKubeletServerCertificate"])
 }
 
 rule_evaluation {
-	common.contains_key_with_value(process_args, "--rotate-server-certificates", "true")
+	audit.process_contains_key_with_value("--rotate-server-certificates", "true")
 }
 
 rule_evaluation {
-	data_adapter.process_config.config.serverTLSBootstrap
+	audit.get_from_config(["serverTLSBootstrap"])
 }
 
-finding = result {
-	# filter
-	data_adapter.is_kubelet
-
-	# set result
-	result := {
-		"evaluation": common.calculate_result(rule_evaluation),
-		"evidence": {
-			"process_args": process_args,
-			"process_config": data_adapter.process_config,
-		},
-	}
-}
+finding = audit.finding(rule_evaluation)

bundle/compliance/cis_eks/rules/cis_3_2_11/test.rego

@@ -1,9 +1,10 @@
 package compliance.cis_eks.rules.cis_3_2_11
 
+import data.compliance.cis_eks.data_adapter
 import data.kubernetes_common.test_data
 import data.lib.test
 
-test_violation {
+violations {
 	test.assert_fail(finding) with input as rule_input("")
 	test.assert_fail(finding) with input as rule_input("--feature-gates RotateKubeletServerCertificate=false")
 	test.assert_fail(finding) with input as rule_input_with_external("", create_process_config(false, false))
@@ -12,7 +13,11 @@ test_violation {
 	test.assert_fail(finding) with input as rule_input_with_external("--feature-gates RotateKubeletServerCertificate false", create_process_config(true, false))
 }
 
-test_pass {
+test_violations {
+	violations with data.benchmark_data_adapter as data_adapter
+}
+
+passes {
 	test.assert_pass(finding) with input as rule_input("--feature-gates RotateKubeletServerCertificate=true")
 	test.assert_pass(finding) with input as rule_input("--rotate-server-certificates true")
 	test.assert_pass(finding) with input as rule_input_with_external("--feature-gates=RotateKubeletServerCertificate=true", create_process_config(true, false))
@@ -24,8 +29,12 @@ test_pass {
 	test.assert_pass(finding) with input as rule_input_with_external("", create_process_config(true, true))
 }
 
+test_pass {
+	passes with data.benchmark_data_adapter as data_adapter
+}
+
 test_not_evaluated {
-	not finding with input as test_data.process_input("some_process", [])
+	not finding with input as test_data.process_input("some_process", []) with data.benchmark_data_adapter as data_adapter
 }
 
 rule_input(argument) = test_data.process_input("kubelet", [argument])

bundle/compliance/cis_eks/rules/cis_3_2_2/rule.rego

@@ -1,43 +1,24 @@
 package compliance.cis_eks.rules.cis_3_2_2
 
-import data.compliance.cis_eks
-import data.compliance.lib.common
-import data.compliance.lib.data_adapter
+import data.compliance.policy.process.ensure_arguments_and_config as audit
 
 # Ensure that the --authorization-mode argument is not set to AlwaysAllow (Automated)
 # If the --authorization-mode argument is present check that it is not set to AlwaysAllow.
-
 default rule_evaluation = false
 
-process_args := cis_eks.data_adapter.process_args
+is_authorization_allow_all {
+	audit.process_arg_not_key_value("--authorization-mode", "--authorization-mode", "AlwaysAllow")
+}
 
 rule_evaluation {
 	is_authorization_allow_all
 }
 
-is_authorization_allow_all {
-	process_args["--authorization-mode"]
-	not common.contains_key_with_value(process_args, "--authorization-mode", "AlwaysAllow")
-}
-
 # In case both flags and configuration file are specified, the executable argument takes precedence.
 # Checks that the entry for authorization:mode is not set to AlwaysAllow.
 rule_evaluation {
 	not is_authorization_allow_all
-	data_adapter.process_config.config.authorization.mode
-	not data_adapter.process_config.config.authorization.mode == "AlwaysAllow"
+	audit.process_filter_variable_multi_comparison(["authorization", "mode"], ["authorization", "mode"], "AlwaysAllow")
 }
 
-finding = result {
-	# filter
-	data_adapter.is_kubelet
-
-	# set result
-	result := {
-		"evaluation": common.calculate_result(rule_evaluation),
-		"evidence": {
-			"process_args": process_args,
-			"process_config": data_adapter.process_config,
-		},
-	}
-}
+finding = audit.finding(rule_evaluation)

bundle/compliance/cis_eks/rules/cis_3_2_2/test.rego

@@ -1,23 +1,32 @@
 package compliance.cis_eks.rules.cis_3_2_2
 
+import data.compliance.cis_eks.data_adapter
 import data.kubernetes_common.test_data
 import data.lib.test
 
-test_violation {
+violations {
 	test.assert_fail(finding) with input as rule_input("")
 	test.assert_fail(finding) with input as rule_input("--authorization-mode AlwaysAllow")
 	test.assert_fail(finding) with input as rule_input_with_external("--authorization-mode AlwaysAllow", create_process_config("AlwaysAllow"))
 	test.assert_fail(finding) with input as rule_input_with_external("", create_process_config("AlwaysAllow"))
 }
 
-test_pass {
+test_violations {
+	violations with data.benchmark_data_adapter as data_adapter
+}
+
+passes {
 	test.assert_pass(finding) with input as rule_input("--authorization-mode Webhook")
 	test.assert_pass(finding) with input as rule_input_with_external("--authorization-mode Webhook", create_process_config("AlwaysAllow"))
 	test.assert_pass(finding) with input as rule_input_with_external("", create_process_config("Webhook"))
 }
 
+test_pass {
+	passes with data.benchmark_data_adapter as data_adapter
+}
+
 test_not_evaluated {
-	not finding with input as test_data.process_input("some_process", [])
+	not finding with input as test_data.process_input("some_process", []) with data.benchmark_data_adapter as data_adapter
 }
 
 rule_input(argument) = test_data.process_input("kubelet", [argument])