Cloudflare audit logs to Sumologic

cloudflare_audit_to_sumologic.rb

# frozen_string_literal: true

require 'net/http'
require 'json'
require 'uri'

module Environment
  SUMOURL = ENV.fetch('SUMO_ENDPOINT')
  ORGID = ENV.fetch('CLOUDFLARE_ORG_ID')
  CLOUDFLAREAUTHEMAIL = ENV.fetch('CLOUDFLARE_AUTH_EMAIL')
  CLOUDFLAREAUTHKEY = ENV.fetch('CLOUDFLARE_AUTH_KEY')
  SOURCECATEGORYOVERRIDE = ENV.fetch('SOURCE_CATEGORY_OVERRIDE', 'none')
  SOURCEHOSTOVERRIDE = ENV.fetch('SOURCE_HOST_OVERRIDE', 'api.cloudflare.com')
  SOURCENAMEOVERRIDE = ENV.fetch('SOURCE_NAME_OVERRIDE', ORGID)
end

def variable_set?(var)
  !var.nil? && var != '' && var != 'none'
end

def sumo_meta_key
  source_category = Environment::SOURCECATEGORYOVERRIDE if variable_set?(Environment::SOURCECATEGORYOVERRIDE)
  source_host = Environment::SOURCEHOSTOVERRIDE if variable_set?(Environment::SOURCEHOSTOVERRIDE)
  source_name = Environment::SOURCENAMEOVERRIDE if variable_set?(Environment::SOURCENAMEOVERRIDE)

  "#{source_name}:#{source_category}:#{source_host}"
end

def post_to_sumologic(context:, messages:)
  messages_total = messages.keys.length
  messages_sent = 0
  message_errors = []

  url_object = URI(Environment::SUMOURL)

  finalize_context = lambda do
    total = messages_sent + message_errors.length

    if total == messages_total
      puts "messages_sent: #{messages_sent} message_errors: #{message_errors.length}"

      raise "errors: #{message_errors}" unless message_errors.empty?
    end
  end

  messages.each do |key, logs|
    header_array = key.split(':')

    headers = {
      'X-Sumo-Name': header_array[0],
      'X-Sumo-Category': header_array[1],
      'X-Sumo-Host': header_array[2]
    }

    Net::HTTP.start(url_object.host, url_object.port, use_ssl: true) do |http|
      request = Net::HTTP::Post.new(url_object)

      headers.each { |header, value| request.public_send(:[]=, header, value) }

      request.body = logs.map do |msg|
        JSON.dump(msg)
      end.join("\n")

      response = http.request(request)

      if response.code == '200'
        messages_sent += 1
      else
        message_errors.push('HTTP Return code ' + res.statusCode)
      end
    rescue StandardError => e
      message_errors.push(e.message)
    ensure
      finalize_context.call
    end
  end
end

def sumo_endpoint_invalid?(url_object)
  url_object.scheme != 'https' || url_object.host.nil? || url_object.path.nil?
end

def format_cloudflare_to_sumologic(str)
  message_list = {}

  logs = JSON.parse(str)['result']

  puts "Log events: #{logs.length}"

  logs.each do |log|
    next if log.empty?

    metadata_key = sumo_meta_key

    if message_list.key?(metadata_key)
      message_list[metadata_key].push(log)
    else
      message_list[metadata_key] = [log]
    end
  end

  message_list
end

def handler(event:, context:)
  if sumo_endpoint_invalid?(URI(Environment::SUMOURL))
    raise "Invalid SUMO_ENDPOINT environment variable: #{Environment::SUMOURL}"
  end

  now = Time.now.utc

  puts "start_time: #{now}"
  puts "end_time: #{now}"

  headers = {
    'X-Auth-Email' => Environment::CLOUDFLAREAUTHEMAIL,
    'X-Auth-Key' => Environment::CLOUDFLAREAUTHKEY
  }

  url_object = URI("https://api.cloudflare.com/client/v4/organizations/#{Environment::ORGID}/audit_logs")
  params = {
    since: now.strftime('%F'),
    before: now.strftime('%F')
  }
  url_object.query = URI.encode_www_form(params)

  Net::HTTP.start(url_object.host, url_object.port, use_ssl: true) do |http|
    request = Net::HTTP::Get.new(url_object)
    headers.each { |header, value| request.send(:[]=, header, value) }

    response = http.request(request)
    puts "response.code: #{response.code}"

    message_list = format_cloudflare_to_sumologic(response.body)

    post_to_sumologic(context: context, messages: message_list) if response.code == '200'

    return 0
  rescue StandardError => e
    raise "Failed to push to Sumologic: #{e.message}"
  end
end