feat(sast): Python cdk policies batch 2

cdk_integration_tests/src/python/APIGatewayAccessLogging/fail__2__.py

@@ -0,0 +1,49 @@
+from aws_cdk import core
+from aws_cdk import aws_elasticache as elasticache
+
+class ElastiCacheReplicationGroupStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an AWS ElastiCache Replication Group
+        replication_group = elasticache.CfnReplicationGroup(
+            self,
+            "MyElastiCacheReplicationGroup",
+            replication_group_id="my-replication-group",
+            replication_group_description="My ElastiCache Replication Group",
+            cache_node_type="cache.m4.large",
+            engine="redis",
+            engine_version="5.0.6",
+            port=6379,
+            num_cache_clusters=2,
+            automatic_failover_enabled=True,
+        )
+
+app = core.App()
+ElastiCacheReplicationGroupStack(app, "ElastiCacheReplicationGroupStack")
+app.synth()
+
+
+from aws_cdk import core
+from aws_cdk import aws_serverless as serverless
+
+class ServerlessApiWithAccessLogStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create a Serverless API
+        serverless.Api(
+            self, "MyApi",
+            default_stage={
+                "stage_name": "prod",
+                "access_log_setting": serverless.AccessLogSetting(
+                    format=serverless.AccessLogFormat.json_with_standard_fields()
+                )
+            }
+        )
+
+app = core.App()
+ServerlessApiWithAccessLogStack(app, "ServerlessApiWithAccessLogStack")
+app.synth()

cdk_integration_tests/src/python/APIGatewayAccessLogging/pass.py

@@ -0,0 +1,70 @@
+from aws_cdk import aws_apigateway as apigateway
+
+cfn_stage = apigateway.CfnStage(self, "MyCfnStage",
+    rest_api_id="restApiId",
+
+    # the properties below are optional
+    access_log_setting=apigateway.CfnStage.AccessLogSettingProperty(
+        destination_arn="destinationArn",
+        format="format"
+    ),
+    cache_cluster_enabled=False,
+    cache_cluster_size="cacheClusterSize",
+    canary_setting=apigateway.CfnStage.CanarySettingProperty(
+        deployment_id="deploymentId",
+        percent_traffic=123,
+        stage_variable_overrides={
+            "stage_variable_overrides_key": "stageVariableOverrides"
+        },
+        use_stage_cache=False
+    ),
+    client_certificate_id="clientCertificateId",
+    deployment_id="deploymentId",
+    description="description",
+    documentation_version="documentationVersion",
+    method_settings=[apigateway.CfnStage.MethodSettingProperty(
+        cache_data_encrypted=False,
+        cache_ttl_in_seconds=123,
+        caching_enabled=False,
+        data_trace_enabled=False,
+        http_method="httpMethod",
+        logging_level="loggingLevel",
+        metrics_enabled=False,
+        resource_path="resourcePath",
+        throttling_burst_limit=123,
+        throttling_rate_limit=123
+    )],
+    stage_name="stageName",
+    tags=[CfnTag(
+        key="key",
+        value="value"
+    )],
+    tracing_enabled=False,
+    variables={
+        "variables_key": "variables"
+    }
+)
+
+from aws_cdk import core
+from aws_cdk import aws_serverless as serverless
+
+class ServerlessApiWithAccessLogStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create a Serverless API
+        serverless.Api(
+            self, "MyApi",
+            default_stage={
+                "stage_name": "prod",
+                "access_log_setting": serverless.AccessLogSetting(
+                    destination_arn="arn:aws:logs:us-east-1:123456789012:log-group/MyLogGroup",
+                    format=serverless.AccessLogFormat.json_with_standard_fields()
+                )
+            }
+        )
+
+app = core.App()
+ServerlessApiWithAccessLogStack(app, "ServerlessApiWithAccessLogStack")
+app.synth()

cdk_integration_tests/src/python/AmazonMQBrokePublicAccess/fail__1__.py

@@ -0,0 +1,21 @@
+from aws_cdk import core
+from aws_cdk import aws_amazonmq as amazonmq
+
+class AmazonMQStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an Amazon MQ broker with PubliclyAccessible set to false
+        amazonmq_broker = amazonmq.CfnBroker(
+            self,
+            "MyAmazonMQBroker",
+            broker_name="my-amazon-mq-broker",
+            engine_type="ACTIVEMQ",
+            host_instance_type="mq.t2.micro",
+            publicly_accessible=True,  # Set PubliclyAccessible to false
+        )
+
+app = core.App()
+AmazonMQStack(app, "AmazonMQStack")
+app.synth()

