nginx-lua: upload request-body to AWS-S3

config.ru

require 'grape'
require 'uuid'
require 'aws-sdk'


class Hello < Grape::API
  HTTP_PROXY = ENV['HTTP_PROXY'] || ENV['HTTPS_PROXY']
  AWS_REGION = ENV['AWS_REGION'] || 'ap-northeast-1'

  module Logic
    def self.s3_object(bucket, key)
      @@s3 ||= Aws::S3::Resource.new(region:AWS_REGION, http_proxy:HTTP_PROXY)
      @@s3.bucket(bucket).object(key)
    end
    def self.signed_upload_uri(bucket, key, type, size)
      s3_object(bucket, key).presigned_url(:put, content_type:type, content_length:size)
    end
  end

  before do
    http_headers = request.env.select { |k, v| k.start_with?('HTTP_') }
    @headers = http_headers.inject({}){|a, (k, v)|
      a[k.sub(/^HTTP_/, "").downcase] = v
      a
    }
    @user = request.env['HTTP_X_USER']
    @time = Time.now.iso8601
    @headers.merge!(user:@user) if @user
    @headers.merge!(time:@time)
  end

  format :json
  default_format :json

  get 'auth_adapter' do
    p @headers.merge(GET:'auth_adapter')
    if @headers['authorization']
      endpoint = @headers['x_upload_s3_endpoint']
      bucket = endpoint.split(".", 2)[0]
      uuid = UUID.generate
      status 200
      header 'X-User', @headers['authorization'].split(/:/)[0]
      header 'X-Upload-Content-Id', uuid
      header 'X-Upload-URI', Logic.signed_upload_uri(bucket, uuid,
                                                     @headers['x_upload_content_type'],
                                                     @headers['x_upload_content_length'])
      {state:'OK'}
    else
      status 401
      {state:'authorization failed'}
    end
  end

  get 's3uploader/:id' do
    status 201
    p @headers.merge(GET:'s3upload/:id', id:params.id)
    {state:'uploaded', id:params.id}
  end
end

run Hello

nginx.conf

user  ec2-user;
worker_processes  1;

#error_log  logs/error.log;
#error_log  logs/error.log  notice;
#error_log  logs/error.log  info;

#pid        logs/nginx.pid;


events {
    worker_connections  1024;
}


