Set up CloudTrail alerts in ITSI

Configure CloudTrail in Amazon Web Services

  1. Install the Splunk Add-on for Amazon Web Services.
  2. From the application, select Configuration then Private account.
  3. Add the Key ID and Secret Key that you generated from configuring CloudTrail to send data to Splunk.
  4. Select Add.
  5. In the IAM Role tab, select Add. Enter the role ARN created from setting up CloudTrail.
  6. From the Inputs tab, select Create New Input, then Cloudtrail, then Generic S3. Enter your AWS account details and select Add.
  7. Verify events in Splunk with the search: 
    sourcetype=aws:cloudtrail

AWS CloudTrail webhook setup

  1. From the AWS console, select Lambda.
  2. Select Author from scratch.
  3. Add a name, and set the Runtime to Python 3.13.
  4. Select Create function.
  5. From the next page, select Add trigger.
  6. Set S3 as the source. Input the bucket where your logs are stored in the Bucket field.
  7. Select Add.
  8. From the Function overview page, select the Code tab and paste the following, replacing the HEC endpoint and token with your Splunk endpoint and tokens: 
    import json
import urllib.parse
import boto3
import gzip
from botocore.vendored import requests
import urllib3

s3 = boto3.client('s3')
http = urllib3.PoolManager()
splunk_hec_endpoint ='https://<<splunk_host>>:8088/services/collector/event'
splunk_hec_token = '<<hec_token>>'

def lambda_handler(event, context):
    # Get the object from the event and show its content type
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    try:
        response = s3.get_object(Bucket=bucket, Key=key)
        with gzip.GzipFile(fileobj=response['Body']) as gz:
            data = gz.read().decode('utf-8')
            data = json.loads(data)
            for cloudtrail_event in data['Records']:
                payload = {
                    'event': cloudtrail_event,
                    'sourcetype': 'aws:cloudtrail'
                }
                payload = json.dumps(payload)
                response = http.request(
                    'POST',
                    splunk_hec_endpoint,
                    body=payload,
                    headers={'Authorization': f'Splunk {splunk_hec_token}'}
                )
                if response.status != 200:
                    print(f'Failed to send event to Splunk: {response.data}')
                else:
                    print('Event sent to Splunk succesfully')
    except Exception as e:
        print(e)
        print('Error getting object. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket))
        raise e
  9. Select Deploy.
  10. Confirm that the webhook works by searching: 
    index=<hec_index> sourcetype="aws:cloudtrail"