Merge branch 'main' into feat/AzureOpenAI

CHANGELOG.md

@@ -1,6 +1,12 @@
 # CHANGELOG
 
-## [Unreleased](https://github.com/bridgecrewio/checkov/compare/2.5.10...HEAD)
+## [Unreleased](https://github.com/bridgecrewio/checkov/compare/2.5.11...HEAD)
+
+## [2.5.11](https://github.com/bridgecrewio/checkov/compare/2.5.10...2.5.11) - 2023-10-17
+
+### Feature
+
+- **sca:** giving file path on relative the the current dir for cases there is no either specified root_folder and the is no repo scan dir - [#5654](https://github.com/bridgecrewio/checkov/pull/5654)
 
 ## [2.5.10](https://github.com/bridgecrewio/checkov/compare/2.5.9...2.5.10) - 2023-10-16
 

checkov/arm/checks/resource/DataFactoryUsesGitRepository.py

@@ -0,0 +1,30 @@
+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 DataFactoryUsesGitRepository(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure that Azure Data Factory uses Git repository for source control"
+        id = "CKV_AZURE_103"
+        supported_resources = ("Microsoft.DataFactory/factories",)
+        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:
+        properties = conf.get("properties")
+        if properties and isinstance(properties, dict):
+            self.evaluated_keys = ["properties/repoConfiguration/type"]
+            repo = properties.get("repoConfiguration")
+            if not repo:
+                return CheckResult.FAILED
+            if repo and isinstance(repo, dict) and repo.get("type") is not None:
+                return CheckResult.PASSED
+            return CheckResult.UNKNOWN
+        return CheckResult.FAILED
+
+
+check = DataFactoryUsesGitRepository()

checkov/arm/checks/resource/MySQLEncryptionEnabled.py

@@ -0,0 +1,32 @@
+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 MySQLEncryptionEnabled(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure that MySQL server enables infrastructure encryption"
+        id = "CKV_AZURE_96"
+        supported_resources = ("Microsoft.DBforMySQL/flexibleServers",)
+        categories = (CheckCategories.ENCRYPTION,)
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, Any], entity_type: str) -> CheckResult:
+        properties = conf.get("properties")
+        if properties and isinstance(properties, dict):
+            self.evaluated_keys = ["properties/dataencryption"]
+            data_encryption = properties.get("dataencryption")
+            if data_encryption and isinstance(data_encryption, dict):
+                if data_encryption is None:
+                    return CheckResult.FAILED
+                return CheckResult.PASSED
+            # unparsed
+            elif data_encryption and isinstance(data_encryption, str):
+                return CheckResult.UNKNOWN
+            return CheckResult.FAILED
+        return CheckResult.UNKNOWN
+
+
+check = MySQLEncryptionEnabled()

checkov/arm/checks/resource/VMEncryptionAtHostEnabled.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.arm.base_resource_check import BaseResourceCheck
+
+from typing import Any
+
+from checkov.common.util.data_structures_utils import find_in_dict
+
+
+class VMEncryptionAtHostEnabled(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure that Virtual machine scale sets have encryption at host enabled"
+        id = "CKV_AZURE_97"
+        supported_resources = ("Microsoft.Compute/virtualMachineScaleSets", "Microsoft.Compute/virtualMachines")
+        categories = (CheckCategories.ENCRYPTION,)
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
+        encryption = ""
+
+        if self.entity_type == "Microsoft.Compute/virtualMachines":
+            self.evaluated_keys = ["properties/securityProfile/encryptionAtHost"]
+            encryption = find_in_dict(input_dict=conf, key_path="properties/securityProfile/encryptionAtHost")
+        elif self.entity_type == "Microsoft.Compute/virtualMachineScaleSets":
+            self.evaluated_keys = ["properties/virtualMachineProfile/securityProfile/encryptionAtHost"]
+            encryption = find_in_dict(
+                input_dict=conf, key_path="properties/virtualMachineProfile/securityProfile/encryptionAtHost"
+            )
+
+        if encryption == "true":
+            return CheckResult.PASSED
+
+        return CheckResult.FAILED
+
+
+check = VMEncryptionAtHostEnabled()

checkov/cloudformation/checks/resource/aws/LambdaEnvironmentEncryptionSettings.py

@@ -1,30 +1,32 @@
-from typing import List
+from __future__ import annotations
+
+from typing import Any
 
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.cloudformation.checks.resource.base_resource_check import BaseResourceCheck
 
 
 class LambdaEnvironmentEncryptionSettings(BaseResourceCheck):
-    def __init__(self):
+    def __init__(self) -> None:
         name = "Check encryption settings for Lambda environmental variable"
         id = "CKV_AWS_173"
-        supported_resources = ['AWS::Lambda::Function', "AWS::Serverless::Function"]
-        categories = [CheckCategories.ENCRYPTION]
+        supported_resources = ("AWS::Lambda::Function", "AWS::Serverless::Function")
+        categories = (CheckCategories.ENCRYPTION,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def scan_resource_conf(self, conf):
-        properties = conf.get('Properties')
+    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
+        properties = conf.get("Properties")
         if properties is not None:
-            env = properties.get('Environment')
+            env = properties.get("Environment")
             if env is not None:
                 if not isinstance(env, dict):
                     return CheckResult.UNKNOWN
-                elif env.get('Variables') and not properties.get('KmsKeyArn'):
+                elif env.get("Variables") and not properties.get("KmsKeyArn"):
                     return CheckResult.FAILED
         return CheckResult.PASSED
 
-    def get_evaluated_keys(self) -> List[str]:
-        return ['Properties/Environment/Variables', 'Properties/KmsKeyArn']
+    def get_evaluated_keys(self) -> list[str]:
+        return ["Properties/KmsKeyArn"]
 
 
 check = LambdaEnvironmentEncryptionSettings()

checkov/terraform/checks/resource/aws/EKSPlatformVersion.py

@@ -24,7 +24,7 @@ def get_inspected_key(self) -> str:
         return "version"
 
     def get_expected_values(self) -> list[Any]:
-        return ["1.23", "1.24", "1.25", "1.26", "1.27"]
+        return ["1.23", "1.24", "1.25", "1.26", "1.27", "1.28"]
 
 
 check = EKSPlatformVersion()

checkov/terraform/checks/resource/aws/EKSPublicAccessCIDR.py

@@ -1,26 +1,28 @@
+from __future__ import annotations
+
+from typing import Any
+
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
 
 
 class EKSPublicAccessCIDR(BaseResourceCheck):
-    def __init__(self):
+    def __init__(self) -> None:
         name = "Ensure Amazon EKS public endpoint not accessible to 0.0.0.0/0"
         id = "CKV_AWS_38"
-        supported_resources = ['aws_eks_cluster']
-        categories = [CheckCategories.KUBERNETES]
+        supported_resources = ('aws_eks_cluster',)
+        categories = (CheckCategories.KUBERNETES,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def scan_resource_conf(self, conf):
+    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
         """
             Looks for public_access_cidrs at aws_eks_cluster:
             https://www.terraform.io/docs/providers/aws/r/eks_cluster.html
         :param conf: aws_eks_cluster configuration
         :return: <CheckResult>
         """
-        self.evaluated_keys = ['vpc_config']
         if "vpc_config" in conf.keys():
             if "endpoint_public_access" in conf["vpc_config"][0] and not conf["vpc_config"][0]["endpoint_public_access"][0]:
-                self.evaluated_keys = ['vpc_config/[0]/endpoint_public_access']
                 return CheckResult.PASSED
             elif "public_access_cidrs" in conf["vpc_config"][0]:
                 self.evaluated_keys = ['vpc_config/[0]/public_access_cidrs']

checkov/terraform/checks/resource/aws/LambdaEnvironmentEncryptionSettings.py

@@ -18,23 +18,19 @@ def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
         # check that if I have env vars I have a KMS key
         if len(conf.get("environment", [])):
             if "kms_key_arn" in conf:
-                if conf["kms_key_arn"] == [""]:
-                    self.evaluated_keys = ["environment/kms_key_arn"]
+                if conf.get("kms_key_arn") == [""]:
                     return CheckResult.FAILED
                 return CheckResult.PASSED
-            self.evaluated_keys = ["environment"]
             return CheckResult.FAILED
 
         # no env vars so should be no key as that causes state mismatch
-        if "kms_key_arn" in conf:
-            if not len(conf["kms_key_arn"]):
-                return CheckResult.PASSED
+        if "kms_key_arn" in conf and len(conf["kms_key_arn"]):
             return CheckResult.FAILED
         # neither env vars nor kms key
         return CheckResult.UNKNOWN
 
     def get_evaluated_keys(self) -> list[str]:
-        return ["environment/[0]/variables"]
+        return ["kms_key_arn"]
 
 
 check = LambdaEnvironmentEncryptionSettings()

checkov/terraform/checks/resource/aws/SNSTopicEncryption.py

@@ -1,4 +1,4 @@
-from typing import List, Any
+from typing import Any
 
 from checkov.common.models.enums import CheckCategories
 from checkov.common.models.consts import ANY_VALUE
@@ -16,8 +16,8 @@ def __init__(self) -> None:
     def get_inspected_key(self) -> str:
         return "kms_master_key_id"
 
-    def get_expected_values(self) -> List[Any]:
-        return [ANY_VALUE]
+    def get_expected_value(self) -> Any:
+        return ANY_VALUE
 
 
 check = SNSTopicEncryption()

checkov/terraform/checks/resource/azure/MariaDBPublicAccessDisabled.py

@@ -1,24 +1,22 @@
-from checkov.common.models.enums import CheckResult, CheckCategories
-from checkov.terraform.checks.resource.base_resource_value_check import BaseResourceCheck
-from typing import List
+from typing import Any
 
+from checkov.common.models.enums import CheckCategories
+from checkov.terraform.checks.resource.base_resource_value_check import BaseResourceValueCheck
 
-class MariaDBPublicAccessDisabled(BaseResourceCheck):
-    def __init__(self):
+
+class MariaDBPublicAccessDisabled(BaseResourceValueCheck):
+    def __init__(self) -> None:
         name = "Ensure 'public network access enabled' is set to 'False' for MariaDB servers"
         id = "CKV_AZURE_48"
-        supported_resources = ['azurerm_mariadb_server']
-        categories = [CheckCategories.NETWORKING]
+        supported_resources = ("azurerm_mariadb_server",)
+        categories = (CheckCategories.NETWORKING,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def scan_resource_conf(self, conf):
-        # Whether or not public network access is allowed for this server. Defaults to true. Which is not optimal
-        if 'public_network_access_enabled' not in conf or conf['public_network_access_enabled'][0]:
-            return CheckResult.FAILED
-        return CheckResult.PASSED
+    def get_inspected_key(self) -> str:
+        return "public_network_access_enabled"
 
-    def get_evaluated_keys(self) -> List[str]:
-        return ['public_network_access_enabled']
+    def get_expected_value(self) -> Any:
+        return False
 
 
 check = MariaDBPublicAccessDisabled()

checkov/terraform/checks/resource/azure/MySQLEncryptionEnaled.py → checkov/terraform/checks/resource/azure/MySQLEncryptionEnabled.py

@@ -2,7 +2,7 @@
 from checkov.terraform.checks.resource.base_resource_value_check import BaseResourceValueCheck
 
 
-class MySQLEncryptionEnaled(BaseResourceValueCheck):
+class MySQLEncryptionEnabled(BaseResourceValueCheck):
     def __init__(self):
         name = "Ensure that MySQL server enables infrastructure encryption"
         id = "CKV_AZURE_96"
@@ -14,4 +14,4 @@ def get_inspected_key(self):
         return 'infrastructure_encryption_enabled'
 
 
-check = MySQLEncryptionEnaled()
+check = MySQLEncryptionEnabled()

checkov/terraform/checks/resource/azure/VMEncryptionAtHostEnabled.py

@@ -3,14 +3,14 @@
 
 
 class VMEncryptionAtHostEnabled(BaseResourceValueCheck):
-    def __init__(self):
+    def __init__(self) -> None:
         name = "Ensure that Virtual machine scale sets have encryption at host enabled"
         id = "CKV_AZURE_97"
         supported_resources = ['azurerm_linux_virtual_machine_scale_set', 'azurerm_windows_virtual_machine_scale_set']
         categories = [CheckCategories.ENCRYPTION]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def get_inspected_key(self):
+    def get_inspected_key(self) -> str:
         return 'encryption_at_host_enabled'
 
 

checkov/terraform/checks/resource/gcp/GKEClusterLogging.py

@@ -1,30 +1,22 @@
-from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
-from checkov.common.models.enums import CheckResult, CheckCategories
-from typing import List
+from typing import Any
 
+from checkov.common.models.enums import CheckCategories
+from checkov.terraform.checks.resource.base_resource_negative_value_check import BaseResourceNegativeValueCheck
 
-class GKEClusterLogging(BaseResourceCheck):
-    def __init__(self):
+
+class GKEClusterLogging(BaseResourceNegativeValueCheck):
+    def __init__(self) -> None:
         name = "Ensure Stackdriver Logging is set to Enabled on Kubernetes Engine Clusters"
         id = "CKV_GCP_1"
-        supported_resources = ['google_container_cluster']
-        categories = [CheckCategories.KUBERNETES]
+        supported_resources = ("google_container_cluster",)
+        categories = (CheckCategories.KUBERNETES,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def scan_resource_conf(self, conf):
-        """
-            Looks for password configuration at azure_instance:
-            https://www.terraform.io/docs/providers/google/r/compute_ssl_policy.html
-        :param conf: google_compute_ssl_policy configuration
-        :return: <CheckResult>
-        """
-        if 'logging_service' in conf.keys():
-            if conf['logging_service'][0] == "none":
-                return CheckResult.FAILED
-        return CheckResult.PASSED
+    def get_inspected_key(self) -> str:
+        return "logging_service"
 
-    def get_evaluated_keys(self) -> List[str]:
-        return ['logging_service']
+    def get_forbidden_values(self) -> Any:
+        return "none"
 
 
 check = GKEClusterLogging()