http {
    include       mime.types;
    default_type  application/octet-stream;

    passenger_root /usr/local/share/ruby/gems/2.0/gems/passenger-5.0.23;
    passenger_ruby /usr/bin/ruby2.0;

    lua_package_path "/opt/nginx/nginx/scripts/?.lua;;";
    init_by_lua_block {
      JSON = require('cjson')
      S3Uploader = require("s3uploader")
    }

    #log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
    #                  '$status $body_bytes_sent "$http_referer" '
    #                  '"$http_user_agent" "$http_x_forwarded_for"';

    #access_log  logs/access.log  main;

    sendfile        on;
    #tcp_nopush     on;

    #keepalive_timeout  0;
    keepalive_timeout  65;

    #gzip  on;

    server {
        listen       127.0.0.1:3000;

        location = /auth_adapter {
          root   html/public;
          passenger_enabled on;
        }

        location /s3uploader/ {
          root   html/public;
          passenger_enabled on;
        }
    }

    server {
        listen       80;
        server_name  localhost;
        resolver     172.31.0.2;

        location = / {
          root   html;
          index  index.html index.htm;
        }
        location /teapot {
          content_by_lua_block {
            ngx.status = 417
            ngx.say("I'm teapot.")
          }
        }

        location = /s3upload {
          set $aws_access_key_id      "....";
          set $aws_secret_access_key  "....";
          set $aws_s3_bucket          "bucket-name";
          set $aws_s3_endpoint        "s3-ap-northeast-1.amazonaws.com";
          limit_except PUT { deny all; }
          access_by_lua_block {
            ngx.req.set_header('X-Upload-S3-Endpoint', ngx.var.aws_s3_bucket.."."..ngx.var.aws_s3_endpoint)
            ngx.req.set_header('X-Upload-Content-Type', ngx.var.content_type)
            ngx.req.set_header('X-Upload-Content-Length', ngx.var.content_length)
            local res = ngx.location.capture('/_px_auth_adapter')
            if not res.status==200 then
              ngx.exit(res.status)
            end
            ngx.req.set_header('X-User', res.header['X-User'])
            ngx.req.set_header('X-Upload-Content-Id', res.header['X-Upload-Content-Id'])
            ngx.req.set_header('X-Upload-URI', res.header['X-Upload-URI'])
          }
          content_by_lua_block {
            local content_id = ngx.req.get_headers()['X-Upload-Content-Id']
            local res;
            if true then
              local uri = ngx.req.get_headers()['X-Upload-URI']
              res = S3Uploader.upload(uri)
            else
              S3Uploader.endpoint   = ngx.var.aws_s3_endpoint
              S3Uploader.access_key = ngx.var.aws_access_key_id
              S3Uploader.secret_key = ngx.var.aws_secret_access_key
              res = S3Uploader.upload(ngx.var.aws_s3_bucket.."/"..content_id)
            end
            if res.status>399 then
              ngx.status = res.status
              ngx.say(res.body)
              ngx.exit(res.status)
            end
            ngx.req.set_method(ngx.HTTP_GET)
            ngx.req.set_header('X-Upload-Status', res.status)
            ngx.req.set_header('Content-Length', '')
            ngx.req.set_header('X-Upload-URI', '')
            ngx.exec('/_px_s3uploader/'..content_id)
          }
        }
        location = /upload {
          content_by_lua_file scripts/upload.lua;
        }

        location ~ ^/_px_(.+)$ {
          internal;
          proxy_pass http://127.0.0.1:3000/$1;
        }

        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }
    }
}

s3uploader.lua

--[[
method1: use AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.
    S3Uploader = require('s3uploader')
    S3Uploader.endpoint = 's3.amazonaws.com'
    res = S3Uploader.upload('bucket-name/object-key')
    #=> res.status, res.content_type, res.content_length, res.body

method2: use signed-upload-uri
    S3Uploader = require('s3uploader')
    res = S3Uploader.upload('https://bucket-name.s3.amazonaws.com/object-key?SIGNATURE')
    #=> res.status, res.content_type, res.content_length, res.body
]]

local _M = {}

_M.chunkBytes = 2048

-- parameters for connect('bucket-name/object-key') method.
_M.endpoint   = os.getenv("AWS_S3_ENDPOINT")
_M.access_key = os.getenv("AWS_ACCESS_KEY_ID")
_M.secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")


local function signature(method, objectkey)
  local date = ngx.http_time(ngx.time())
  local headers = ngx.req.get_headers()
  local s2s = method.."\n"
  s2s = s2s .. (headers["content-md5"] or "") .."\n"
  s2s = s2s .. (headers["content-type"] or "").."\n"
  s2s = s2s .. date.."\n"
  s2s = s2s .. "/"..objectkey
  local auth = "AWS ".._M.access_key..":"..ngx.encode_base64(ngx.hmac_sha1(_M.secret_key, s2s))
  return auth, date
end

local function header_join_with_crlf(header)
  local str = "", k, v
  for k,v in pairs(header) do
    str = str..v.."\r\n"
  end
  return str
end

