Python AWS CDK: Creating a Trail in CloudTrail

CloudTrail is a very useful tool in AWS especially for more sensitive environments where logging is essential. AWS CDK is a great tool that lets you create your stack by writing actual code instead of writing YAML or JSON files. What happens when you combined the two? Well, right now CDK's CloudTrail support is in developer preview so accomplishing some of the things you might want gets a little tricky. Let's run through a few things and see what CDK supports and where you have to fall back to the Cfn* functions.

aws_cdk.aws_cloudtrail.Trail

The basic construct for creating a Trail in CloudTrail is using the Trail construct. This has some basic support for creating a trail. Here is an example of creating a basic Trail using CDK:

from aws_cdk import core, aws_cloudtrail

...

    trail = aws_cloudtrail.Trail(
      self, "MyTrail",
      management_events=aws_cloudtrail.ReadWriteType.ALL
    )

In the above example, we let CDK do a lot of the magic for us. It will create a trail, an S3 bucket, and start logging stuff. We've told it to also include all events that it can log in there. This is a pretty basic use case but will get you up and running with CloudTrail. But what if you want to do different things? Let's break down a few of them.

Creating your own Bucket

This is currently supported. You have the option of passing in a bucket as a parameter to the Trail. It will use this bucket as the storage destination for all the logs in the Trail. There is one catch to this. If you want to use your own bucket, you'll need to create the correct policies to allow CloudTrail to write to this bucket.

from aws_cdk import core, aws_s3, aws_cloudtrail, aws_iam

...

        log_bucket = aws_s3.Bucket(
            self, "MyLogBucket"
        )

        policyStatement = aws_iam.PolicyStatement(
            actions=["s3:GetBucketAcl"],
            resources=[logging_bucket.arn_for_objects("")]
        )
        aws_iam.Policy(
            self, "MyLogsPolicy",
            statements=[policyStatement]
        )

        get_bucket_acl_policy = aws_iam.PolicyStatement(
            effect=aws_iam.Effect.ALLOW,
            actions=["s3:GetBucketAcl"],
            resources=[log_bucket.bucket_arn],
        )
        get_bucket_acl_policy.add_service_principal(
            "cloudtrail.amazonaws.com"
        )

        put_log_policy = aws_iam.PolicyStatement(
            effect=aws_iam.Effect.ALLOW,
            actions=["s3:PutObject"],
            resources=[log_bucket.arn_for_objects(
                f"AWSLogs/{core.Stack.of(self).account}/*"
            )],
            conditions={
                "StringEquals": {
                    "s3:x-amz-acl":"bucket-owner-full-control"
                }
            }
        )
        put_log_policy.add_service_principal(
            "cloudtrail.amazonaws.com"
        )

        log_bucket.add_to_resource_policy(get_bucket_acl_policy)
        log_bucket.add_to_resource_policy(put_log_policy)

        trail = aws_cloudtrail.Trail(
            self, "MyTrail",
            bucket=log_bucket,
            management_events=aws_cloudtrail.ReadWriteType.ALL
        )

As you can see, choosing to set up the bucket yourself requires a bit more code to get the policies correct so the trail can write to the bucket. However, that may be something you need. Why would you need this? Here are some reasons:
  1. You need the logs to be encrypted
  2. You require greater control over the IAM and/or bucket policies
  3. You require greater control over the configuration of that bucket (retention period, versioning, etc.)
So, for some, creating your own bucket is worth it. For others it may not be.

Sending to CloudWatch

CDK's Trail construct has an option to let you send your logs to CloudWatch automatically which is quite a useful feature. However, there is at least one thing that may cause you to think twice about turning this feature on. The main issue here is that it doesn't (yet) expose the log group that it creates for you. So, if you want to create alarms and metrics based on that log group, you're out of luck with the current functionality. However, there is a way around this: create the log group and accompanying rights yourself in CDK.

So, how do you do that? Here's what you'll need:

  1. A LogGroup
  2. An IAM Role
  3. An IAM Policy
  4. A CfnTrail

from aws_cdk import core, aws_s3, aws_cloudtrail, aws_iam, aws_logs

...

        log_bucket = aws_s3.Bucket(
            self, "MyLogBucket"
        )

        policyStatement = aws_iam.PolicyStatement(
            actions=["s3:GetBucketAcl"],
            resources=[log_bucket.arn_for_objects("")]
        )
        aws_iam.Policy(
            self, "MyLogsPolicy",
            statements=[policyStatement]
        )

        get_bucket_acl_policy = aws_iam.PolicyStatement(
            effect=aws_iam.Effect.ALLOW,
            actions=["s3:GetBucketAcl"],
            resources=[log_bucket.bucket_arn],
        )

        get_bucket_acl_policy.add_service_principal(
            "cloudtrail.amazonaws.com"
        )

        put_log_policy = aws_iam.PolicyStatement(
            effect=aws_iam.Effect.ALLOW,
            actions=["s3:PutObject"],
            resources=[log_bucket.arn_for_objects(
                f"AWSLogs/{core.Stack.of(self).account}/*"
            )],
            conditions={
                "StringEquals": {
                    "s3:x-amz-acl":"bucket-owner-full-control"
                }
            }
        )

        put_log_policy.add_service_principal(
            "cloudtrail.amazonaws.com
        )

        log_bucket.add_to_resource_policy(get_bucket_acl_policy)
        log_bucket.add_to_resource_policy(put_log_policy)

        log_group = aws_logs.LogGroup(
            scope=self,
            id="MyLogGroup",
            log_group_name="MyLogGroup"
        )
        
        role = aws_iam.Role(
            self,
            "MyTrailLogsRole",
            role_name="MyTrailLogsRole",
            assumed_by=aws_iam.ServicePrincipal(
                "cloudtrail.amazonaws.com"
            )
        )
        policyStatement = aws_iam.PolicyStatement(
            actions=["logs:PutLogEvents", "logs:CreateLogStream"],
            resources=[log_group.log_group_arn]
        )
        role.add_to_policy(policyStatement)
        aws_iam.Policy(
            self,
            "MyTrailLogsDefaultPolicy",
            policy_name="MyTrailLogsDefaultPolicy",
            statements=[policyStatement]
        )

        aws_cloudtrail.CfnTrail(
            scope=self,
            id="MyTrail",
            is_logging=True,
            s3_bucket_name=log_bucket.bucket_name,
            cloud_watch_logs_log_group_arn=log_group.log_group_arn,
            cloud_watch_logs_role_arn=role.role_arn,
            include_global_service_events=True
        )

And, with that, you've created all you need for a basic Trail that you send to CloudWatch. There are a few things to note.

If you wish to use an Event Selector, you'll need to use the aws_cloudtrail.CfnTrail.EventSelectorProperty construct as well as the aws_cloudtrail.CfnTrail.DataResourceProperty construct. These aren't mentioned in the Python documentation but you can cross-reference with the normal documentation and the TypeScript documentation to discover that they need to be used.

Comments