[aws-kinesis] Read permissions to stream doesn't include kinesis:DescribeStream

[Alexander-D-Jensen]

When granting an IAM role read permissions on a Kinesis stream, the resulting policy does not include the "kinesis:DescribeStream" permission, which is needed for at least Kinesis Firehose to read from it, perhaps other consumers as well.

Reproduction Steps

from aws_cdk import (
    core,
    aws_kinesis as kds,
    aws_iam as iam,
)
class DFPStack(core.NestedStack):
    def __init__(
        self, scope: core.Construct, id: str, *, stream=kds.Stream, **kwargs
    ) -> None:

        self.kfh_reader = iam.Role(
            self,
            "KFHReader",
            assumed_by=iam.ServicePrincipal(service="firehose.amazonaws.com"),
        )
        self.stream = stream
        self.stream.grant_read(self.kfh_reader)

Results in the following output:

"KFHReaderDefaultPolicy98F05724": {
      "Type": "AWS::IAM::Policy",
      "Properties": {
        "PolicyDocument": {
          "Statement": [
            {
              "Action": [
                "kinesis:DescribeStreamSummary",
                "kinesis:GetRecords",
                "kinesis:GetShardIterator",
                "kinesis:ListShards",
                "kinesis:SubscribeToShard"
              ],
              "Effect": "Allow",
              "Resource": {
                "Ref": "referencetoLongboatkinesisNestedStackkinesisNestedStackResourceFA86427BOutputsLongboatkinesisstreamDE86A4D8Arn"
              }
            }
          ],
          "Version": "2012-10-17"
        },
        "PolicyName": "KFHReaderDefaultPolicy98F05724",
        "Roles": [
          {
            "Ref": "KFHReader8A09BD1B"
          }
        ]
      },
      "Metadata": {
        "aws:cdk:path": "Longboat/DFPStack/KFHReader/DefaultPolicy/Resource"
      }
    },

What did you expect to happen?

I expected the resulting IAM role and policy to actually be able to read from the supplied Kinesis stream when used by Kinesis Firehose. This would require the "kinesis:DescribeStream" permission to be given.

What actually happened?

Stack creation fails, because the "kinesis:DescribeStream" permission is lacking.

