Add an optional proxy for signing AWS Elasticsearch requests.

config/default.yml

@@ -199,6 +199,9 @@ elasticsearch:
       breaker:
         fielddata:
           limit: 60%
+  aws_signing_proxy:
+    host: 127.0.0.1
+    port: 14017
 analytics:
   adapter: elasticsearch
   timezone: UTC

src/api-umbrella/cli/read_config.lua

@@ -356,9 +356,9 @@ local function set_computed_config()
     },
     ["_service_general_db_enabled?"] = array_includes(config["services"], "general_db"),
     ["_service_log_db_enabled?"] = array_includes(config["services"], "log_db"),
+    ["_service_elasticsearch_aws_signing_proxy_enabled?"] = array_includes(config["services"], "elasticsearch_aws_signing_proxy"),
     ["_service_router_enabled?"] = array_includes(config["services"], "router"),
     ["_service_web_enabled?"] = array_includes(config["services"], "web"),
-    ["_service_nginx_reloader_enabled?"] = (array_includes(config["services"], "router") and config["nginx"]["_reloader_frequency"]),
     router = {
       trusted_proxies = trusted_proxies,
     },

src/api-umbrella/cli/setup.lua

@@ -1,3 +1,4 @@
+local deep_merge_overwrite_arrays = require "api-umbrella.utils.deep_merge_overwrite_arrays"
 local dir = require "pl.dir"
 local file = require "pl.file"
 local invert_table = require "api-umbrella.utils.invert_table"
@@ -214,6 +215,9 @@ local function activate_services()
   if config["_service_log_db_enabled?"] then
     active_services["elasticsearch"] = 1
   end
+  if config["_service_elasticsearch_aws_signing_proxy_enabled?"] then
+    active_services["elasticsearch-aws-signing-proxy"] = 1
+  end
   if config["_service_router_enabled?"] then
     active_services["geoip-auto-updater"] = 1
     active_services["mora"] = 1

src/api-umbrella/elasticsearch-aws-signing-proxy/init.lua

@@ -0,0 +1,2 @@
+inspect = require "inspect"
+config = require "api-umbrella.proxy.models.file_config"

src/api-umbrella/elasticsearch-aws-signing-proxy/proxy.lua

@@ -0,0 +1,46 @@
+local username = config["elasticsearch"]["aws_signing_proxy"]["username"]
+if not username then
+  ngx.say("elasticsearch.aws_signing_proxy.username must be configured in /etc/api-umbrella/api-umbrella.yml")
+  return ngx.exit(ngx.HTTP_INTERNAL_SERVER_ERROR)
+end
+
+local password = config["elasticsearch"]["aws_signing_proxy"]["password"]
+if not password then
+  ngx.say("elasticsearch.aws_signing_proxy.password must be configured in /etc/api-umbrella/api-umbrella.yml")
+  return ngx.exit(ngx.HTTP_INTERNAL_SERVER_ERROR)
+end
+
+local aws_region = config["elasticsearch"]["aws_signing_proxy"]["aws_region"]
+if not aws_region then
+  ngx.say("elasticsearch.aws_signing_proxy.aws_region must be configured in /etc/api-umbrella/api-umbrella.yml")
+  return ngx.exit(ngx.HTTP_INTERNAL_SERVER_ERROR)
+end
+
+local aws_access_key_id = config["elasticsearch"]["aws_signing_proxy"]["aws_access_key_id"]
+if not aws_access_key_id then
+  ngx.say("elasticsearch.aws_signing_proxy.aws_access_key_id must be configured in /etc/api-umbrella/api-umbrella.yml")
+  return ngx.exit(ngx.HTTP_INTERNAL_SERVER_ERROR)
+end
+
+local aws_secret_access_key = config["elasticsearch"]["aws_signing_proxy"]["aws_secret_access_key"]
+if not aws_secret_access_key then
+  ngx.say("elasticsearch.aws_signing_proxy.aws_access_key_id must be configured in /etc/api-umbrella/api-umbrella.yml")
+  return ngx.exit(ngx.HTTP_INTERNAL_SERVER_ERROR)
+end
+
+local remote_username = ngx.var.remote_user
+local remote_password = ngx.var.remote_passwd
+if not ngx.var.remote_user or not remote_password then
+  ngx.header["WWW-Authenticate"] = 'Basic realm="Restricted"'
+  return ngx.exit(ngx.HTTP_UNAUTHORIZED)
+end
+
+if remote_username ~= username or remote_password ~= password then
+  return ngx.exit(ngx.HTTP_FORBIDDEN)
+end
+
+local host = config["elasticsearch"]["aws_signing_proxy"]["aws_host"]
+ngx.req.set_header("Host", host)
+
+local signing = require "api-umbrella.utils.aws_signing_v4"
+signing.sign_request(aws_region, aws_access_key_id, aws_secret_access_key)

