Kinesis Client

kinesis_client.rb

require 'json'
require 'openssl'
require 'net/http'
require 'net/https'
require 'time'
require 'stringio'
require 'zlib'

module Kinesis
  class Client

    SERVICE_NAME = 'kinesis'
    API_VERSION  = '2013-12-02'
    USER_AGENT   = "ruby"

    DEFAULT_TIMEOUT = 60

    def initialize(accessKeyId, secretAccessKey)
      @accessKeyId = accessKeyId
      @secretAccessKey = secretAccessKey
      @timeout = DEFAULT_TIMEOUT
    end

    def query(action, hash)
      req_body = JSON.dump(hash)
      date = Time.now.getutc

      headers = {
        'Content-Type'         => 'application/x-amz-json-1.1',
        'X-Amz-Target'         => "Kinesis_#{API_VERSION.gsub('-', '')}.#{action}",
        'Content-Length'       => req_body.length.to_s,
        'User-Agent'           => USER_AGENT,
        'Host'                 => 'kinesis.us-east-1.amazonaws.com',
        'X-Amz-Date'           => iso8601(date),
        'X-Amz-Content-Sha256' => hexhash(req_body),
        'Accept'               => '*/*',
        'Accept-Encoding'      => 'gzip',
     }

      headers['Authorization'] = authorization(date, headers, req_body)

      Net::HTTP.version_1_2
      http = Net::HTTP.new('kinesis.us-east-1.amazonaws.com', 443)
      http.use_ssl = true
      http.verify_mode = OpenSSL::SSL::VERIFY_NONE

      http.open_timeout = @timeout
      http.read_timeout = @timeout

      res_code = nil
      res_msg  = nil

      res_body = http.start do |w|
        req = Net::HTTP::Post.new('/', headers)
        req.body = req_body
        res = w.request(req)

        res_code = res.code.to_i
        res_msg  = res.message

        if res['Content-Encoding'] == 'gzip'
          StringIO.open(res.body, 'rb') do |f|
            Zlib::GzipReader.wrap(f).read
          end
        else
          res.body
        end
      end

      res_body
    end

    private

    def authorization(date, headers, body)
      headers = headers.sort_by {|name, value| name }

      # Task 1: Create a Canonical Request For Signature Version 4
      # http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html

      canonicalHeaders = headers.map {|name, value|
        name.downcase + ':' + value
      }.join("\n") + "\n"

      signedHeaders = headers.map {|name, value| name.downcase }.join(';')

      canonicalRequest = [
        'POST', # HTTPRequestMethod
        '/',    # CanonicalURI
        '',     # CanonicalQueryString
        canonicalHeaders,
        signedHeaders,
        hexhash(body),
      ].join("\n")

      # Task 2: Create a String to Sign for Signature Version 4
      # http://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html

      credentialScope = [
        date.strftime('%Y%m%d'),
        'us-east-1',
        SERVICE_NAME,
        'aws4_request',
      ].join('/')

      stringToSign = [
        'AWS4-HMAC-SHA256', # Algorithm
        iso8601(date),      # RequestDate
        credentialScope,
        hexhash(canonicalRequest),
      ].join("\n")

      # Task 3: Calculate the AWS Signature Version 4
      # http://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html

      kDate     = hmac('AWS4' + @secretAccessKey, date.strftime('%Y%m%d'))
      kRegion   = hmac(kDate, 'us-east-1')
      kService  = hmac(kRegion, SERVICE_NAME)
      kSigning  = hmac(kService, 'aws4_request')
      signature = hexhmac(kSigning, stringToSign)

      'AWS4-HMAC-SHA256 Credential=%s/%s, SignedHeaders=%s, Signature=%s' % [
        @accessKeyId,
        credentialScope,
        signedHeaders,
        signature,
      ]
    end

    def iso8601(utc)
      utc.strftime('%Y%m%dT%H%M%SZ')
    end

    def hexhash(data)
      OpenSSL::Digest::SHA256.new.hexdigest(data)
    end

    def hmac(key, data)
      OpenSSL::HMAC.digest(OpenSSL::Digest::SHA256.new, key, data)
    end

    def hexhmac(key, data)
      OpenSSL::HMAC.hexdigest(OpenSSL::Digest::SHA256.new, key, data)
    end

    #def escape(str)
    #  CGI.escape(str.to_s).gsub('+', '%20')
    #end
  end
end

client = Kinesis::Client.new(ENV['AWS_ACCESS_KEY_ID'], ENV['AWS_SECRET_ACCESS_KEY'])
p client.query('ListStreams', {})