Class: Threatstack::Jobs::EventSubmitter

Inherits:
Object
  • Object
show all
Includes:
Singleton, Constants
Defined in:
lib/jobs/event_submitter.rb

Overview

Singleton class that creates a job to submit events asynchronously

Constant Summary collapse

SUBMITTER_NAME = 
'EventSubmitter'

Constants included from Constants

Constants::AGENT_ID, Constants::AGENT_INSTANCE_ID, Constants::AGENT_NAME, Constants::APPSEC_BASE_URL, Constants::APPSEC_EVENTS_URL, Constants::ATTACK, Constants::AWS_METADATA_URL, Constants::BLOCK_SQLI, Constants::BLOCK_XSS, Constants::CGI_VARIABLES, Constants::DEPENDENCIES, Constants::DETECTED_NOT_BLOCKED, Constants::DISABLED, Constants::DROP_FIELDS, Constants::ENVIRONMENT, Constants::EVENTS_PER_REQ, Constants::INSTRUMENTATION, Constants::IPV4, Constants::IPV6, Constants::JOB_INTERVAL, Constants::LOG_COLORS, Constants::LOG_LEVEL, Constants::MANUAL_INIT, Constants::REDACTED, Constants::REQUEST_BLOCKED, Constants::RUBY, Constants::SQLI, Constants::TRUTHY, Constants::XSS

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Constants

env, is_truthy

Constructor Details

#initialize ⇒ EventSubmitter

Returns a new instance of EventSubmitter.



24
25
26
27
28
 
# File 'lib/jobs/event_submitter.rb', line 24

def initialize
  @pid = nil
  @mutex = Mutex.new
  @accumulator = Threatstack::Events::EventAccumulator.instance
end

Instance Attribute Details

#job ⇒ Object (readonly)

Returns the value of attribute job.



22
23
24
 
# File 'lib/jobs/event_submitter.rb', line 22

def job
  @job
end

Instance Method Details

#fetch_events ⇒ Object 



91
92
93
94
95
96
97
98
 
# File 'lib/jobs/event_submitter.rb', line 91

def fetch_events
  # check total number of events first
  total_events = @accumulator.events.length
  return [] unless total_events > 0

  # retrieve events from the accumulator in a thread-safe manner
  @accumulator.remove_events total_events
end

#queue_event(event) ⇒ Object 



86
87
88
89
 
# File 'lib/jobs/event_submitter.rb', line 86

def queue_event(event)
  start
  @accumulator.add_event event
end

#start ⇒ Object

start recurrent job



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/jobs/event_submitter.rb', line 31

def start
  pid = Process.pid
  if @pid != pid
    @mutex.synchronize do
      if @pid != pid
        @logger = Threatstack::Utils::TSLogger.create SUBMITTER_NAME
        @logger.debug 'Launching an EventSubmitter instance'
        # clear events in case the accumulator was forked and already had queued events
        @accumulator.clear_events
        # submit every JOB_INTERVAL seconds
        @job = RecurrentJob.new(@logger, JOB_INTERVAL, 5) { submit }
        @pid = pid
      end
    end
  end
end

#submit ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/jobs/event_submitter.rb', line 48

def submit
  @logger.debug 'Starting event submission'

  # fetch events from the accumulator
  events = fetch_events
  @logger.debug "Found a total of #{events.length} events"

  if events.empty?
    @logger.debug 'No events to report'
    return
  end

  # split events array into groups of EVENTS_PER_REQ
  events.each_slice(EVENTS_PER_REQ) do |group|
    # convert payload to JSON
    json_payload = "[#{group.collect(&:to_json_string).join(', ')}]"
    # submit http request
    uri = URI.parse(APPSEC_BASE_URL + APPSEC_EVENTS_URL)
    headers = {
      'Content-Type' => 'application/json',
      'bluefyre-agent-id' => AGENT_ID,
      'bluefyre-agent-instance-id' => AGENT_INSTANCE_ID
    }
    http = Net::HTTP.new(uri.host, uri.port)
    http.use_ssl = true
    http.verify_mode = OpenSSL::SSL::VERIFY_NONE
    req = Net::HTTP::Post.new(uri.request_uri, headers)
    req.body = json_payload
    @logger.debug "Sending #{group.length} events with payload: #{json_payload}"
    begin
      resp = http.request(req)
      @logger.debug "Sent #{group.length} events, HTTP code: #{resp.code}"
    rescue StandardError => e
      @logger.error "StandardError: #{e.inspect}"
    end
  end
end