cloudtrail decompresser

README.md

cloudtrail decompresser

install libraries

$ get https://gist.githubusercontent.com/ixixi/c9ee060a823e69d97e74/raw/7cc8d7a195111372bf4a543ecb51b112ff0994b4/cloudtrail_decomp.rb
$ gem install aws-sdk msgpack
$ export AWS_ACCESS_KEY_ID={YOUR_AWS_ACCESS_KEY_ID}
$ export AWS_SECRET_ACCESS_KEY={YOUR_AWS_SECRET_ACCESS_KEY}

add "out_exec_filter" to fluentd.conf

<match foo.before>
  type exec_filter
  command /path/to/cloudtrail_decomp.rb
  in_format json
  out_format msgpack
  tag foo.after
  flush_interval 1s
<match>

foo.before

{
  "Type" : "Notification",
  "MessageId" : "xxx",
  "TopicArn" : "arn:aws:sns:us-east-1:yyy:CloudFront_Splunk_Alerts",
  "Message" : "{\"s3Bucket\":\"smg-cloudtrail-us-east-1\",\"s3ObjectKey\":[\"AWSLogs/123456789/CloudTrail/us-west-1/2014/05/24/440742521871_CloudTrail_us-west-1_20140524T0340Z_abcabcabc.json.gz\"]}",
  "Timestamp" : "2014-05-24T06:12:44.115Z",
  "SignatureVersion" : "1",
  "Signature" : "XXXXXXX==",
  "SigningCertURL" : "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-XXXX.pem",
  "UnsubscribeURL" : "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:ZZZZ:CloudFront_Alerts:XXXXXX"
}

foo.after

{
  "Type" : "Notification",
  "MessageId" : "xxx",
  "TopicArn" : "arn:aws:sns:us-east-1:yyy:CloudFront_Splunk_Alerts",
  "Message" : "{\"s3Bucket\":\"smg-cloudtrail-us-east-1\",\"s3ObjectKey\":[\"AWSLogs/123456789/CloudTrail/us-west-1/2014/05/24/440742521871_CloudTrail_us-west-1_20140524T0340Z_abcabcabc.json.gz\"]}",
  "Timestamp" : "2014-05-24T06:12:44.115Z",
  "SignatureVersion" : "1",
  "Signature" : "XXXXXXX==",
  "SigningCertURL" : "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-XXXX.pem",
  "UnsubscribeURL" : "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:ZZZZ:CloudFront_Alerts:XXXXXX"
  "fullLogData" : [{"Records"=>[{"eventVersion"=>"1.01", "userIdentity"=>{"type"=>"Root", "principalId"=>"12345", "arn"=>"arn:aws:iam::12345:root",......................]
}

cloudtrail_decomp.rb

require 'msgpack'
require 'aws-sdk'
require 'json'
require 'zlib'
require 'logger'

$S3 = AWS::S3.new(
  :access_key_id => ENV['AWS_ACCESS_KEY_ID'],
  :secret_access_key => ENV['AWS_SECRET_ACCESS_KEY']
)

$logger = Logger.new('/tmp/cloudtrail_logfile.log')
$logger.level = Logger::DEBUG

$tmpfile = '/tmp/cloudtrail_tmpfile.json.gz'

def download( compressed_log , tmpfile)
  $logger.debug('download start')
  File.open(tmpfile, 'wb') do |file|
    compressed_log.read do |chunk|
      file.write(chunk)
    end
  end
  $logger.debug('download end')
end

def decompress(tmpfile)
  $logger.debug('decompress start')
  logdata = nil
  Zlib::GzipReader.open(tmpfile) do |gz|
    logdata = gz.read
  end
  $logger.debug('decompress end')
  JSON.parse(logdata)
end

def get_full_logdata( bucket_name, object_keys)

  bucket = $S3.buckets[bucket_name]

  full_logdata = []

  object_keys.each do | object_key|
    compressed_log = bucket.objects[object_key]
    download(compressed_log, $tmpfile)
    full_logdata << decompress($tmpfile)
  end
  full_logdata
end

STDIN.each_line do |line|
  $logger.debug('start')
  $logger.debug(line)
  begin
    event = JSON.parse(line)
    message = JSON.parse(event['Message'])
    event['fullLogData'] = get_full_logdata(message['s3Bucket'],message['s3ObjectKey'])
    $stdout.binmode.write(event.to_msgpack)
  rescue Exception => e
    $logger.error(e.class)
    $logger.error(e.message)
    $logger.error(e.backtrace.join("\n"))
    exit(1)
  end
  $logger.debug('end')
end