def create_firehose_role()

in src/integ_test_resources/ios/sdk/integration/cdk/cdk_integration_tests_ios/firehose_stack.py [0:0]

• 51 lines of code
• 1 McCabe index (conditional complexity)

    def create_firehose_role(self, delivery_bucket) -> str:
        """
        Creates an IAM role to allow Kinesis to deliver records to S3, per
        https://docs.aws.amazon.com/firehose/latest/dev/controlling-access.html

        :param delivery_bucket: The destination bucket
        :return: IAM Role ARN
        """
        firehose_role = aws_iam.Role(
            self,
            "integ_test_firehose_delivery_role",
            assumed_by=aws_iam.ServicePrincipal("firehose.amazonaws.com"),
        )

        firehose_role.add_to_policy(
            aws_iam.PolicyStatement(
                effect=aws_iam.Effect.ALLOW,
                actions=[
                    "s3:AbortMultipartUpload",
                    "s3:GetBucketLocation",
                    "s3:GetObject",
                    "s3:ListBucket",
                    "s3:ListBucketMultipartUploads",
                    "s3:PutObject",
                ],
                resources=[delivery_bucket.bucket_arn, f"{delivery_bucket.bucket_arn}/*"],
            )
        )

        firehose_role.add_to_policy(
            aws_iam.PolicyStatement(
                effect=aws_iam.Effect.ALLOW,
                actions=[
                    "kinesis:DescribeStream",
                    "kinesis:GetShardIterator",
                    "kinesis:GetRecords",
                    "kinesis:ListShards",
                ],
                resources=[f"arn:aws:kinesis:{self.region}:{self.account}:stream/*"],
            )
        )

        log_stream_arn = ":".join(
            [
                "arn:aws:logs",
                self.region,
                self.account,
                "log-group",
                FirehoseStack.LOG_GROUP_NAME,
                "log-stream",
                FirehoseStack.LOG_STREAM_NAME,
            ]
        )
        firehose_role.add_to_policy(
            aws_iam.PolicyStatement(
                effect=aws_iam.Effect.ALLOW,
                actions=["logs:PutLogEvents"],
                resources=[log_stream_arn],
            )
        )
        return firehose_role.role_arn