cdk_integration_tests/src/python/AmazonMQBrokePublicAccess/pass.py

@@ -0,0 +1,21 @@
+from aws_cdk import core
+from aws_cdk import aws_amazonmq as amazonmq
+
+class AmazonMQStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an Amazon MQ broker with PubliclyAccessible set to false
+        amazonmq_broker = amazonmq.CfnBroker(
+            self,
+            "MyAmazonMQBroker",
+            broker_name="my-amazon-mq-broker",
+            engine_type="ACTIVEMQ",
+            host_instance_type="mq.t2.micro",
+            publicly_accessible=False,  # Set PubliclyAccessible to false
+        )
+
+app = core.App()
+AmazonMQStack(app, "AmazonMQStack")
+app.synth()

cdk_integration_tests/src/python/AppSyncFieldLevelLogs/fail__1__.py

@@ -0,0 +1,24 @@
+from aws_cdk import core
+from aws_cdk import aws_appsync as appsync
+
+class AppSyncStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define the GraphQL API using CfnGraphQLApi
+        graphql_api = appsync.CfnGraphQLApi(
+            self,
+            "AppSyncGraphQLApi",
+            name="MyAppSyncAPI",
+            authentication_type="API_KEY",  # You can change the authentication type
+            log_config=appsync.CfnGraphQLApi.LogConfigProperty(
+                exclude_verbose_content=False,
+                field_log_level="fieldLogLevel"
+            ),
+        )
+
+
+app = core.App()
+AppSyncStack(app, "AppSyncStack")
+app.synth()

cdk_integration_tests/src/python/AppSyncFieldLevelLogs/pass.py

@@ -0,0 +1,25 @@
+from aws_cdk import core
+from aws_cdk import aws_appsync as appsync
+
+class AppSyncStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define the GraphQL API using CfnGraphQLApi
+        graphql_api = appsync.CfnGraphQLApi(
+            self,
+            "AppSyncGraphQLApi",
+            name="MyAppSyncAPI",
+            authentication_type="API_KEY",  # You can change the authentication type
+            log_config=appsync.CfnGraphQLApi.LogConfigProperty(
+                cloud_watch_logs_role_arn="cloudWatchLogsRoleArn",
+                exclude_verbose_content=False,
+                field_log_level="fieldLogLevel"
+            ),
+        )
+
+
+app = core.App()
+AppSyncStack(app, "AppSyncStack")
+app.synth()

cdk_integration_tests/src/python/AppSyncLogging/fail__1__.py

@@ -0,0 +1,24 @@
+from aws_cdk import core
+from aws_cdk import aws_appsync as appsync
+
+class AppSyncStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define the GraphQL API using CfnGraphQLApi
+        graphql_api = appsync.CfnGraphQLApi(
+            self,
+            "AppSyncGraphQLApi",
+            name="MyAppSyncAPI",
+            authentication_type="API_KEY",  # You can change the authentication type
+            log_config=appsync.CfnGraphQLApi.LogConfigProperty(
+                exclude_verbose_content=False,
+                field_log_level="fieldLogLevel"
+            ),
+        )
+
+
+app = core.App()
+AppSyncStack(app, "AppSyncStack")
+app.synth()

cdk_integration_tests/src/python/AppSyncLogging/pass.py

@@ -0,0 +1,25 @@
+from aws_cdk import core
+from aws_cdk import aws_appsync as appsync
+
+class AppSyncStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define the GraphQL API using CfnGraphQLApi
+        graphql_api = appsync.CfnGraphQLApi(
+            self,
+            "AppSyncGraphQLApi",
+            name="MyAppSyncAPI",
+            authentication_type="API_KEY",  # You can change the authentication type
+            log_config=appsync.CfnGraphQLApi.LogConfigProperty(
+                cloud_watch_logs_role_arn="cloudWatchLogsRoleArn",
+                exclude_verbose_content=False,
+                field_log_level="fieldLogLevel"
+            ),
+        )
+
+
+app = core.App()
+AppSyncStack(app, "AppSyncStack")
+app.synth()