src/api-umbrella/utils/aws_signing_v4.lua

@@ -0,0 +1,179 @@
+local nettle_hmac = require "resty.nettle.hmac"
+local resty_sha256 = require "resty.sha256"
+local to_hex = require("resty.string").to_hex
+
+local escape_uri = ngx.escape_uri
+local gsub = ngx.re.gsub
+
+local AWS_SERVICE = "es"
+local UNSIGNED_HEADERS = {
+  authorization = 1,
+  expect = 1,
+}
+
+local _M = {}
+
+local function hmac(secret_key, value)
+  assert(secret_key)
+  assert(value)
+
+  local hmac_sha256 = nettle_hmac.sha256.new(secret_key)
+  hmac_sha256:update(value)
+  local binary = hmac_sha256:digest()
+
+  return binary
+end
+
+local function sha256_hexdigest(value)
+  local sha256 = resty_sha256:new()
+  sha256:update(value or "")
+  return to_hex(sha256:final())
+end
+
+local function canonical_header_name(name)
+  return string.lower(name)
+end
+
+local function canonical_header_value(value)
+  return gsub(value, [[\s+]], " ", "jo")
+end
+
+local function escape_uri_component(value)
+  if(value == true) then
+    return ""
+  else
+    return escape_uri(value or "")
+  end
+end
+
+local function get_headers()
+  local headers = {}
+
+  local raw_headers = ngx.req.get_headers()
+  for name, value in pairs(raw_headers) do
+    if type(value) == "table" then
+      for multi_name, multi_value in pairs(value) do
+        table.insert(headers, {
+          name = canonical_header_name(multi_name),
+          value = canonical_header_value(multi_value),
+        })
+      end
+    else
+      table.insert(headers, {
+        name = canonical_header_name(name),
+        value = canonical_header_value(value),
+      })
+    end
+  end
+
+  return headers
+end
+
+local function get_canonical_headers(headers)
+  local canonical = {}
+  for _, header in ipairs(headers) do
+    if not UNSIGNED_HEADERS[header.name] then
+      table.insert(canonical, header.name .. ":" .. header.value)
+    end
+  end
+
+  table.sort(canonical)
+  return table.concat(canonical, "\n")
+end
+
+local function get_signed_headers(headers)
+  local signed = {}
+  for _, header in ipairs(headers) do
+    if not UNSIGNED_HEADERS[header.name] then
+      table.insert(signed, header.name)
+    end
+  end
+
+  table.sort(signed)
+  return table.concat(signed, ";")
+end
+
+local function get_canonical_query_string()
+  local canonical = {}
+  local args = ngx.req.get_uri_args()
+  for name, value in pairs(args) do
+    if type(value) == "table" then
+      for multi_name, multi_value in pairs(value) do
+        table.insert(canonical, escape_uri_component(multi_name) .. "=" .. escape_uri_component(multi_value))
+      end
+    else
+      table.insert(canonical, escape_uri_component(name) .. "=" .. escape_uri_component(value))
+    end
+  end
+
+  table.sort(canonical)
+  return table.concat(canonical, "&")
+end
+
+local function get_canonical_request(headers, signed_headers, content_sha256)
+  return table.concat({
+    ngx.var.request_method,
+    gsub(escape_uri(ngx.var.uri), [[%2F]], "/", "ijo"),
+    get_canonical_query_string(),
+    get_canonical_headers(headers) .. "\n",
+    signed_headers,
+    content_sha256,
+  }, "\n")
+end
+
+local function get_credential_scope(aws_region, date)
+  return table.concat({
+    date,
+    aws_region,
+    AWS_SERVICE,
+    "aws4_request",
+  }, "/")
+end
+
+local function get_string_to_sign(datetime, credential_scope, canonical_request)
+  return table.concat({
+    "AWS4-HMAC-SHA256",
+    datetime,
+    credential_scope,
+    sha256_hexdigest(canonical_request),
+  }, "\n")
+end
+
+local function get_signature(aws_region, aws_secret_access_key, date, string_to_sign)
+  local k_date = hmac("AWS4" .. aws_secret_access_key, date)
+  local k_region = hmac(k_date, aws_region)
+  local k_service = hmac(k_region, AWS_SERVICE)
+  local k_credentials = hmac(k_service, "aws4_request")
+  return to_hex(hmac(k_credentials, string_to_sign))
+end
+
+local function get_authorization(aws_access_key_id, credential_scope, signed_headers, signature)
+  return table.concat({
+    "AWS4-HMAC-SHA256 Credential=" .. aws_access_key_id .. "/" .. credential_scope,
+    "SignedHeaders=" .. signed_headers,
+    "Signature=" .. signature,
+  }, ", ")
+end
+
+function _M.sign_request(aws_region, aws_access_key_id, aws_secret_access_key)
+  local datetime = os.date("!%Y%m%dT%H%M%SZ", ngx.now())
+  local date = string.sub(datetime, 1, 8)
+  ngx.req.set_header("X-Amz-Date", os.date("!%Y%m%dT%H%M%SZ", ngx.now()))
+
+  ngx.req.read_body()
+  local body = ngx.req.get_body_data()
+  local content_sha256 = sha256_hexdigest(body)
+  ngx.req.set_header("X-Amz-Content-Sha256", content_sha256)
+
+  local headers = get_headers()
+  local signed_headers = get_signed_headers(headers)
+  local credential_scope = get_credential_scope(aws_region, date)
+
+  local canonical_request = get_canonical_request(headers, signed_headers, content_sha256)
+  local string_to_sign = get_string_to_sign(datetime, credential_scope, canonical_request)
+  local signature = get_signature(aws_region, aws_secret_access_key, date, string_to_sign)
+  local authorization = get_authorization(aws_access_key_id, credential_scope, signed_headers, signature)
+  ngx.req.set_header("Authorization", authorization)
+end
+
+return _M

