Class: ActiveElasticJob::Rack::SqsMessageConsumer

Inherits:
Object
  • Object
show all
Defined in:
lib/active_elastic_job/rack/sqs_message_consumer.rb

Overview

This middleware intercepts requests which are sent by the SQS daemon running in Amazon Elastic Beanstalk worker environments. It does this by looking at the User-Agent header. Furthermore, it verifies the digest which is sent along with a legit SQS message, and passed as an HTTP header in the resulting request. The digest is based on Rails’ secrets.secret_key_base. Therefore, the application running in the web environment, which generates the digest, and the application running in the worker environment, which verifies the digest, have to use the same secrets.secret_key_base setting.

Constant Summary collapse

USER_AGENT_PREFIX = 
'aws-sqsd'.freeze
DIGEST_HEADER_NAME = 
'HTTP_X_AWS_SQSD_ATTR_MESSAGE_DIGEST'.freeze
ORIGIN_HEADER_NAME = 
'HTTP_X_AWS_SQSD_ATTR_ORIGIN'.freeze
CONTENT_TYPE = 
'application/json'.freeze
CONTENT_TYPE_HEADER_NAME = 
'Content-Type'.freeze
OK_RESPONSE_CODE = 
'200'.freeze
INSIDE_DOCKER_CONTAINER = 
`cat /proc/1/cgroup` =~ /docker/
DOCKER_HOST_IP = 
"172.17.0.1".freeze

Instance Method Summary collapse

Constructor Details

#initialize(app) ⇒ SqsMessageConsumer

:nodoc:



25
26
27
 
# File 'lib/active_elastic_job/rack/sqs_message_consumer.rb', line 25

def initialize(app) #:nodoc:
  @app = app
end

Instance Method Details

#call(env) ⇒ Object

:nodoc:



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/active_elastic_job/rack/sqs_message_consumer.rb', line 29

def call(env) #:nodoc:
  request = ActionDispatch::Request.new env
  if enabled? && aws_sqsd?(request) && originates_from_gem?(request)
    unless request.local? || sent_from_docker_host?(request)
      m = "Accepts only requests from localhost for job processing".freeze
      return ['403', {CONTENT_TYPE_HEADER_NAME => 'text/plain' }, [ m ]]
    end
    begin
      verify!(request)
      job = JSON.load(request.body)
      ActiveJob::Base.execute(job)
    rescue ActiveElasticJob::MessageVerifier::InvalidDigest => e
      return [
        '403',
        {CONTENT_TYPE_HEADER_NAME => 'text/plain' },
        ["incorrect digest"]]
    end
    return [
      OK_RESPONSE_CODE ,
      {CONTENT_TYPE_HEADER_NAME => CONTENT_TYPE },
      [ '' ]]
  end
  @app.call(env)
end