-- @return client, err
local function connect(endpoint, objectkey, signed_uri)
  local ok, err
  local obj = {}
  obj.clientsock, err = ngx.req.socket()
  if err then
    return nil, "failed to get sock - "..err
  end
  obj.serversock = ngx.socket.tcp()
  ok, err = obj.serversock:connect(endpoint, 443)
  if not ok then
    return nil, "failed to connect: "..endpoint.." - "..err
  end
  local session, err = obj.serversock:sslhandshake()
  if err then
    return nil, "failed to sslhandshake - "..err
  end
  obj.send_header = function(self)
    local headers = ngx.req.get_headers()
    local reqtype = headers["content-type"] or "application/json"
    local reqsize = headers["content-length"] or "0"
    local header = {"PUT /"..objectkey.." HTTP/1.0",
                    "Host: "..endpoint,
                    "Content-Type: "..reqtype,
                    "Content-Length: "..reqsize}
    if not signed_uri then
      local auth , date = signature("PUT", objectkey)
      table.insert(header, "Authorization: "..auth)
      table.insert(header, "Date: "..date)
    end
    local bytes, err = self.serversock:send(header_join_with_crlf(header).."\r\n")
    if err then
      self.err = "failed to send header - "..err
      return nil
    end
    self.reqsize = reqsize
    return true
  end
  obj.response = {}
  obj.response.dump = function(self)
    local s = ""
    s = s.."\r\nstatus: "..self.status
    s = s.."\r\ncontent_type: "..(self.content_type or "(none)")
    s = s.."\r\ncontent_length: "..(self.content_length or "(none)")
    s = s.."\r\nlocation: "..(self.location or "(none)")
    s = s.."\r\nbody: "..(self.body or "(none)")
    s = s.."\r\n"
    print(s)
  end
  obj.send_body = function(self)
    local size = tonumber(self.reqsize)
    while size>0 do
      local rsize = _M.chunkBytes
      if size<rsize then rsize=size end
      local data, err, partial = self.clientsock:receive(rsize)
      if err then
        self.err = "failed to receive request body - "..err
        return nil
      end
      if not chunk then chunk = partial end
      local bytes, err = self.serversock:send(data)
      if err then
        self.err = "failed to send request body - "..err
        return nil
      end
      size = size - rsize
    end
    return true
  end
  obj.recv_header = function(self)
    while true do
      local data, err, partial = self.serversock:receive()
      if err then
        self.err = "failed to receive response - " .. err
        return nil
      end
      if not data then data = partial end
      local m, err = ngx.re.match(data, "HTTP/1.[01] *([0-9]+)", "i")
      if m then self.response.status = tonumber(m[1]) end
      m, err = ngx.re.match(data, "content-type: *(.+)", "i")
      if m then self.response.content_type = m[1] end
      m, err = ngx.re.match(data, "content-length: *([0-9]+)", "i")
      if m then self.response.content_length = tonumber(m[1]) end
      m, err = ngx.re.match(data, "location: *(.+)", "i")
      if m then self.response.location = tonumber(m[1]) end
      if data=="" then break end
    end
    return true
  end
  obj.recv_body = function(self)
    local length = self.response.content_length
    if length==nil then length = 256 end
    if length>0 then
      local data, err, partial = self.serversock:receive(length)
      if err then
        self.err = "failed to receive response - "..err
        return nil
      end
      if not data then data = partial end
      self.response.content_length = #data
      self.response.body = data
    end
    return true
  end
  return obj
end

_M.upload = function(objectkey)
  local m, err = ngx.re.match(objectkey, "^https://([^/]+)/(.+)$")
  local endpoint = _M.endpoint
  local signed = false
  if m and m[1] and m[1] then
    endpoint = m[1]
    objectkey = m[2]
    signed = true
  elseif not endpoint then
    ngx.log(ngx.ERR, 'AWS S3 endpoint is not configured.')
    ngx.exit(500)
  end
  local client, err = connect(endpoint, objectkey, signed)
  if err then
    ngx.log(ngx.ERR, err)
    return ngx.exit(500)
  end
  local ok = client:send_header() and client:send_body() and client:recv_header() and client:recv_body()
  client.serversock:close()
  if not ok then
    ngx.log(ngx.ERR, client.err)
    return ngx.exit(500)
  end
  return client.response
end

return _M