Using Amazon Data Firehose with Amazon PrivateLink

Sign in to the Amazon Web Services Management Console and open the Amazon VPC console at https://console.amazonaws.cn/vpc/.

In the VPC Dashboard choose Endpoints.

Choose Create Endpoint.

In the list of service names, choose com.amazonaws.your_region.kinesis-firehose.

Choose the VPC and one or more subnets in which to create the endpoint.

Choose one or more security groups to associate with the endpoint.

For Policy, choose Custom and paste the following policy:

{
    "Statement": [
        {
            "Sid": "Allow-only-specific-PrivateAPIs",
            "Principal": "*",
            "Action": [
                "firehose:ListDeliveryStreams"
            ],
            "Effect": "Allow",
            "Resource": [
                "*"
            ]
        },
        {
            "Sid": "Allow-only-specific-PrivateAPIs",
            "Principal": "*",
            "Action": [
                "firehose:DescribeDeliveryStream"
            ],
            "Effect": "Deny",
            "Resource": [
                "*"
            ]
        }
    ]
}

Choose Create endpoint.

Open the IAM console at https://console.amazonaws.cn/iam/.

In the left pane, chose Roles and then choose Create role.

Under Select type of trusted entity, leave the default selection Amazon service.

Under Choose the service that will use this role, choose Lambda.

Choose Next: Permissions.

In the list of policies, search for and add the two policies named AmazonLambdaVPCAccessExecutionRole and AmazonDataFirehoseReadOnlyAccess.

Choose Next: Tags. You don't need to add tags for the purpose of this exercise. Choose Next: Review.

Enter a name for the role, then choose Create role.

Open the Amazon Lambda console at https://console.amazonaws.cn/lambda/.

Choose Create function.

Choose Author from scratch.

Enter a name for the function, then set Runtime to Python 3.9 or higher.

Under Permissions, expand Choose or create an execution role.

In the Execution role list, choose Use an existing role.

In the Existing role list, choose the role you created above.

Choose Create function.

Under Function code, paste the following code.

    import json
    import boto3
    import os
    from botocore.exceptions import ClientError
     
    def lambda_handler(event, context):
        REGION = os.environ['AWS_REGION']
        client = boto3.client(
            'firehose',
            REGION
            
        )
        print("Calling list_delivery_streams with ListDeliveryStreams allowed policy.")
        delivery_stream_request = client.list_delivery_streams()
        print("Successfully returned list_delivery_streams request %s." % (
            delivery_stream_request
        ))
        describe_access_denied = False
        try:
            print("Calling describe_delivery_stream with DescribeDeliveryStream denied policy.")
            delivery_stream_info = client.describe_delivery_stream(DeliveryStreamName='test-describe-denied')
        except ClientError as e:
            error_code = e.response['Error']['Code']
            print ("Caught %s." % (error_code))
            if error_code == 'AccessDeniedException':
                describe_access_denied = True
        
        if not describe_access_denied:
            raise
        else:
            print("Access denied test succeeded.")

Under Basic settings, set the timeout to 1 minute.

Under Network, choose the VPC where you created the endpoint above, then choose the subnets and security group that you associated with the endpoint when you created it.

Near the top of the page, choose Save.

Choose Test.

Enter an event name, then choose Create.

Choose Test again. This causes the function to run. After the execution result appears, expand Details and compare the log output to the function code. Successful results show a list of the Firehose streams in the Region, as well as the following output:

Calling describe_delivery_stream.

AccessDeniedException

Access denied test succeeded.