feat: add 4 Security Hub TLS checks (CKV_AWS_394-397)

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

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+
+
+class APIGatewayV2IntegrationTLS(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure API Gateway V2 private integrations use HTTPS via tls_config"
+        id = "CKV_AWS_396"
+        supported_resources = ["aws_apigatewayv2_integration"]
+        categories = [CheckCategories.ENCRYPTION]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
+        # Only applies to VPC_LINK (private) integrations
+        connection_type = conf.get("connection_type")
+        if connection_type:
+            if isinstance(connection_type, list):
+                connection_type = connection_type[0]
+        # Non-VPC_LINK integrations auto-pass
+        if not connection_type or connection_type != "VPC_LINK":
+            return CheckResult.PASSED
+
+        self.evaluated_keys = ["tls_config/[0]/server_name_to_verify"]
+        tls_config = conf.get("tls_config")
+        if tls_config:
+            if isinstance(tls_config, list):
+                tls_config = tls_config[0]
+            if isinstance(tls_config, dict):
+                server_name = tls_config.get("server_name_to_verify")
+                if server_name:
+                    if isinstance(server_name, list):
+                        server_name = server_name[0]
+                    if BaseResourceCheck._is_variable_dependant(server_name):
+                        return CheckResult.UNKNOWN
+                    if isinstance(server_name, str) and server_name.strip():
+                        return CheckResult.PASSED
+
+        # FAIL if tls_config is absent or server_name_to_verify is empty
+        return CheckResult.FAILED
+
+
+check = APIGatewayV2IntegrationTLS()

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

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+
+
+class DMSRedisEndpointTLS(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure DMS Redis endpoint has TLS enabled"
+        id = "CKV_AWS_397"
+        supported_resources = ["aws_dms_endpoint"]
+        categories = [CheckCategories.ENCRYPTION]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
+        # Only applies to Redis endpoints
+        engine_name = conf.get("engine_name")
+        if engine_name:
+            if isinstance(engine_name, list):
+                engine_name = engine_name[0]
+        if not engine_name or engine_name != "redis":
+            return CheckResult.PASSED
+
+        self.evaluated_keys = ["redis_settings/[0]/ssl_security_protocol"]
+        redis_settings = conf.get("redis_settings")
+        if redis_settings:
+            if isinstance(redis_settings, list):
+                redis_settings = redis_settings[0]
+            if isinstance(redis_settings, dict):
+                ssl_protocol = redis_settings.get("ssl_security_protocol")
+                if ssl_protocol:
+                    if isinstance(ssl_protocol, list):
+                        ssl_protocol = ssl_protocol[0]
+                    if BaseResourceCheck._is_variable_dependant(ssl_protocol):
+                        return CheckResult.UNKNOWN
+                    if isinstance(ssl_protocol, str) and ssl_protocol.lower() == "plaintext":
+                        return CheckResult.FAILED
+                    # Explicit ssl-encryption passes
+                    if isinstance(ssl_protocol, str) and ssl_protocol.lower() == "ssl-encryption":
+                        return CheckResult.PASSED
+
+        # If redis_settings is absent or ssl_security_protocol is absent,
+        # AWS defaults to ssl-encryption, so PASS
+        return CheckResult.PASSED
+
+
+check = DMSRedisEndpointTLS()

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

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+
+
+class ELBv2TargetGroupHealthCheckProtocol(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure ELBv2 target group health check uses HTTPS"
+        id = "CKV_AWS_394"
+        supported_resources = ["aws_lb_target_group", "aws_alb_target_group"]
+        categories = [CheckCategories.ENCRYPTION]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
+        # Skip if target_type is lambda (no health checks apply)
+        target_type = conf.get("target_type")
+        if target_type:
+            if isinstance(target_type, list):
+                target_type = target_type[0]
+            if target_type == "lambda":
+                return CheckResult.PASSED
+
+        # Skip if transport protocol is TCP, UDP, TCP_UDP, or GENEVE
+        # (these protocols cannot use HTTPS health checks)
+        protocol = conf.get("protocol")
+        if protocol:
+            if isinstance(protocol, list):
+                protocol = protocol[0]
+            if isinstance(protocol, str) and protocol.upper() in ("TCP", "UDP", "TCP_UDP", "GENEVE"):
+                return CheckResult.PASSED
+
+        # Check health_check block
+        self.evaluated_keys = ["health_check/[0]/protocol"]
+        health_check = conf.get("health_check")
+        if health_check:
+            if isinstance(health_check, list):
+                health_check = health_check[0]
+            if isinstance(health_check, dict):
+                hc_protocol = health_check.get("protocol")
+                if hc_protocol:
+                    if isinstance(hc_protocol, list):
+                        hc_protocol = hc_protocol[0]
+                    if BaseResourceCheck._is_variable_dependant(hc_protocol):
+                        return CheckResult.UNKNOWN
+                    if isinstance(hc_protocol, str) and hc_protocol.upper() == "HTTPS":
+                        return CheckResult.PASSED
+
+        # FAIL if health_check block is absent or protocol is not HTTPS
+        return CheckResult.FAILED
+
+
+check = ELBv2TargetGroupHealthCheckProtocol()

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

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+
+
+class ELBv2TargetGroupProtocolEncrypted(BaseResourceCheck):
+    def __init__(self) -> None:
+        name = "Ensure ELBv2 target group uses an encrypted protocol (HTTPS or TLS)"
+        id = "CKV_AWS_395"
+        supported_resources = ["aws_lb_target_group", "aws_alb_target_group"]
+        categories = [CheckCategories.ENCRYPTION]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
+        # Skip if target_type is lambda (no transport protocol)
+        target_type = conf.get("target_type")
+        if target_type:
+            if isinstance(target_type, list):
+                target_type = target_type[0]
+            if target_type == "lambda":
+                return CheckResult.PASSED
+
+        self.evaluated_keys = ["protocol"]
+        protocol = conf.get("protocol")
+        if protocol:
+            if isinstance(protocol, list):
+                protocol = protocol[0]
+            if BaseResourceCheck._is_variable_dependant(protocol):
+                return CheckResult.UNKNOWN
+            if isinstance(protocol, str) and protocol.upper() in ("HTTPS", "TLS"):
+                return CheckResult.PASSED
+
+        # FAIL for HTTP, TCP, UDP, TCP_UDP, GENEVE, or missing protocol (defaults to HTTP)
+        return CheckResult.FAILED
+
+
+check = ELBv2TargetGroupProtocolEncrypted()

tests/terraform/checks/resource/aws/example_APIGatewayV2IntegrationTLS/main.tf

@@ -0,0 +1,52 @@
+# PASS: VPC_LINK with tls_config and server_name_to_verify set
+resource "aws_apigatewayv2_integration" "pass_vpc_link_tls" {
+  api_id             = "api-123"
+  integration_type   = "HTTP_PROXY"
+  integration_uri    = "https://example.com"
+  connection_type    = "VPC_LINK"
+  connection_id      = "vpc-link-123"
+  integration_method = "GET"
+
+  tls_config {
+    server_name_to_verify = "example.com"
+  }
+}
+
+# PASS: non-VPC_LINK connection_type (auto-pass, not a private integration)
+resource "aws_apigatewayv2_integration" "pass_internet" {
+  api_id           = "api-123"
+  integration_type = "HTTP_PROXY"
+  integration_uri  = "https://example.com"
+  connection_type  = "INTERNET"
+}
+
+# PASS: no connection_type specified (defaults to INTERNET, auto-pass)
+resource "aws_apigatewayv2_integration" "pass_no_connection_type" {
+  api_id           = "api-123"
+  integration_type = "HTTP_PROXY"
+  integration_uri  = "https://example.com"
+}
+
+# FAIL: VPC_LINK without tls_config block
+resource "aws_apigatewayv2_integration" "fail_no_tls_config" {
+  api_id             = "api-123"
+  integration_type   = "HTTP_PROXY"
+  integration_uri    = "https://example.com"
+  connection_type    = "VPC_LINK"
+  connection_id      = "vpc-link-123"
+  integration_method = "GET"
+}
+
+# FAIL: VPC_LINK with tls_config but empty server_name_to_verify
+resource "aws_apigatewayv2_integration" "fail_empty_server_name" {
+  api_id             = "api-123"
+  integration_type   = "HTTP_PROXY"
+  integration_uri    = "https://example.com"
+  connection_type    = "VPC_LINK"
+  connection_id      = "vpc-link-123"
+  integration_method = "GET"
+
+  tls_config {
+    server_name_to_verify = ""
+  }
+}

tests/terraform/checks/resource/aws/example_DMSRedisEndpointTLS/main.tf

@@ -0,0 +1,57 @@
+# PASS: Redis endpoint with explicit ssl-encryption
+resource "aws_dms_endpoint" "pass_ssl_encryption" {
+  endpoint_id   = "redis-pass-ssl"
+  endpoint_type = "target"
+  engine_name   = "redis"
+
+  redis_settings {
+    auth_type             = "auth-token"
+    auth_user_name        = "myuser"
+    auth_password         = "mypassword"
+    server_name           = "redis.example.com"
+    port                  = 6379
+    ssl_security_protocol = "ssl-encryption"
+  }
+}
+
+# PASS: Redis endpoint without ssl_security_protocol (AWS defaults to ssl-encryption)
+resource "aws_dms_endpoint" "pass_default_ssl" {
+  endpoint_id   = "redis-pass-default"
+  endpoint_type = "target"
+  engine_name   = "redis"
+
+  redis_settings {
+    auth_type      = "auth-token"
+    auth_user_name = "myuser"
+    auth_password  = "mypassword"
+    server_name    = "redis.example.com"
+    port           = 6379
+  }
+}
+
+# PASS: non-Redis endpoint (auto-pass)
+resource "aws_dms_endpoint" "pass_mysql" {
+  endpoint_id   = "mysql-pass"
+  endpoint_type = "source"
+  engine_name   = "mysql"
+  server_name   = "mysql.example.com"
+  port          = 3306
+  username      = "admin"
+  password      = "password"
+}
+
+# FAIL: Redis endpoint with plaintext
+resource "aws_dms_endpoint" "fail_plaintext" {
+  endpoint_id   = "redis-fail-plaintext"
+  endpoint_type = "target"
+  engine_name   = "redis"
+
+  redis_settings {
+    auth_type             = "auth-token"
+    auth_user_name        = "myuser"
+    auth_password         = "mypassword"
+    server_name           = "redis.example.com"
+    port                  = 6379
+    ssl_security_protocol = "plaintext"
+  }
+}