feat(sast): python Cdk policies batch 3 (#5820)

* 10 policies

* 20 policies

* fixes

* fix

* small change

* added tests

* small change

* small change

* bandit

* secrets

---------

Co-authored-by: achia <[email protected]>

.github/checkov.yaml

@@ -21,4 +21,10 @@ skip-path:
 - tests/terraform/runner/tf_plan_skip_check_regex/resource/tfplan1.json
 - tests/terraform/runner/tfplan2.json
 - tests/unit/test_secrets.py
+- /cdk_integration_tests/src/python/ElasticacheReplicationGroupEncryptionAtTransitAuthToken/fail__2__.py
+- /cdk_integration_tests/src/python/ElasticacheReplicationGroupEncryptionAtTransitAuthToken/pass.py
+- /checkov/cdk/checks/python/GlueDataCatalogEncryption.yaml
+- /checkov/cdk/checks/python/GlueDataCatalogEncryption.yaml
+- /checkov/cdk/checks/python/GlueDataCatalogEncryption.yaml
+- /checkov/cdk/checks/python/GlueDataCatalogEncryption.yaml
 summary-position: bottom

.github/exclude-patterns.txt

@@ -33,5 +33,4 @@ tests/terraform/runner/resources/get_graph_resource_entity_config/main.tf
 tests/terraform/runner/tf_plan_skip_check_regex/resource/.*
 tests/terraform/runner/tfplan2.json
 .*Scans.md
-.*Pipfile.lock
-cdk_integration_tests/src/python
+.*Pipfile.lock

cdk_integration_tests/src/python/APIGatewayAuthorization/fail__1__.py

@@ -0,0 +1,21 @@
+from aws_cdk import core
+from aws_cdk import aws_apigateway as apigw
+
+class MyApiGatewayMethodStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create the API Gateway Method based on the conditions
+        api_method = apigw.Method(
+            self, 'MyApiGatewayMethod',
+            http_method='GET',  # Replace with your desired HTTP method
+            resource=self.node.try_get_context('resource'),  # Replace with your API resource
+            rest_api=self.node.try_get_context('rest_api'),  # Replace with your REST API
+            authorization_type=apigw.AuthorizationType.NONE,  # Set the AuthorizationType to NONE
+            api_key_required=False  # Set ApiKeyRequired to false
+            # You can add other properties as needed for your method
+        )
+
+app = core.App()
+MyApiGatewayMethodStack(app, "MyApiGatewayMethodStack")
+app.synth()

cdk_integration_tests/src/python/APIGatewayAuthorization/pass.py

@@ -0,0 +1,21 @@
+from aws_cdk import core
+from aws_cdk import aws_apigateway as apigw
+
+class MyApiGatewayMethodStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create the API Gateway Method based on the conditions
+        api_method = apigw.Method(
+            self, 'MyApiGatewayMethod',
+            http_method='OPTIONS',  # Replace with your desired HTTP method
+            resource=self.node.try_get_context('resource'),  # Replace with your API resource
+            rest_api=self.node.try_get_context('rest_api'),  # Replace with your REST API
+            authorization_type=apigw.AuthorizationType.NONE,  # Set the AuthorizationType to NONE
+            api_key_required=True  # Set ApiKeyRequired to false
+            # You can add other properties as needed for your method
+        )
+
+app = core.App()
+MyApiGatewayMethodStack(app, "MyApiGatewayMethodStack")
+app.synth()

cdk_integration_tests/src/python/APIGatewayV2AccessLogging/fail__2__.py

@@ -0,0 +1,37 @@
+from aws_cdk import core
+from aws_cdk import aws_apigatewayv2 as apigatewayv2
+
+class MyApiGatewayV2StageStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define API Gateway V2 Stage with AccessLogSettings/DestinationArn set
+        api_stage = apigatewayv2.CfnStage(
+            self, 'MyApiGatewayV2Stage',
+            api_id='api_id_here',  # Replace with your API ID
+            stage_name='myStage',
+            # Add other properties as needed for your stage
+        )
+
+app = core.App()
+MyApiGatewayV2StageStack(app, "MyApiGatewayV2StageStack")
+app.synth()
+
+from aws_cdk import core
+from aws_cdk import aws_apigatewayv2 as apigatewayv2
+
+class MyServerlessHttpApiStack2(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define a Serverless HTTP API with access log settings
+        serverless_api = apigatewayv2.CfnApi(
+            self, 'MyServerlessHttpApi',
+            name='MyHTTPAPI',
+            protocol_type='HTTP',
+            # Add other properties as needed for your HTTP API
+        )
+
+app = core.App()
+MyServerlessHttpApiStack2(app, "MyServerlessHttpApiStack2")
+app.synth()

cdk_integration_tests/src/python/APIGatewayV2AccessLogging/pass.py

@@ -0,0 +1,45 @@
+from aws_cdk import core
+from aws_cdk import aws_apigatewayv2 as apigatewayv2
+
+class MyApiGatewayV2StageStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define API Gateway V2 Stage with AccessLogSettings/DestinationArn set
+        api_stage = apigatewayv2.CfnStage(
+            self, 'MyApiGatewayV2Stage',
+            api_id='api_id_here',  # Replace with your API ID
+            stage_name='myStage',
+            access_log_settings=apigatewayv2.CfnStage.AccessLogSettingsProperty(
+                destination_arn='arn:aws:logs:REGION:ACCOUNT_ID:log-group:LOG_GROUP_NAME'
+                # Replace with the actual DestinationArn value
+            )
+            # Add other properties as needed for your stage
+        )
+
+app = core.App()
+MyApiGatewayV2StageStack(app, "MyApiGatewayV2StageStack")
+app.synth()
+
+from aws_cdk import core
+from aws_cdk import aws_apigatewayv2 as apigatewayv2
+
+class MyServerlessHttpApiStack2(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define a Serverless HTTP API with access log settings
+        serverless_api = apigatewayv2.CfnApi(
+            self, 'MyServerlessHttpApi',
+            name='MyHTTPAPI',
+            protocol_type='HTTP',
+            access_log_settings=apigatewayv2.CfnApi.AccessLogSettingsProperty(
+                destination_arn='arn:aws:logs:REGION:ACCOUNT_ID:log-group:LOG_GROUP_NAME'
+                # Replace with the actual DestinationArn value
+            )
+            # Add other properties as needed for your HTTP API
+        )
+
+app = core.App()
+MyServerlessHttpApiStack2(app, "MyServerlessHttpApiStack2")
+app.synth()

cdk_integration_tests/src/python/APIGatewayXray/fail__2__.py

@@ -0,0 +1,44 @@
+from aws_cdk import core
+from aws_cdk import aws_apigateway as apigateway
+from aws_cdk import aws_apigatewayv2 as apigatewayv2
+
+class MyApiGatewayStageStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define API Gateway Stage with Tracing Enabled
+        apigateway.CfnStage(
+            self, 'MyApiGatewayStage',
+            stage_name='my-stage',
+            rest_api_id='your-rest-api-id',  # Replace with your RestApi Id
+            tracing_enabled=False
+            # Other properties for your API Gateway Stage
+        )
+
+app = core.App()
+MyApiGatewayStageStack(app, "MyApiGatewayStageStack")
+app.synth()
+
+class MyServerlessApiStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define Serverless API with Tracing Enabled
+        api = apigatewayv2.CfnApi(
+            self, 'MyServerlessApi',
+            name='my-serverless-api',
+            protocol_type='HTTP'
+            # Other properties for your Serverless API
+        )
+
+        stage = apigatewayv2.CfnStage(
+            self, 'MyServerlessApiStage',
+            api_id=api.ref,
+            stage_name='my-stage',
+            tracing_enabled=False
+            # Other properties for your API Gatewayv2 Stage
+        )
+
+app = core.App()
+MyServerlessApiStack(app, "MyServerlessApiStack")
+app.synth()

cdk_integration_tests/src/python/APIGatewayXray/pass.py

@@ -0,0 +1,44 @@
+from aws_cdk import core
+from aws_cdk import aws_apigateway as apigateway
+from aws_cdk import aws_apigatewayv2 as apigatewayv2
+
+class MyApiGatewayStageStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define API Gateway Stage with Tracing Enabled
+        apigateway.CfnStage(
+            self, 'MyApiGatewayStage',
+            stage_name='my-stage',
+            rest_api_id='your-rest-api-id',  # Replace with your RestApi Id
+            tracing_enabled=True
+            # Other properties for your API Gateway Stage
+        )
+
+app = core.App()
+MyApiGatewayStageStack(app, "MyApiGatewayStageStack")
+app.synth()
+
+class MyServerlessApiStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define Serverless API with Tracing Enabled
+        api = apigatewayv2.CfnApi(
+            self, 'MyServerlessApi',
+            name='my-serverless-api',
+            protocol_type='HTTP'
+            # Other properties for your Serverless API
+        )
+
+        stage = apigatewayv2.CfnStage(
+            self, 'MyServerlessApiStage',
+            api_id=api.ref,
+            stage_name='my-stage',
+            tracing_enabled=True
+            # Other properties for your API Gatewayv2 Stage
+        )
+
+app = core.App()
+MyServerlessApiStack(app, "MyServerlessApiStack")
+app.synth()

cdk_integration_tests/src/python/CloudFrontTLS12/fail__2__.py

@@ -0,0 +1,40 @@
+from aws_cdk import core
+from aws_cdk import aws_cloudfront as cloudfront
+
+class MyCloudFrontDistributionStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        distribution = cloudfront.CfnDistribution(
+            self, 'MyCloudFrontDistribution',
+            distribution_config=cloudfront.CfnDistribution.DistributionConfigProperty(
+                viewer_certificate=cloudfront.CfnDistribution.ViewerCertificateProperty(
+                    cloudfront_default_certificate=False,
+                    minimum_protocol_version='TLSv1.1'  # Define the minimum supported TLS version
+                ),
+                # Other distribution configuration properties
+            )
+        )
+
+app = core.App()
+MyCloudFrontDistributionStack(app, "MyCloudFrontDistributionStack")
+app.synth()
+
+class MyCloudFrontDistributionStack2(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        distribution = cloudfront.CfnDistribution(
+            self, 'MyCloudFrontDistribution',
+            distribution_config=cloudfront.CfnDistribution.DistributionConfigProperty(
+                viewer_certificate=cloudfront.CfnDistribution.ViewerCertificateProperty(
+                    cloudfront_default_certificate=False,
+                    minimum_protocol_version='TLSv1.0'  # Define the minimum supported TLS version
+                ),
+                # Other distribution configuration properties
+            )
+        )
+
+app = core.App()
+MyCloudFrontDistributionStack2(app, "MyCloudFrontDistributionStack2")
+app.synth()

cdk_integration_tests/src/python/CloudFrontTLS12/pass.py

@@ -0,0 +1,21 @@
+from aws_cdk import core
+from aws_cdk import aws_cloudfront as cloudfront
+
+class MyCloudFrontDistributionStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        distribution = cloudfront.CfnDistribution(
+            self, 'MyCloudFrontDistribution',
+            distribution_config=cloudfront.CfnDistribution.DistributionConfigProperty(
+                viewer_certificate=cloudfront.CfnDistribution.ViewerCertificateProperty(
+                    cloudfront_default_certificate=False,
+                    minimum_protocol_version='TLSv1.2'  # Define the minimum supported TLS version
+                ),
+                # Other distribution configuration properties
+            )
+        )
+
+app = core.App()
+MyCloudFrontDistributionStack(app, "MyCloudFrontDistributionStack")
+app.synth()

cdk_integration_tests/src/python/CloudWatchLogGroupRetention/fail__1__.py

@@ -0,0 +1,16 @@
+from aws_cdk import core
+from aws_cdk import aws_logs as logs
+
+class MyLogGroupStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define CloudWatch Logs Log Group with Retention Period
+        logs.CfnLogGroup(
+            self, 'MyLogGroup',
+            log_group_name='my-log-group',
+        )
+
+app = core.App()
+MyLogGroupStack(app, "MyLogGroupStack")
+app.synth()

cdk_integration_tests/src/python/CloudWatchLogGroupRetention/pass.py

@@ -0,0 +1,18 @@
+from aws_cdk import core
+from aws_cdk import aws_logs as logs
+
+class MyLogGroupStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define CloudWatch Logs Log Group with Retention Period
+        logs.CfnLogGroup(
+            self, 'MyLogGroup',
+            log_group_name='my-log-group',
+            retention_in_days=30  # Replace with your desired retention period in days
+            # Other properties for your Log Group
+        )
+
+app = core.App()
+MyLogGroupStack(app, "MyLogGroupStack")
+app.synth()

cdk_integration_tests/src/python/CloudfrontDistributionEncryption/fail__2__.py

@@ -0,0 +1,44 @@
+from aws_cdk import core
+from aws_cdk import aws_cloudfront as cloudfront
+
+class MyCloudFrontDistributionStack(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define CloudFront Distribution with ViewerProtocolPolicy set to allow_all
+        distribution = cloudfront.CfnDistribution(
+            self, 'MyCloudFrontDistribution',
+            distribution_config=cloudfront.CfnDistribution.DistributionConfigProperty(
+                default_cache_behavior=cloudfront.CfnDistribution.DefaultCacheBehaviorProperty(
+                    viewer_protocol_policy='allow-all'
+                ),
+                # Add other properties for the distribution config as needed
+            )
+        )
+
+app = core.App()
+MyCloudFrontDistributionStack(app, "MyCloudFrontDistributionStack")
+app.synth()
+
+class MyCloudFrontDistributionStack2(core.Stack):
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define CloudFront Distribution with CacheBehavior and ViewerProtocolPolicy
+        distribution = cloudfront.CfnDistribution(
+            self, 'MyCloudFrontDistribution',
+            distribution_config=cloudfront.CfnDistribution.DistributionConfigProperty(
+                cache_behaviors=[
+                    cloudfront.CfnDistribution.CacheBehaviorProperty(
+                        path_pattern='/path-to-cache',
+                        target_origin_id='my-target-origin-id',
+                        viewer_protocol_policy='allow-all'
+                    )
+                ],
+                # Other distribution configuration properties
+            )
+        )
+
+app = core.App()
+MyCloudFrontDistributionStack2(app, "MyCloudFrontDistributionStack2")
+app.synth()