Cloudformation gives the following error:
Role arn:aws:iam::<account>:role/<role_name> is not authorized to perform: kinesis:DescribeStream on resource arn:aws:kinesis:eu-west-1:<account>:stream/<stream_name>. (Service: Firehose, Status Code: 400, Request ID: <request_id>, Extended Request ID: <extended_request_id>

Environment

• CLI Version: 1.66.0 (build 459488d)
• Framework Version: 1.66.0
• Node.js Version: v12.19.0
• OS: KDE Neon 5.19 (Ubuntu 20.04 focal)
• Language (Version): Python 3.8.5

Other

---

This is 🐛 Bug Report

[iliapolo]

@shivlaks Do you know why was the DescribeStream action explicitly removed in this commit?

@Alexander-D-Jensen - As a workaround, you can use Custom Permissions to add this action.

[Alexander-D-Jensen]

@iliapolo

I have done so afterwards, via the following:
self.stream.grant(self.kfh_reader, "kinesis:DescribeStream")

However, I run into issues with Cloudformation, which tries to create the Firehose before the IAM policy is ready, and then it fails.

In other words, despite adding the permissions explicitly, I am no closer to a successful deployment of the stack at that point.

Finally, I've arrived at the following workaround:

grant_describe = self.stream.grant(self.kfh_reader, "kinesis:DescribeStream")
<create firehose here>
grant_describe.apply_before(self.firehose)

I am not sure if this is within the scope of CDK to handle or not, though I suspect the better solution would be for CloudFormation to handle it (correctly).

[iliapolo]

@Alexander-D-Jensen Thanks, we'll take a look.

Do you mind sharing the entire code you use? Including where you define the firehose and the stream itself?

[Alexander-D-Jensen]

Sure @iliapolo!
I've redacted some info (account numbers, company name etc.) and a few tags, as well as services, modules etc. which aren't of any consequence. Below should be as close to a complete setup needed to replicate the error as I can easily get.

This is stack A, the stack which had the issue:

from aws_cdk import (
    core,
    aws_kinesis as kds,
    aws_sqs as sqs,
    aws_iam as iam,
    aws_s3 as s3,
    aws_kinesisfirehose as kfh,
)

import os
import pathlib

from .settings import ACCOUNT_ALIAS

commit_stream = os.popen(
    "git log -n 1 --pretty=format:%H -- {0}".format(pathlib.Path(__file__))
)
commit_hash = commit_stream.read()

tags = {
    "Last_Change": commit_hash,
    "Feature": "DFP Handover",
}


dfp_buckets = {
    "Dev": "arn:aws:s3:::<bucket_name>",
    "Test": "arn:aws:s3:::<bucket_name>",
    "Prod": "arn:aws:s3:::<bucket_name>",
}


class DFPStack(core.NestedStack):
    def __init__(
        self, scope: core.Construct, id: str, *, stream=kds.Stream, **kwargs
    ) -> None:
        super().__init__(scope, id, **kwargs)
        for k, v in tags.items():
            core.Tags.of(self).add(k, v)

        self.dfp_bucket = s3.Bucket.from_bucket_arn(
            self, "DFPBucket", bucket_arn=dfp_buckets[ACCOUNT_ALIAS]
        )

        self.kfh_reader = iam.Role(
            self,
            "KFHReader",
            assumed_by=iam.ServicePrincipal(service="firehose.amazonaws.com"),
        )
        self.kfh_writer = iam.Role(
            self,
            "KFHWriter",
            assumed_by=iam.ServicePrincipal(service="firehose.amazonaws.com"),
        )

        self.stream = stream

        self.stream.grant_read(self.kfh_reader)
        grant_describe = self.stream.grant(self.kfh_reader, "kinesis:DescribeStream")

        self.kfh_writer.add_to_policy(
            iam.PolicyStatement(
                actions=["s3:PutObjectAcl"], resources=[self.dfp_bucket.bucket_arn]
            )
        )
        self.dfp_bucket.grant_read_write(self.kfh_writer)

        self.firehose = kfh.CfnDeliveryStream(
            self,
            "DFPDelivery",
            delivery_stream_type="KinesisStreamAsSource",
            extended_s3_destination_configuration=kfh.CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty(
                bucket_arn=self.dfp_bucket.bucket_arn,
                role_arn=self.kfh_writer.role_arn,
                buffering_hints=kfh.CfnDeliveryStream.BufferingHintsProperty(
                    interval_in_seconds=300, size_in_m_bs=128
                ),
                compression_format="GZIP",
                prefix="<company_name>/v1/",
            ),
            kinesis_stream_source_configuration=kfh.CfnDeliveryStream.KinesisStreamSourceConfigurationProperty(
                kinesis_stream_arn=self.stream.stream_arn,
                role_arn=self.kfh_reader.role_arn,
            ),
        )
        self.firehose.CloudWatchLoggingOptionsProperty(enabled=True)

        grant_describe.apply_before(self.firehose)

        core.CfnOutput(scope=self, id="WriterRoleArn", value=self.kfh_writer.role_arn)

This is stack B, which creates the Kinesis stream

from aws_cdk import core, aws_kinesis as kds, aws_sqs as sqs, aws_iam as iam

import os
import pathlib

from .settings import ACCOUNT_ALIAS

commit_stream = os.popen(
    "git log -n 1 --pretty=format:%H -- {0}".format(pathlib.Path(__file__))
)
commit_hash = commit_stream.read()

tags = {
    "Last_Change": commit_hash,
    "Feature": "Kinesis Stream",
}

class KinesisStack(core.NestedStack):
    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)
        for k, v in tags.items():
            core.Tags.of(self).add(k, v)

        self.stream = kds.Stream(
            self,
            "stream",
            retention_period=core.Duration.hours(72),
            shard_count=6 if ACCOUNT_ALIAS == "Prod" else 1,
        )

        self.dead_letter_queue = sqs.Queue(
            self,
            "DeadLetterQueue",
        )

And finally, we have the stack which coordinates the whole thing:

from aws_cdk import core, aws_iam

from .kinesis_stack import KinesisStack
from .dfp_stack import DFPStack

tags = {"Service": "Something"}


class LongboatStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)
        for k, v in tags.items():
            core.Tags.of(self).add(k, v)

        stream = KinesisStack(self, "kinesis")
        self.dfp_stack = DFPStack(self, "DFPStack", stream=self.stream.stream)

I hope this helps :)

[iliapolo]

Hi @Alexander-D-Jensen - thanks for sharing the code. It was very helpful.

So there are actually two separate issues here:

First is that stream.grantRead does not grant kinesis:DescribeStream operation. This was actually intentional because the DescribeStream API is deprecated in favor of DescribeStreamSummary. Seems like CfnDeliveryStream still uses this deprecated API however.

We will consider adding this action back to the policy.

Second is that the CfnDeliveryStream is created before the necessary policies are applied to the kinesis reader role.
What happens here is that the reader policy is a separate resource (AWS::IAM::Policy) that is created after the role is created, and since the CfnDeliveryStream depends only on the role, not the policy, this race condition occurs.

This type of thing is exactly what an L2 for a CfnDeliveryStream can solve, but until we have that, the way to tackle this problem is exactly what you did.

I've created an issue for the L2 construct to make sure we consider this scenario when we start working on it.

[github-actions]

⚠️COMMENT VISIBILITY WARNING⚠️

Comments on closed issues are hard for our team to see.
If you need more assistance, please either tag a team member or open a new issue that references this one.
If you wish to keep having a conversation with other community members under this issue feel free to do so.