templates/etc/nginx/elasticsearch-aws-signing-proxy.conf.mustache

@@ -0,0 +1,69 @@
+worker_processes 1;
+error_log stderr notice;
+daemon off;
+pid {{run_dir}}/elasticsearch-aws-signing-proxy.pid;
+
+{{#user}}
+user {{user}} {{group}};
+{{/user}}
+
+events {
+  worker_connections {{nginx.worker_connections}};
+}
+
+env API_UMBRELLA_SRC_ROOT;
+env API_UMBRELLA_RUNTIME_CONFIG;
+
+pcre_jit on;
+
+http {
+  access_log {{log_dir}}/elasticsearch-aws-signing-proxy/{{nginx.access_log_filename}} combined;
+
+  client_body_temp_path {{tmp_dir}}/elasticsearch-aws-signing-proxy-client_body_temp;
+  proxy_temp_path {{tmp_dir}}/elasticsearch-aws-signing-proxy-proxy_temp;
+  fastcgi_temp_path {{tmp_dir}}/elasticsearch-aws-signing-proxy-fastcgi_temp;
+  uwsgi_temp_path {{tmp_dir}}/elasticsearch-aws-signing-proxy-uwsgi_temp;
+  scgi_temp_path {{tmp_dir}}/elasticsearch-aws-signing-proxy-scgi_temp;
+  server_tokens off;
+
+  # FIXME: Detect path/make configurable.
+  lua_ssl_trusted_certificate /etc/ssl/certs/ca-certificates.crt;
+  lua_ssl_verify_depth 5;
+
+  lua_package_path '{{_src_root_dir}}/src/api-umbrella/elasticsearch-aws-signing-proxy/?.lua;{{_package_path}}';
+  lua_package_cpath '{{_package_cpath}}';
+  lua_check_client_abort on;
+  if_modified_since off;
+
+  lua_shared_dict locks {{nginx.shared_dicts.locks.size}};
+
+  {{#dns_resolver._nameservers_nginx}}
+    # FIXME: Make ipv6 configurable.
+    resolver {{dns_resolver._nameservers_nginx}} ipv6=off;
+    resolver_timeout 12s;
+  {{/dns_resolver._nameservers_nginx}}
+
+  include ./mime.conf;
+  include ./realip.conf;
+
+  client_max_body_size 10m;
+  client_body_buffer_size 10m;
+
+  init_by_lua_file '{{_src_root_dir}}/src/api-umbrella/elasticsearch-aws-signing-proxy/init.lua';
+
+  server {
+    listen {{elasticsearch.aws_signing_proxy.host}}:{{elasticsearch.aws_signing_proxy.port}};
+
+    {{#_development_env?}}
+      lua_code_cache off;
+    {{/_development_env?}}
+
+    location / {
+      access_by_lua_file '{{_src_root_dir}}/src/api-umbrella/elasticsearch-aws-signing-proxy/proxy.lua';
+
+      proxy_buffering off;
+      set $backend_upstream "https://{{elasticsearch.aws_signing_proxy.aws_host}}:443";
+      proxy_pass $backend_upstream;
+    }
+  }
+}

templates/etc/perp/elasticsearch-aws-signing-proxy/rc.env.mustache

@@ -0,0 +1 @@
+API_UMBRELLA_RUNTIME_CONFIG={{_api_umbrella_config_runtime_file}}

templates/etc/perp/elasticsearch-aws-signing-proxy/rc.log

@@ -0,0 +1,2 @@
+#!/usr/bin/env bash
+exec ../rc.log "$@"

templates/etc/perp/elasticsearch-aws-signing-proxy/rc.main.mustache

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+
+# Redirect stderr to stdout
+exec 2>&1
+
+if [ "${1}" = "start" ]; then
+  echo "starting ${2}..."
+
+  run_args=("-e" "rc.env" "-c" "{{_src_root_dir}}")
+  exec runtool "${run_args[@]}" nginx -p "{{_src_root_dir}}/" -c "{{etc_dir}}/nginx/elasticsearch-aws-signing-proxy.conf"
+fi
+
+exit 0

test/support/models/log_item.rb

@@ -1,5 +1,3 @@
-#require "elasticsearch/persistence/model"
-
 class LogItem
   include ActiveAttr::Model
 
@@ -71,7 +69,7 @@ def self.clean_indices!
       opts[:search_type] = "scan"
     end
     result = self.client.search(opts)
-    while true
+    loop do
       hits = result["hits"]["hits"]
       break if hits.empty?
       hits.each do |hit|
@@ -102,8 +100,8 @@ def serializable_hash
 
   def save
     index_time = self.request_at
-    if(index_time.kind_of?(Fixnum))
-      index_time = Time.at(index_time / 1000.0)
+    if(index_time.kind_of?(Integer))
+      index_time = Time.at(index_time / 1000.0).utc
     end
 
     index_name = "api-umbrella-logs-write-#{index_time.utc.strftime("%Y-%m")}"