Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(arm): add CKV_AZURE_85 to ensure that Azure Defender is set to On for Kubernetes #6279

Merged
merged 8 commits into from
Jul 3, 2024
26 changes: 26 additions & 0 deletions checkov/arm/checks/resource/AzureDefenderOnKubernetes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
from __future__ import annotations
from typing import Any
from checkov.common.models.enums import CheckCategories, CheckResult
from checkov.arm.base_resource_check import BaseResourceCheck


class AzureDefenderOnKubernetes(BaseResourceCheck):
def __init__(self) -> None:
name = "Ensure that Azure Defender is set to On for Kubernetes"
id = "CKV_AZURE_85"
supported_resources = ("Microsoft.Security/pricings",)
categories = (CheckCategories.GENERAL_SECURITY,)
super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources,)

def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
return (
CheckResult.PASSED
if conf.get("name") != "KubernetesService" or str(conf["properties"]["pricingTier"]).lower() == "standard"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Im not sure that this is the right check - conf.get("name") != "KubernetesService", the name can be everything..

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did like the TF check

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can remove the name from the condition, it is not the same as TF.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ChanochShayner I believe it should only fail if the "name" : "KubernetesService" and the pricing tier is not Standard. Otherwise, it will fail for resources that are not K8s based and can't have Defender, despite the name of the policy. It's a very narrow scope, but I believe that's the purpose of the check.

else CheckResult.FAILED
)

def get_evaluated_keys(self) -> list[str]:
return ["name", "pricingTier"]


check = AzureDefenderOnKubernetes()
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
{
"$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
"contentVersion": "1.0.0.0",
"parameters": {
"pricing": {
"type": "string",
"allowedValues": [
"Standard",
"Free"
]
}
},

"resources": [
{
"type": "Microsoft.Security/pricings",
"apiVersion": "2017-08-01-preview",
"name": "KubernetesService",
"properties": {
"pricingTier": "Free"
}
},
{
"type": "Microsoft.Compute/disks",
"apiVersion": "2023-01-02",
"name": "[parameters('disks_acctestmd1_name')]",
"location": "westus2",
"tags": {
"environment": "staging"
},
"sku": {
"name": "Standard_LRS",
"tier": "Standard"
},
"properties": {
"creationData": {
"createOption": "Empty"
},
"diskSizeGB": 1,
"diskIOPSReadWrite": 500,
"diskMBpsReadWrite": 60,
"encryption": {
"type": "EncryptionAtRestWithPlatformKey"
},
"networkAccessPolicy": "AllowAll",
"publicNetworkAccess": "Enabled",
"diskState": "Unattached"
}
}

]}
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
{
"$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
"contentVersion": "1.0.0.0",
"parameters": {
"pricing": {
"type": "string",
"allowedValues": [
"Standard",
"Free"
]
}
},
"resources": [

{
"type": "Microsoft.Security/pricings",
"apiVersion": "2018-06-01",
"name": "KubernetesService",
"dependsOn": [
"[concat('Microsoft.Security/pricings/default')]"
],
"properties": {
"pricingTier": "Standard"
}
},
{
"type": "Microsoft.Security/pricings",
"apiVersion": "2018-06-01",
"name": "KeyVaults",
"dependsOn": [
"[concat('Microsoft.Security/pricings/SqlServers')]"
],
"properties": {
"pricingTier": "Free"
}
},
{
"type": "Microsoft.Security/pricings",
"apiVersion": "2018-06-01",
"name": "SqlServerVirtualMachines",
"dependsOn": [
"[concat('Microsoft.Security/pricings/AppServices')]"
],
"properties": {
"pricingTier": "Free"
}
}
]
}
41 changes: 41 additions & 0 deletions tests/arm/checks/resource/test_AzureDefenderOnKubernetes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
import unittest
import os
from checkov.arm.checks.resource.AzureDefenderOnKubernetes import check
from checkov.arm.runner import Runner
from checkov.runner_filter import RunnerFilter


class TestAzureDefenderOnKubernetes(unittest.TestCase):
def test_summary(self):
current_dir = os.path.dirname(os.path.realpath(__file__))
# given
test_files_dir = current_dir + "/example_AzureDefenderOnKubernetes"

# when
report = Runner().run(root_folder=str(test_files_dir), runner_filter=RunnerFilter(checks=[check.id]))

# then
summary = report.get_summary()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please assert the resources names as well

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please ☝️

passing_resources = {
"Microsoft.Security/pricings.KubernetesService",
"Microsoft.Security/pricings.KeyVaults",
"Microsoft.Security/pricings.SqlServerVirtualMachines",
}
failing_resources = {
"Microsoft.Security/pricings.KubernetesService",
}

passed_check_resources = {c.resource for c in report.passed_checks}
failed_check_resources = {c.resource for c in report.failed_checks}

self.assertEqual(summary['passed'], 3)
self.assertEqual(summary['failed'], 1)
self.assertEqual(summary['skipped'], 0)
self.assertEqual(summary['parsing_errors'], 0)

self.assertEqual(passing_resources, passed_check_resources)
self.assertEqual(failing_resources, failed_check_resources)


if __name__ == "__main__":
unittest.main()
Loading