Filtering and Processing CloudTrail Logs

[Bin-security]

Hey @jshlbrd and the Substation team. We are ingesting CloudTrail logs through Substation. Right now it uses the transfer transformer. CloudTrail logs are quite big and they are tricky to process and filter in the sense that it is just a giant json string, which is not even line-separated. It can be quite large and up to 50 MB in compressed format. Do you guys do filtering and processing on the CloudTrail logs? What are good ways to do that?

[jshlbrd]

Hey @Bin-security 👋

As you've probably seen, CloudTrail events are written to S3 objects in a single JSON event that contains an array of records. Here's an example:

{
	"Records": [
		{
			"eventVersion": "1.05",
			"userIdentity": {
				"type": "IAMUser",
				"principalId": "AIDABCD1234567890",
				"arn": "arn:aws:iam::123456789012:user/jdoe",
				"accountId": "123456789012",
				"accessKeyId": "AKIAIJKLMNOPQRSTU",
				"userName": "jdoe"
			},
			"eventTime": "2023-04-19T08:30:45Z",
			"eventSource": "s3.amazonaws.com",
			"eventName": "PutObject",
			"awsRegion": "us-west-2",
			"sourceIPAddress": "203.0.113.1",
			"userAgent": "aws-cli/1.16.310 Python/3.7.3 Darwin/18.7.0 botocore/1.12.200",
			"requestParameters": {
				"bucketName": "my-s3-bucket",
				"key": "example-file.txt",
				"contentType": "text/plain",
				"contentLength": 1024
			},
			"responseElements": {
				"x-amz-request-id": "ABCD123EFGHIJKL",
				"x-amz-id-2": "abc123def456"
			},
			"s3": {
				"s3SchemaVersion": "1.0",
				"configurationId": "MyConfigRule",
				"bucket": {
					"name": "my-s3-bucket",
					"ownerIdentity": {
						"principalId": "A1BCDEFGHIJKLMNOP"
					},
					"arn": "arn:aws:s3:::my-s3-bucket"
				},
				"object": {
					"key": "example-file.txt",
					"size": 1024,
					"eTag": "d41d8cd98f00b204e9800998ecf8427e",
					"versionId": "null"
				}
			}
		}
	]
}

You'll need to transform these events to turn the Records into individual events. You can use the batch transform and Expand processor to do that. Here's an example that you can test on substation.run along with the event above:

Configuration:

[
    // pretty print is used to convert the multi-line JSON to a single line.
    // this is not needed when ingesting data from CloudTrail, only the expand
    // processor is required.
    sub.interfaces.processor.pretty_print(
    	options={ direction: 'from' }
    ),
    sub.interfaces.processor.expand(
    	settings={ key: 'Records' },
    ),
]

By the way, next time feel free to use @brexhq/substation for reaching out to the team!

[Bin-security]

Thanks Josh.

[jshlbrd]

No problem!