cdk_integration_tests/src/python/AthenaWorkgroupConfiguration/fail__1__.py

@@ -0,0 +1,28 @@
+from aws_cdk import core
+from aws_cdk import aws_athena as athena
+
+class AthenaStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an Athena WorkGroup
+        workgroup = athena.CfnWorkGroup(
+            self,
+            "MyAthenaWorkGroup",
+            name="my-workgroup",
+            description="My Athena WorkGroup",
+            state="ENABLED",  # You can change the state
+            work_group_configuration=athena.CfnWorkGroup.WorkGroupConfigurationProperty(
+                additional_configuration="additionalConfiguration",
+                bytes_scanned_cutoff_per_query=123,
+                customer_content_encryption_configuration=athena.CfnWorkGroup.CustomerContentEncryptionConfigurationProperty(
+                    kms_key="kmsKey"
+                ),
+                enforce_work_group_configuration=False,
+            )
+        )
+
+app = core.App()
+AthenaStack(app, "AthenaStack")
+app.synth()

cdk_integration_tests/src/python/AthenaWorkgroupConfiguration/pass.py

@@ -0,0 +1,28 @@
+from aws_cdk import core
+from aws_cdk import aws_athena as athena
+
+class AthenaStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an Athena WorkGroup
+        workgroup = athena.CfnWorkGroup(
+            self,
+            "MyAthenaWorkGroup",
+            name="my-workgroup",
+            description="My Athena WorkGroup",
+            state="ENABLED",  # You can change the state
+            work_group_configuration=athena.CfnWorkGroup.WorkGroupConfigurationProperty(
+                additional_configuration="additionalConfiguration",
+                bytes_scanned_cutoff_per_query=123,
+                customer_content_encryption_configuration=athena.CfnWorkGroup.CustomerContentEncryptionConfigurationProperty(
+                    kms_key="kmsKey"
+                ),
+                enforce_work_group_configuration=True,
+            )
+        )
+
+app = core.App()
+AthenaStack(app, "AthenaStack")
+app.synth()

cdk_integration_tests/src/python/CloudTrailLogValidation/fail__1__.py

@@ -0,0 +1,24 @@
+from aws_cdk import core
+from aws_cdk import aws_cloudtrail as cloudtrail
+from aws_cdk import aws_iam as iam
+
+class CloudTrailStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an AWS CloudTrail trail using CfnTrail
+        trail = cloudtrail.CfnTrail(
+            self,
+            "MyCloudTrail",
+            is_logging=True,
+            enable_log_file_validation=False,
+            management_events=[
+                cloudtrail.ReadWriteType.WRITE_ONLY,
+            ],
+            include_global_service_events=True,
+        )
+
+app = core.App()
+CloudTrailStack(app, "CloudTrailStack")
+app.synth()

cdk_integration_tests/src/python/CloudTrailLogValidation/pass.py

@@ -0,0 +1,24 @@
+from aws_cdk import core
+from aws_cdk import aws_cloudtrail as cloudtrail
+from aws_cdk import aws_iam as iam
+
+class CloudTrailStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Create an AWS CloudTrail trail using CfnTrail
+        trail = cloudtrail.CfnTrail(
+            self,
+            "MyCloudTrail",
+            is_logging=True,
+            enable_log_file_validation=True,  # Enable log file validation
+            management_events=[
+                cloudtrail.ReadWriteType.WRITE_ONLY,
+            ],
+            include_global_service_events=True,
+        )
+
+app = core.App()
+CloudTrailStack(app, "CloudTrailStack")
+app.synth()

cdk_integration_tests/src/python/DocDBAuditLogs/fail__1__.py

@@ -0,0 +1,22 @@
+from aws_cdk import core
+from aws_cdk import aws_docdb as docdb
+
+class DocDBStack(core.Stack):
+
+    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Define the DocDB Cluster Parameter Group
+        db_parameter_group = docdb.CfnDBClusterParameterGroup(
+            self,
+            "DocDBClusterParameterGroup",
+            description="Custom DocDB Cluster Parameter Group",
+            family="docdb4.0",
+            parameters={
+                "audit_logs": "disabled",
+            }
+        )
+
+app = core.App()
+DocDBStack(app, "DocDBStack")
+app.synth()