Merge remote-tracking branch 'origin/master' into stream-timestamp-offsetstream-timestamp-offset

8 files changed, 1727 insertions, 0 deletions

diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws.erl b/deps/rabbitmq_aws/src/rabbitmq_aws.erl
new file mode 100644
index 0000000000..4a152f3b21
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws.erl
@@ -0,0 +1,472 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @doc rabbitmq_aws client library
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws).
+
+-behavior(gen_server).
+
+%% API exports
+-export([get/2, get/3,
+         post/4,
+         refresh_credentials/0,
+         request/5, request/6, request/7,
+         set_credentials/2,
+         has_credentials/0,
+         set_region/1]).
+
+%% gen-server exports
+-export([start_link/0,
+         init/1,
+         terminate/2,
+         code_change/3,
+         handle_call/3,
+         handle_cast/2,
+         handle_info/2]).
+
+%% Export all for unit tests
+-ifdef(TEST).
+-compile(export_all).
+-endif.
+
+-include("rabbitmq_aws.hrl").
+
+%%====================================================================
+%% exported wrapper functions
+%%====================================================================
+
+-spec get(Service :: string(),
+          Path :: path()) -> result().
+%% @doc Perform a HTTP GET request to the AWS API for the specified service. The
+%%      response will automatically be decoded if it is either in JSON or XML
+%%      format.
+%% @end
+get(Service, Path) ->
+  get(Service, Path, []).
+
+
+-spec get(Service :: string(),
+          Path :: path(),
+          Headers :: headers()) -> result().
+%% @doc Perform a HTTP GET request to the AWS API for the specified service. The
+%%      response will automatically be decoded if it is either in JSON or XML
+%%      format.
+%% @end
+get(Service, Path, Headers) ->
+  request(Service, get, Path, "", Headers).
+
+
+-spec post(Service :: string(),
+           Path :: path(),
+           Body :: body(),
+           Headers :: headers()) -> result().
+%% @doc Perform a HTTP Post request to the AWS API for the specified service. The
+%%      response will automatically be decoded if it is either in JSON or XML
+%%      format.
+%% @end
+post(Service, Path, Body, Headers) ->
+  request(Service, post, Path, Body, Headers).
+
+
+-spec refresh_credentials() -> ok | error.
+%% @doc Manually refresh the credentials from the environment, filesystem or EC2
+%%      Instance metadata service.
+%% @end
+refresh_credentials() ->
+  gen_server:call(rabbitmq_aws, refresh_credentials).
+
+
+-spec request(Service :: string(),
+              Method :: method(),
+              Path :: path(),
+              Body :: body(),
+              Headers :: headers()) -> result().
+%% @doc Perform a HTTP request to the AWS API for the specified service. The
+%%      response will automatically be decoded if it is either in JSON or XML
+%%      format.
+%% @end
+request(Service, Method, Path, Body, Headers) ->
+  gen_server:call(rabbitmq_aws, {request, Service, Method, Headers, Path, Body, [], undefined}).
+
+
+-spec request(Service :: string(),
+              Method :: method(),
+              Path :: path(),
+              Body :: body(),
+              Headers :: headers(),
+              HTTPOptions :: http_options()) -> result().
+%% @doc Perform a HTTP request to the AWS API for the specified service. The
+%%      response will automatically be decoded if it is either in JSON or XML
+%%      format.
+%% @end
+request(Service, Method, Path, Body, Headers, HTTPOptions) ->
+  gen_server:call(rabbitmq_aws, {request, Service, Method, Headers, Path, Body, HTTPOptions, undefined}).
+
+
+-spec request(Service :: string(),
+              Method :: method(),
+              Path :: path(),
+              Body :: body(),
+              Headers :: headers(),
+              HTTPOptions :: http_options(),
+              Endpoint :: host()) -> result().
+%% @doc Perform a HTTP request to the AWS API for the specified service, overriding
+%%      the endpoint URL to use when invoking the API. This is useful for local testing
+%%      of services such as DynamoDB. The response will automatically be decoded
+%%      if it is either in JSON or XML format.
+%% @end
+request(Service, Method, Path, Body, Headers, HTTPOptions, Endpoint) ->
+  gen_server:call(rabbitmq_aws, {request, Service, Method, Headers, Path, Body, HTTPOptions, Endpoint}).
+
+
+-spec set_credentials(access_key(), secret_access_key()) -> ok.
+%% @doc Manually set the access credentials for requests. This should
+%%      be used in cases where the client application wants to control
+%%      the credentials instead of automatically discovering them from
+%%      configuration or the AWS Instance Metadata service.
+%% @end
+set_credentials(AccessKey, SecretAccessKey) ->
+  gen_server:call(rabbitmq_aws, {set_credentials, AccessKey, SecretAccessKey}).
+
+
+-spec set_region(Region :: string()) -> ok.
+%% @doc Manually set the AWS region to perform API requests to.
+%% @end
+set_region(Region) ->
+  gen_server:call(rabbitmq_aws, {set_region, Region}).
+
+
+%%====================================================================
+%% gen_server functions
+%%====================================================================
+
+start_link() ->
+  gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+-spec init(list()) -> {ok, state()}.
+init([]) ->
+  {ok, #state{}}.
+
+
+terminate(_, _) ->
+  ok.
+
+
+code_change(_, _, State) ->
+  {ok, State}.
+
+
+handle_call(Msg, _From, #state{region = undefined}) ->
+    %% Delay initialisation until a RabbitMQ plugin require the AWS backend
+    {ok, Region} = rabbitmq_aws_config:region(),
+    {_, State} = load_credentials(#state{region = Region}),
+    handle_msg(Msg, State);
+handle_call(Msg, _From, State) ->
+    handle_msg(Msg, State).
+
+
+handle_cast(_Request, State) ->
+  {noreply, State}.
+
+
+handle_info(_Info, State) ->
+  {noreply, State}.
+
+%%====================================================================
+%% Internal functions
+%%====================================================================
+handle_msg({request, Service, Method, Headers, Path, Body, Options, Host}, State) ->
+    {Response, NewState} = perform_request(State, Service, Method, Headers, Path, Body, Options, Host),
+    {reply, Response, NewState};
+
+handle_msg(get_state, State) ->
+    {reply, {ok, State}, State};
+
+handle_msg(refresh_credentials, State) ->
+    {Reply, NewState} = load_credentials(State),
+    {reply, Reply, NewState};
+
+handle_msg({set_credentials, AccessKey, SecretAccessKey}, State) ->
+    {reply, ok, State#state{access_key = AccessKey,
+                            secret_access_key = SecretAccessKey,
+                            security_token = undefined,
+                            expiration = undefined,
+                            error = undefined}};
+
+handle_msg({set_region, Region}, State) ->
+    {reply, ok, State#state{region = Region}};
+
+handle_msg(has_credentials, State) ->
+    {reply, has_credentials(State), State};
+
+handle_msg(_Request, State) ->
+    {noreply, State}.
+
+-spec endpoint(State :: state(), Host :: string(),
+               Service :: string(), Path :: string()) -> string().
+%% @doc Return the endpoint URL, either by constructing it with the service
+%%      information passed in or by using the passed in Host value.
+%% @ednd
+endpoint(#state{region = Region}, undefined, Service, Path) ->
+  lists:flatten(["https://", endpoint_host(Region, Service), Path]);
+endpoint(_, Host, _, Path) ->
+  lists:flatten(["https://", Host, Path]).
+
+
+-spec endpoint_host(Region :: region(), Service :: string()) -> host().
+%% @doc Construct the endpoint hostname for the request based upon the service
+%%      and region.
+%% @end
+endpoint_host(Region, Service) ->
+  lists:flatten(string:join([Service, Region, endpoint_tld(Region)], ".")).
+
+
+-spec endpoint_tld(Region :: region()) -> host().
+%% @doc Construct the endpoint hostname TLD for the request based upon the region.
+%%      See https://docs.aws.amazon.com/general/latest/gr/rande.html#ec2_region for details.
+%% @end
+endpoint_tld("cn-north-1") ->
+    "amazonaws.com.cn";
+endpoint_tld("cn-northwest-1") ->
+    "amazonaws.com.cn";
+endpoint_tld(_Other) ->
+    "amazonaws.com".
+
+-spec format_response(Response :: httpc_result()) -> result().
+%% @doc Format the httpc response result, returning the request result data
+%% structure. The response body will attempt to be decoded by invoking the
+%% maybe_decode_body/2 method.
+%% @end
+format_response({ok, {{_Version, 200, _Message}, Headers, Body}}) ->
+  {ok, {Headers, maybe_decode_body(get_content_type(Headers), Body)}};
+format_response({ok, {{_Version, StatusCode, Message}, Headers, Body}}) when StatusCode >= 400 ->
+  {error, Message, {Headers, maybe_decode_body(get_content_type(Headers), Body)}};
+format_response({error, Reason}) ->
+  {error, Reason, undefined}.
+
+-spec get_content_type(Headers :: headers()) -> {Type :: string(), Subtype :: string()}.
+%% @doc Fetch the content type from the headers and return it as a tuple of
+%%      {Type, Subtype}.
+%% @end
+get_content_type(Headers) ->
+  Value = case proplists:get_value("content-type", Headers, undefined) of
+    undefined ->
+      proplists:get_value("Content-Type", Headers, "text/xml");
+    Other -> Other
+  end,
+  parse_content_type(Value).
+
+-spec has_credentials() -> true | false.
+has_credentials() ->
+  gen_server:call(rabbitmq_aws, has_credentials).
+
+-spec has_credentials(state()) -> true | false.
+%% @doc check to see if there are credentials made available in the current state
+%%      returning false if not or if they have expired.
+%% @end
+has_credentials(#state{error = Error}) when Error /= undefined -> false;
+has_credentials(#state{access_key = Key}) when Key /= undefined -> true;
+has_credentials(_) -> false.
+
+
+-spec expired_credentials(Expiration :: calendar:datetime()) -> true | false.
+%% @doc Indicates if the date that is passed in has expired.
+%% end
+expired_credentials(undefined) -> false;
+expired_credentials(Expiration) ->
+  Now = calendar:datetime_to_gregorian_seconds(local_time()),
+  Expires = calendar:datetime_to_gregorian_seconds(Expiration),
+  Now >= Expires.
+
+
+-spec load_credentials(State :: state()) -> {ok, state()} | {error, state()}.
+%% @doc Load the credentials using the following order of configuration precedence:
+%%        - Environment variables
+%%        - Credentials file
+%%        - EC2 Instance Metadata Service
+%% @end
+load_credentials(#state{region = Region}) ->
+  case rabbitmq_aws_config:credentials() of
+    {ok, AccessKey, SecretAccessKey, Expiration, SecurityToken} ->
+      {ok, #state{region = Region,
+                  error = undefined,
+                  access_key = AccessKey,
+                  secret_access_key = SecretAccessKey,
+                  expiration = Expiration,
+                  security_token = SecurityToken}};
+    {error, Reason} ->
+      error_logger:error_msg("Could not load AWS credentials from environment variables, AWS_CONFIG_FILE, AWS_SHARED_CREDENTIALS_FILE or EC2 metadata endpoint: ~p. Will depend on config settings to be set.~n.", [Reason]),
+      {error, #state{region = Region,
+                     error = Reason,
+                     access_key = undefined,
+                     secret_access_key = undefined,
+                     expiration = undefined,
+                     security_token = undefined}}
+  end.
+
+
+-spec local_time() -> calendar:datetime().
+%% @doc Return the current local time.
+%% @end
+local_time() ->
+  [Value] = calendar:local_time_to_universal_time_dst(calendar:local_time()),
+  Value.
+
+
+-spec maybe_decode_body(ContentType :: {nonempty_string(), nonempty_string()}, Body :: body()) -> list() | body().
+%% @doc Attempt to decode the response body based upon the mime type that is
+%%      presented.
+%% @end.
+maybe_decode_body({"application", "x-amz-json-1.0"}, Body) ->
+  rabbitmq_aws_json:decode(Body);
+maybe_decode_body({"application", "json"}, Body) ->
+  rabbitmq_aws_json:decode(Body);
+maybe_decode_body({_, "xml"}, Body) ->
+  rabbitmq_aws_xml:parse(Body);
+maybe_decode_body(_ContentType, Body) ->
+  Body.
+
+
+-spec parse_content_type(ContentType :: string()) -> {Type :: string(), Subtype :: string()}.
+%% @doc parse a content type string returning a tuple of type/subtype
+%% @end
+parse_content_type(ContentType) ->
+  Parts = string:tokens(ContentType, ";"),
+  [Type, Subtype] = string:tokens(lists:nth(1, Parts), "/"),
+  {Type, Subtype}.
+
+
+-spec perform_request(State :: state(), Service :: string(), Method :: method(),
+                      Headers :: headers(), Path :: path(), Body :: body(),
+                      Options :: http_options(), Host :: string() | undefined)
+    -> {Result :: result(), NewState :: state()}.
+%% @doc Make the API request and return the formatted response.
+%% @end
+perform_request(State, Service, Method, Headers, Path, Body, Options, Host) ->
+  perform_request_has_creds(has_credentials(State), State, Service, Method,
+                            Headers, Path, Body, Options, Host).
+
+
+-spec perform_request_has_creds(true | false, State :: state(),
+                                Service :: string(), Method :: method(),
+                                Headers :: headers(), Path :: path(), Body :: body(),
+                                Options :: http_options(), Host :: string() | undefined)
+    -> {Result :: result(), NewState :: state()}.
+%% @doc Invoked after checking to see if there are credentials. If there are,
+%%      validate they have not or will not expire, performing the request if not,
+%%      otherwise return an error result.
+%% @end
+perform_request_has_creds(true, State, Service, Method, Headers, Path, Body, Options, Host) ->
+  perform_request_creds_expired(expired_credentials(State#state.expiration), State,
+                                Service, Method, Headers, Path, Body, Options, Host);
+perform_request_has_creds(false, State, _, _, _, _, _, _, _) ->
+  perform_request_creds_error(State).
+
+
+-spec perform_request_creds_expired(true | false, State :: state(),
+                                    Service :: string(), Method :: method(),
+                                    Headers :: headers(), Path :: path(), Body :: body(),
+                                    Options :: http_options(), Host :: string() | undefined)
+  -> {Result :: result(), NewState :: state()}.
+%% @doc Invoked after checking to see if the current credentials have expired.
+%%      If they haven't, perform the request, otherwise try and refresh the
+%%      credentials before performing the request.
+%% @end
+perform_request_creds_expired(false, State, Service, Method, Headers, Path, Body, Options, Host) ->
+  perform_request_with_creds(State, Service, Method, Headers, Path, Body, Options, Host);
+perform_request_creds_expired(true, State, Service, Method, Headers, Path, Body, Options, Host) ->
+  perform_request_creds_refreshed(load_credentials(State), Service, Method, Headers, Path, Body, Options, Host).
+
+
+-spec perform_request_creds_refreshed({ok, State :: state()} | {error, State :: state()},
+                                      Service :: string(), Method :: method(),
+                                      Headers :: headers(), Path :: path(), Body :: body(),
+                                      Options :: http_options(), Host :: string() | undefined)
+    -> {Result :: result(), NewState :: state()}.
+%% @doc If it's been determined that there are credentials but they have expired,
+%%      check to see if the credentials could be loaded and either make the request
+%%      or return an error.
+%% @end
+perform_request_creds_refreshed({ok, State}, Service, Method, Headers, Path, Body, Options, Host) ->
+  perform_request_with_creds(State, Service, Method, Headers, Path, Body, Options, Host);
+perform_request_creds_refreshed({error, State}, _, _, _, _, _, _, _) ->
+  perform_request_creds_error(State).
+
+
+-spec perform_request_with_creds(State :: state(), Service :: string(), Method :: method(),
+                                 Headers :: headers(), Path :: path(), Body :: body(),
+                                 Options :: http_options(), Host :: string() | undefined)
+    -> {Result :: result(), NewState :: state()}.
+%% @doc Once it is validated that there are credentials to try and that they have not
+%%      expired, perform the request and return the response.
+%% @end
+perform_request_with_creds(State, Service, Method, Headers, Path, Body, Options, Host) ->
+  URI = endpoint(State, Host, Service, Path),
+  SignedHeaders = sign_headers(State, Service, Method, URI, Headers, Body),
+  ContentType = proplists:get_value("content-type", SignedHeaders, undefined),
+  perform_request_with_creds(State, Method, URI, SignedHeaders, ContentType, Body, Options).
+
+
+-spec perform_request_with_creds(State :: state(), Method :: method(), URI :: string(),
+                                 Headers :: headers(), ContentType :: string() | undefined,
+                                 Body :: body(), Options :: http_options())
+    -> {Result :: result(), NewState :: state()}.
+%% @doc Once it is validated that there are credentials to try and that they have not
+%%      expired, perform the request and return the response.
+%% @end
+perform_request_with_creds(State, Method, URI, Headers, undefined, "", Options0) ->
+  Options1 = ensure_timeout(Options0),
+  Response = httpc:request(Method, {URI, Headers}, Options1, []),
+  {format_response(Response), State};
+perform_request_with_creds(State, Method, URI, Headers, ContentType, Body, Options0) ->
+  Options1 = ensure_timeout(Options0),
+  Response = httpc:request(Method, {URI, Headers, ContentType, Body}, Options1, []),
+  {format_response(Response), State}.
+
+
+-spec perform_request_creds_error(State :: state()) ->
+  {result_error(), NewState :: state()}.
+%% @doc Return the error response when there are not any credentials to use with
+%%      the request.
+%% @end
+perform_request_creds_error(State) ->
+  {{error, {credentials, State#state.error}}, State}.
+
+
+%% @doc Ensure that the timeout option is set and greater than 0 and less
+%%      than about 1/2 of the default gen_server:call timeout. This gives
+%%      enough time for a long connect and request phase to succeed.
+%% @end
+-spec ensure_timeout(Options :: http_options()) -> http_options().
+ensure_timeout(Options) ->
+    case proplists:get_value(timeout, Options) of
+        undefined ->
+            Options ++ [{timeout, ?DEFAULT_HTTP_TIMEOUT}];
+        Value when is_integer(Value) andalso Value >= 0 andalso Value =< ?DEFAULT_HTTP_TIMEOUT ->
+            Options;
+        _ ->
+            Options1 = proplists:delete(timeout, Options),
+            Options1 ++ [{timeout, ?DEFAULT_HTTP_TIMEOUT}]
+    end.
+
+
+-spec sign_headers(State :: state(), Service :: string(), Method :: method(),
+                   URI :: string(), Headers :: headers(), Body :: body()) -> headers().
+%% @doc Build the signed headers for the API request.
+%% @end
+sign_headers(#state{access_key = AccessKey,
+                    secret_access_key = SecretKey,
+                    security_token = SecurityToken,
+                    region = Region}, Service, Method, URI, Headers, Body) ->
+  rabbitmq_aws_sign:headers(#request{access_key = AccessKey,
+                                  secret_access_key = SecretKey,
+                                  security_token = SecurityToken,
+                                  region = Region,
+                                  service = Service,
+                                  method = Method,
+                                  uri = URI,
+                                  headers = Headers,
+                                  body = Body}).
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_app.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_app.erl
new file mode 100644
index 0000000000..b01196ec30
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_app.erl
@@ -0,0 +1,22 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @doc rabbitmq_aws application startup
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_app).
+
+-behaviour(application).
+
+%% Application callbacks
+-export([start/2, stop/1]).
+
+%% ===================================================================
+%% Application callbacks
+%% ===================================================================
+
+start(_StartType, _StartArgs) ->
+  rabbitmq_aws_sup:start_link().
+
+stop(_State) ->
+  ok.
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_config.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_config.erl
new file mode 100644
index 0000000000..09b7606799
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_config.erl
@@ -0,0 +1,694 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @copyright 2016-2020 VMware, Inc. or its affiliates.
+%% @private
+%% @doc rabbitmq_aws configuration functionality
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_config).
+
+%% API
+-export([credentials/0,
+         credentials/1,
+         value/2,
+         values/1,
+         instance_metadata_url/1,
+         instance_credentials_url/1,
+         instance_availability_zone_url/0,
+         instance_role_url/0,
+         region/0,
+         region/1]).
+
+%% Export all for unit tests
+-ifdef(TEST).
+-compile(export_all).
+-endif.
+
+-include("rabbitmq_aws.hrl").
+
+-spec credentials() -> security_credentials().
+%% @doc Return the credentials from environment variables, configuration or the
+%%      EC2 local instance metadata server, if available.
+%%
+%%      If the ``AWS_ACCESS_KEY_ID`` and ``AWS_SECRET_ACCESS_KEY`` environment
+%%      variables are set, those values will be returned. If they are not, the
+%%      local configuration file or shared credentials file will be consulted.
+%%      If either exists and can be checked, they will attempt to return the
+%%      authentication credential values for the ``default`` profile if the
+%%      ``AWS_DEFAULT_PROFILE`` environment is not set.
+%%
+%%      When checking for the configuration file, it will attempt to read the
+%%      file from ``~/.aws/config`` if the ``AWS_CONFIG_FILE`` environment
+%%      variable is not set. If the file is found, and both the access key and
+%%      secret access key are set for the profile, they will be returned. If not
+%%      it will attempt to consult the shared credentials file.
+%%
+%%      When checking for the shared credentials file, it will attempt to read
+%%      read from ``~/.aws/credentials`` if the ``AWS_SHARED_CREDENTIALS_FILE``
+%%      environment variable is not set. If the file is found and the both the
+%%      access key and the secret access key are set for the profile, they will
+%%      be returned.
+%%
+%%      If credentials are returned at any point up through this stage, they
+%%      will be returned as ``{ok, AccessKey, SecretKey, undefined}``,
+%%      indicating the credentials are locally configured, and are not
+%%      temporary.
+%%
+%%      If no credentials could be resolved up until this point, there will be
+%%      an attempt to contact a local EC2 instance metadata service for
+%%      credentials.
+%%
+%%      When the EC2 instance metadata server is checked for but does not exist,
+%%      the operation will timeout in ``?DEFAULT_HTTP_TIMEOUT``ms.
+%%
+%%      When the EC2 instance metadata server exists, but data is not returned
+%%      quickly, the operation will timeout in ``?DEFAULT_HTTP_TIMEOUT``ms.
+%%
+%%      If the service does exist, it will attempt to use the
+%%      ``/meta-data/iam/security-credentials`` endpoint to request expiring
+%%      request credentials to use. If they are found, a tuple of
+%%      ``{ok, AccessKey, SecretAccessKey, SecurityToken}`` will be returned
+%%      indicating the credentials are temporary and require the use of the
+%%      ``X-Amz-Security-Token`` header should be used.
+%%
+%%      Finally, if no credentials are found by this point, an error tuple
+%%      will be returned.
+%% @end
+credentials() ->
+  credentials(profile()).
+
+-spec credentials(string()) -> security_credentials().
+%% @doc Return the credentials from environment variables, configuration or the
+%%      EC2 local instance metadata server, if available.
+%%
+%%      If the ``AWS_ACCESS_KEY_ID`` and ``AWS_SECRET_ACCESS_KEY`` environment
+%%      variables are set, those values will be returned. If they are not, the
+%%      local configuration file or shared credentials file will be consulted.
+%%
+%%      When checking for the configuration file, it will attempt to read the
+%%      file from ``~/.aws/config`` if the ``AWS_CONFIG_FILE`` environment
+%%      variable is not set. If the file is found, and both the access key and
+%%      secret access key are set for the profile, they will be returned. If not
+%%      it will attempt to consult the shared credentials file.
+%%
+%%      When checking for the shared credentials file, it will attempt to read
+%%      read from ``~/.aws/credentials`` if the ``AWS_SHARED_CREDENTIALS_FILE``
+%%      environment variable is not set. If the file is found and the both the
+%%      access key and the secret access key are set for the profile, they will
+%%      be returned.
+%%
+%%      If credentials are returned at any point up through this stage, they
+%%      will be returned as ``{ok, AccessKey, SecretKey, undefined}``,
+%%      indicating the credentials are locally configured, and are not
+%%      temporary.
+%%
+%%      If no credentials could be resolved up until this point, there will be
+%%      an attempt to contact a local EC2 instance metadata service for
+%%      credentials.
+%%
+%%      When the EC2 instance metadata server is checked for but does not exist,
+%%      the operation will timeout in ``?DEFAULT_HTTP_TIMEOUT``ms.
+%%
+%%      When the EC2 instance metadata server exists, but data is not returned
+%%      quickly, the operation will timeout in ``?DEFAULT_HTTP_TIMEOUT``ms.
+%%
+%%      If the service does exist, it will attempt to use the
+%%      ``/meta-data/iam/security-credentials`` endpoint to request expiring
+%%      request credentials to use. If they are found, a tuple of
+%%      ``{ok, AccessKey, SecretAccessKey, SecurityToken}`` will be returned
+%%      indicating the credentials are temporary and require the use of the
+%%      ``X-Amz-Security-Token`` header should be used.
+%%
+%%      Finally, if no credentials are found by this point, an error tuple
+%%      will be returned.
+%% @end
+credentials(Profile) ->
+  lookup_credentials(Profile,
+                     os:getenv("AWS_ACCESS_KEY_ID"),
+                     os:getenv("AWS_SECRET_ACCESS_KEY")).
+
+
+-spec region() -> {ok, string()}.
+%% @doc Return the region as configured by ``AWS_DEFAULT_REGION`` environment
+%%      variable or as configured in the configuration file using the default
+%%      profile or configured ``AWS_DEFAULT_PROFILE`` environment variable.
+%%
+%%      If the environment variable is not set and a configuration
+%%      file is not found, it will try and return the region from the EC2
+%%      local instance metadata server.
+%% @end
+region() ->
+  region(profile()).
+
+
+-spec region(Region :: string()) -> {ok, region()}.
+%% @doc Return the region as configured by ``AWS_DEFAULT_REGION`` environment
+%%      variable or as configured in the configuration file using the specified
+%%      profile.
+%%
+%%      If the environment variable is not set and a configuration
+%%      file is not found, it will try and return the region from the EC2
+%%      local instance metadata server.
+%% @end
+region(Profile) ->
+  case lookup_region(Profile, os:getenv("AWS_DEFAULT_REGION")) of
+    {ok, Region} -> {ok, Region};
+    _ -> {ok, ?DEFAULT_REGION}
+  end.
+
+
+-spec value(Profile :: string(), Key :: atom())
+  -> Value :: any() | {error, Reason :: atom()}.
+%% @doc Return the configuration data for the specified profile or an error
+%%      if the profile is not found.
+%% @end
+value(Profile, Key) ->
+  get_value(Key, values(Profile)).
+
+
+-spec values(Profile :: string())
+  -> Settings :: list()
+  | {error, Reason :: atom()}.
+%% @doc Return the configuration data for the specified profile or an error
+%%      if the profile is not found.
+%% @end
+values(Profile) ->
+  case config_file_data() of
+    {error, Reason} ->
+      {error, Reason};
+    Settings ->
+      Prefixed = lists:flatten(["profile ", Profile]),
+      proplists:get_value(Profile, Settings,
+                          proplists:get_value(Prefixed,
+                                              Settings, {error, undefined}))
+  end.
+
+
+%% -----------------------------------------------------------------------------
+%% Private / Internal Methods
+%% -----------------------------------------------------------------------------
+
+
+-spec config_file() -> string().
+%% @doc Return the configuration file to test using either the value of the
+%%      AWS_CONFIG_FILE or the default location where the file is expected to
+%%      exist.
+%% @end
+config_file() ->
+  config_file(os:getenv("AWS_CONFIG_FILE")).
+
+
+-spec config_file(Path :: false | string()) -> string().
+%% @doc Return the configuration file to test using either the value of the
+%%      AWS_CONFIG_FILE or the default location where the file is expected to
+%%      exist.
+%% @end
+config_file(false) ->
+  filename:join([home_path(), ".aws", "config"]);
+config_file(EnvVar) ->
+  EnvVar.
+
+
+-spec config_file_data() -> list() | {error, Reason :: atom()}.
+%% @doc Return the values from a configuration file as a proplist by section
+%% @end
+config_file_data() ->
+  ini_file_data(config_file()).
+
+
+-spec credentials_file() -> string().
+%% @doc Return the shared credentials file to test using either the value of the
+%%      AWS_SHARED_CREDENTIALS_FILE or the default location where the file
+%%      is expected to exist.
+%% @end
+credentials_file() ->
+  credentials_file(os:getenv("AWS_SHARED_CREDENTIALS_FILE")).
+
+
+-spec credentials_file(Path :: false | string()) -> string().
+%% @doc Return the shared credentials file to test using either the value of the
+%%      AWS_SHARED_CREDENTIALS_FILE or the default location where the file
+%%      is expected to exist.
+%% @end
+credentials_file(false) ->
+  filename:join([home_path(), ".aws", "credentials"]);
+credentials_file(EnvVar) ->
+  EnvVar.
+
+-spec credentials_file_data() -> list() | {error, Reason :: atom()}.
+%% @doc Return the values from a configuration file as a proplist by section
+%% @end
+credentials_file_data() ->
+  ini_file_data(credentials_file()).
+
+
+-spec get_value(Key :: atom(), Settings :: list()) -> any();
+               (Key :: atom(), {error, Reason :: atom()}) -> {error, Reason :: atom()}.
+%% @doc Get the value for a key from a settings proplist.
+%% @end
+get_value(Key, Settings) when is_list(Settings) ->
+  proplists:get_value(Key, Settings, {error, undefined});
+get_value(_, {error, Reason}) -> {error, Reason}.
+
+
+-spec home_path() -> string().
+%% @doc Return the path to the current user's home directory, checking for the
+%%      HOME environment variable before returning the current working
+%%      directory if it's not set.
+%% @end
+home_path() ->
+  home_path(os:getenv("HOME")).
+
+
+-spec home_path(Value :: string() | false) -> string().
+%% @doc Return the path to the current user's home directory, checking for the
+%%      HOME environment variable before returning the current working
+%%      directory if it's not set.
+%% @end
+home_path(false) -> filename:absname(".");
+home_path(Value) -> Value.
+
+
+-spec ini_file_data(Path :: string())
+  -> {ok, list()} | {error, atom()}.
+%% @doc Return the parsed ini file for the specified path.
+%% @end
+ini_file_data(Path) ->
+  ini_file_data(Path, filelib:is_file(Path)).
+
+
+-spec ini_file_data(Path :: string(), FileExists :: true | false)
+  -> {ok, list()} | {error, atom()}.
+%% @doc Return the parsed ini file for the specified path.
+%% @end
+ini_file_data(Path, true) ->
+  case read_file(Path) of
+    {ok, Lines}     -> ini_parse_lines(Lines, none, none, []);
+    {error, Reason} -> {error, Reason}
+  end;
+ini_file_data(_, false) -> {error, enoent}.
+
+
+-spec ini_format_key(any()) -> atom() | {error, type}.
+%% @doc Converts a ini file key to an atom, stripping any leading whitespace
+%% @end
+ini_format_key(Key) ->
+  case io_lib:printable_list(Key) of
+    true -> list_to_atom(string:strip(Key));
+    false -> {error, type}
+  end.
+
+
+-spec ini_parse_line(Section :: list(),
+                     Key :: atom(),
+                     Line :: binary())
+  -> {Section :: list(), Key :: string() | none}.
+%% @doc Parse the AWS configuration INI file, returning a proplist
+%% @end
+ini_parse_line(Section, Parent, <<" ", Line/binary>>) ->
+  Child = proplists:get_value(Parent, Section, []),
+  {ok, NewChild} = ini_parse_line_parts(Child, ini_split_line(Line)),
+  {lists:keystore(Parent, 1, Section, {Parent, NewChild}), Parent};
+ini_parse_line(Section, _, Line) ->
+  case ini_parse_line_parts(Section, ini_split_line(Line)) of
+    {ok, NewSection} -> {NewSection, none};
+    {new_parent, Parent} -> {Section, Parent}
+  end.
+
+
+-spec ini_parse_line_parts(Section :: list(),
+                           Parts :: list())
+  -> {ok, list()} | {new_parent, atom()}.
+%% @doc Parse the AWS configuration INI file, returning a proplist
+%% @end
+ini_parse_line_parts(Section, []) -> {ok, Section};
+ini_parse_line_parts(Section, [RawKey, Value]) ->
+  Key = ini_format_key(RawKey),
+  {ok, lists:keystore(Key, 1, Section, {Key, maybe_convert_number(Value)})};
+ini_parse_line_parts(_, [RawKey]) ->
+  {new_parent, ini_format_key(RawKey)}.
+
+
+-spec ini_parse_lines(Lines::[binary()],
+                      SectionName :: string() | atom(),
+                      Parent :: atom(),
+                      Accumulator :: list())
+  -> list().
+%% @doc Parse the AWS configuration INI file
+%% @end
+ini_parse_lines([], _, _, Settings) -> Settings;
+ini_parse_lines([H|T], SectionName, Parent, Settings) ->
+  {ok, NewSectionName} = ini_parse_section_name(SectionName, H),
+  {ok, NewParent, NewSettings} = ini_parse_section(H, NewSectionName,
+                                                   Parent, Settings),
+  ini_parse_lines(T, NewSectionName, NewParent, NewSettings).
+
+
+-spec ini_parse_section(Line :: binary(),
+                        SectionName :: string(),
+                        Parent :: atom(),
+                        Section :: list())
+  -> {ok, NewParent :: atom(), Section :: list()}.
+%% @doc Parse a line from the ini file, returning it as part of the appropriate
+%%      section.
+%% @end
+ini_parse_section(Line, SectionName, Parent, Settings) ->
+  Section = proplists:get_value(SectionName, Settings, []),
+  {NewSection, NewParent} = ini_parse_line(Section, Parent, Line),
+  {ok, NewParent, lists:keystore(SectionName, 1, Settings,
+                                 {SectionName, NewSection})}.
+
+
+-spec ini_parse_section_name(CurrentSection :: string() | atom(),
+                             Line :: binary())
+  -> {ok, SectionName :: string()}.
+%% @doc Attempts to parse a section name from the current line, returning either
+%%      the new parsed section name, or the current section name.
+%% @end
+ini_parse_section_name(CurrentSection, Line) ->
+  Value = binary_to_list(Line),
+  case re:run(Value, "\\[([\\w\\s+\\-_]+)\\]", [{capture, all, list}]) of
+    {match, [_, SectionName]} -> {ok, SectionName};
+    nomatch -> {ok, CurrentSection}
+  end.
+
+
+-spec ini_split_line(binary()) -> list().
+%% @doc Split a key value pair delimited by ``=`` to a list of strings.
+%% @end
+ini_split_line(Line) ->
+  string:tokens(string:strip(binary_to_list(Line)), "=").
+
+
+-spec instance_availability_zone_url() -> string().
+%% @doc Return the URL for querying the availability zone from the Instance
+%%      Metadata service
+%% @end
+instance_availability_zone_url() ->
+  instance_metadata_url(string:join([?INSTANCE_METADATA_BASE, ?INSTANCE_AZ], "/")).
+
+
+-spec instance_credentials_url(string()) -> string().
+%% @doc Return the URL for querying temporary credentials from the Instance
+%%      Metadata service for the specified role
+%% @end
+instance_credentials_url(Role) ->
+  instance_metadata_url(string:join([?INSTANCE_METADATA_BASE, ?INSTANCE_CREDENTIALS, Role], "/")).
+
+
+-spec instance_metadata_url(string()) -> string().
+%% @doc Build the Instance Metadata service URL for the specified path
+%% @end
+instance_metadata_url(Path) ->
+  rabbitmq_aws_urilib:build(#uri{scheme = "http",
+                                 authority = {undefined, ?INSTANCE_HOST, undefined},
+                                 path = Path, query = []}).
+
+
+-spec instance_role_url() -> string().
+%% @doc Return the URL for querying the role associated with the current
+%%      instance from the Instance Metadata service
+%% @end
+instance_role_url() ->
+  instance_metadata_url(string:join([?INSTANCE_METADATA_BASE, ?INSTANCE_CREDENTIALS], "/")).
+
+
+-spec lookup_credentials(Profile :: string(),
+                         AccessKey :: string() | false,
+                         SecretKey :: string() | false)
+    -> security_credentials().
+%% @doc Return the access key and secret access key if they are set in
+%%      environment variables, otherwise lookup the credentials from the config
+%%      file for the specified profile.
+%% @end
+lookup_credentials(Profile, false, _) ->
+  lookup_credentials_from_config(Profile,
+                                 value(Profile, aws_access_key_id),
+                                 value(Profile, aws_secret_access_key));
+lookup_credentials(Profile, _, false) ->
+  lookup_credentials_from_config(Profile,
+                                 value(Profile, aws_access_key_id),
+                                 value(Profile, aws_secret_access_key));
+lookup_credentials(_, AccessKey, SecretKey) ->
+  {ok, AccessKey, SecretKey, undefined, undefined}.
+
+
+-spec lookup_credentials_from_config(Profile :: string(),
+                                     access_key() | {error, Reason :: atom()},
+                                     secret_access_key()| {error, Reason :: atom()})
+    -> security_credentials().
+%% @doc Return the access key and secret access key if they are set in
+%%      for the specified profile in the config file, if it exists. If it does
+%%      not exist or the profile is not set or the values are not set in the
+%%      profile, look up the values in the shared credentials file
+%% @end
+lookup_credentials_from_config(Profile, {error,_}, _) ->
+  lookup_credentials_from_file(Profile, credentials_file_data());
+lookup_credentials_from_config(_, AccessKey, SecretKey) ->
+  {ok, AccessKey, SecretKey, undefined, undefined}.
+
+
+-spec lookup_credentials_from_file(Profile :: string(),
+                                   Credentials :: list())
+    -> security_credentials().
+%% @doc Check to see if the shared credentials file exists and if it does,
+%%      invoke ``lookup_credentials_from_shared_creds_section/2`` to attempt to
+%%      get the credentials values out of it. If the file does not exist,
+%%      attempt to lookup the values from the EC2 instance metadata service.
+%% @end
+lookup_credentials_from_file(_, {error,_}) ->
+  lookup_credentials_from_instance_metadata();
+lookup_credentials_from_file(Profile, Credentials) ->
+  Section = proplists:get_value(Profile, Credentials),
+  lookup_credentials_from_section(Section).
+
+
+-spec lookup_credentials_from_section(Credentials :: list() | undefined)
+    -> security_credentials().
+%% @doc Return the access key and secret access key if they are set in
+%%      for the specified profile from the shared credentials file. If the
+%%      profile is not set or the values are not set in the profile, attempt to
+%%      lookup the values from the EC2 instance metadata service.
+%% @end
+lookup_credentials_from_section(undefined) ->
+  lookup_credentials_from_instance_metadata();
+lookup_credentials_from_section(Credentials) ->
+  AccessKey = proplists:get_value(aws_access_key_id, Credentials, undefined),
+  SecretKey = proplists:get_value(aws_secret_access_key, Credentials, undefined),
+  lookup_credentials_from_proplist(AccessKey, SecretKey).
+
+
+-spec lookup_credentials_from_proplist(AccessKey :: access_key(),
+                                       SecretAccessKey :: secret_access_key())
+    -> security_credentials().
+%% @doc Process the contents of the Credentials proplists checking if the
+%%      access key and secret access key are both set.
+%% @end
+lookup_credentials_from_proplist(undefined, _) ->
+  lookup_credentials_from_instance_metadata();
+lookup_credentials_from_proplist(_, undefined) ->
+  lookup_credentials_from_instance_metadata();
+lookup_credentials_from_proplist(AccessKey, SecretKey) ->
+  {ok, AccessKey, SecretKey, undefined, undefined}.
+
+
+-spec lookup_credentials_from_instance_metadata()
+    -> security_credentials().
+%% @spec lookup_credentials_from_instance_metadata() -> Result.
+%% @doc Attempt to lookup the values from the EC2 instance metadata service.
+%% @end
+lookup_credentials_from_instance_metadata() ->
+  Role = maybe_get_role_from_instance_metadata(),
+  maybe_get_credentials_from_instance_metadata(Role).
+
+
+-spec lookup_region(Profile :: string(),
+                    Region :: false | string())
+  -> {ok, string()} | {error, undefined}.
+%% @doc If Region is false, lookup the region from the config or the EC2
+%%      instance metadata service.
+%% @end
+lookup_region(Profile, false) ->
+  lookup_region_from_config(values(Profile));
+lookup_region(_, Region) -> {ok, Region}.
+
+
+-spec lookup_region_from_config(Settings :: list() | {error, enoent})
+  -> {ok, string()} | {error, undefined}.
+%% @doc Return the region from the local configuration file. If local config
+%%      settings are not found, try to lookup the region from the EC2 instance
+%%      metadata service.
+%% @end
+lookup_region_from_config({error, enoent}) ->
+  maybe_get_region_from_instance_metadata();
+lookup_region_from_config(Settings) ->
+  lookup_region_from_settings(proplists:get_value(region, Settings)).
+
+
+-spec lookup_region_from_settings(any() | undefined)
+  -> {ok, string()} | {error, undefined}.
+%% @doc Decide if the region should be loaded from the Instance Metadata service
+%%      of if it's already set.
+%% @end
+lookup_region_from_settings(undefined) ->
+  maybe_get_region_from_instance_metadata();
+lookup_region_from_settings(Region) ->
+  {ok, Region}.
+
+
+-spec maybe_convert_number(string()) -> integer() | float().
+%% @doc Returns an integer or float from a string if possible, otherwise
+%%      returns the string().
+%% @end
+maybe_convert_number(Value) ->
+  Stripped = string:strip(Value),
+  case string:to_float(Stripped) of
+      {error,no_float} ->
+        try
+            list_to_integer(Stripped)
+        catch
+            error:badarg -> Stripped
+        end;
+      {F,_Rest} -> F
+  end.
+
+
+-spec maybe_get_credentials_from_instance_metadata({ok, Role :: string()} |
+                                                   {error, undefined})
+  ->  security_credentials().
+%% @doc Try to query the EC2 local instance metadata service to get temporary
+%%      authentication credentials.
+%% @end
+maybe_get_credentials_from_instance_metadata({error, undefined}) ->
+  {error, undefined};
+maybe_get_credentials_from_instance_metadata({ok, Role}) ->
+  URL = instance_credentials_url(Role),
+  parse_credentials_response(perform_http_get(URL)).
+
+
+-spec maybe_get_region_from_instance_metadata()
+  -> {ok, Region :: string()} | {error, Reason :: atom()}.
+%% @doc Try to query the EC2 local instance metadata service to get the region
+%% @end
+maybe_get_region_from_instance_metadata() ->
+  URL = instance_availability_zone_url(),
+  parse_az_response(perform_http_get(URL)).
+
+
+%% @doc Try to query the EC2 local instance metadata service to get the role
+%%      assigned to the instance.
+%% @end
+maybe_get_role_from_instance_metadata() ->
+  URL = instance_role_url(),
+  parse_body_response(perform_http_get(URL)).
+
+
+-spec parse_az_response(httpc_result())
+  -> {ok, Region :: string()} | {error, Reason :: atom()}.
+%% @doc Parse the response from the Availability Zone query to the
+%%      Instance Metadata service, returning the Region if successful.
+%% end.
+parse_az_response({error, _}) -> {error, undefined};
+parse_az_response({ok, {{_, 200, _}, _, Body}})
+  -> {ok, region_from_availability_zone(Body)};
+parse_az_response({ok, {{_, _, _}, _, _}}) -> {error, undefined}.
+
+
+-spec parse_body_response(httpc_result())
+ -> {ok, Value :: string()} | {error, Reason :: atom()}.
+%% @doc Parse the return response from the Instance Metadata service where the
+%%      body value is the string to process.
+%% end.
+parse_body_response({error, _}) -> {error, undefined};
+parse_body_response({ok, {{_, 200, _}, _, Body}}) -> {ok, Body};
+parse_body_response({ok, {{_, _, _}, _, _}}) -> {error, undefined}.
+
+
+-spec parse_credentials_response(httpc_result()) -> security_credentials().
+%% @doc Try to query the EC2 local instance metadata service to get the role
+%%      assigned to the instance.
+%% @end
+parse_credentials_response({error, _}) -> {error, undefined};
+parse_credentials_response({ok, {{_, 404, _}, _, _}}) -> {error, undefined};
+parse_credentials_response({ok, {{_, 200, _}, _, Body}}) ->
+  Parsed = rabbitmq_aws_json:decode(Body),
+  {ok,
+   proplists:get_value("AccessKeyId", Parsed),
+   proplists:get_value("SecretAccessKey", Parsed),
+   parse_iso8601_timestamp(proplists:get_value("Expiration", Parsed)),
+   proplists:get_value("Token", Parsed)}.
+
+
+-spec perform_http_get(string()) -> httpc_result().
+%% @doc Wrap httpc:get/4 to simplify Instance Metadata service requests
+%% @end
+perform_http_get(URL) ->
+  httpc:request(get, {URL, []},
+                [{timeout, ?DEFAULT_HTTP_TIMEOUT}], []).
+
+
+-spec parse_iso8601_timestamp(Timestamp :: string() | binary()) -> calendar:datetime().
+%% @doc Parse a ISO8601 timestamp, returning a datetime() value.
+%% @end
+parse_iso8601_timestamp(Timestamp) when is_binary(Timestamp) ->
+  parse_iso8601_timestamp(binary_to_list(Timestamp));
+parse_iso8601_timestamp(Timestamp) ->
+  [Date, Time] = string:tokens(Timestamp, "T"),
+  [Year, Month, Day] = string:tokens(Date, "-"),
+  [Hour, Minute, Second] = string:tokens(Time, ":"),
+  {{list_to_integer(Year), list_to_integer(Month), list_to_integer(Day)},
+   {list_to_integer(Hour), list_to_integer(Minute), list_to_integer(string:left(Second,2))}}.
+
+
+-spec profile() -> string().
+%% @doc Return the value of the AWS_DEFAULT_PROFILE environment variable or the
+%%      "default" profile.
+%% @end
+profile() -> profile(os:getenv("AWS_DEFAULT_PROFILE")).
+
+
+-spec profile(false | string()) -> string().
+%% @doc Process the value passed in to determine if we will return the default
+%%      profile or the value from the environment variable.
+%% @end
+profile(false) -> ?DEFAULT_PROFILE;
+profile(Value) -> Value.
+
+
+-spec read_file(string()) -> list() | {error, Reason :: atom()}.
+%% @doc Read the specified file, returning the contents as a list of strings.
+%% @end
+read_file(Path) ->
+  read_from_file(file:open(Path, [read])).
+
+
+-spec read_from_file({ok, file:fd()} | {error, Reason :: atom()})
+  -> list() | {error, Reason :: atom()}.
+%% @doc Read the specified file, returning the contents as a list of strings.
+%% @end
+read_from_file({ok, Fd}) ->
+  read_file(Fd, []);
+read_from_file({error, Reason}) ->
+  {error, Reason}.
+
+
+-spec read_file(Fd :: file:io_device(), Lines :: list())
+  -> list() | {error, Reason :: atom()}.
+%% @doc Read from the open file, accumulating the lines in a list.
+%% @end
+read_file(Fd, Lines) ->
+  case file:read_line(Fd) of
+    {ok, Value} ->
+      Line = string:strip(Value, right, $\n),
+      read_file(Fd, lists:append(Lines, [list_to_binary(Line)]));
+    eof -> {ok, Lines};
+    {error, Reason} -> {error, Reason}
+  end.
+
+
+-spec region_from_availability_zone(Value :: string()) -> string().
+%% @doc Strip the availability zone suffix from the region.
+%% @end
+region_from_availability_zone(Value) ->
+  string:sub_string(Value, 1, length(Value) - 1).
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_json.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_json.erl
new file mode 100644
index 0000000000..5b4e3b2f45
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_json.erl
@@ -0,0 +1,62 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @doc Wrap a JSON parser to provide easy abstractions across
+%%      implementations and ensure a consistent return interface.
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_json).
+
+-export([decode/1]).
+
+-spec decode(Value :: string() | binary()) -> list().
+%% @doc Decode a JSON string returning a proplist
+%% @end
+decode(Value) when is_list(Value) ->
+  decode(list_to_binary(Value));
+decode(Value) when is_binary(Value) ->
+  % We set an empty list of options because we don't want the default
+  % options set in rabbit_json:cecode/1. And we can't override
+  % 'return_maps' with '{return_maps, false}' because of a bug in jsx's
+  % options handler.
+  % See https://github.com/talentdeficit/jsx/pull/115
+  Decoded0 = rabbit_json:decode(Value, []),
+  Decoded = if
+    is_map(Decoded0)  -> maps:to_list(Decoded0);
+    is_list(Decoded0) -> Decoded0
+  end,
+  convert_binary_values(Decoded, []).
+
+
+-spec convert_binary_values(Value :: list(), Accumulator :: list()) -> list().
+%% @doc Convert the binary key/value pairs returned by rabbit_json to strings.
+%% @end
+convert_binary_values([], Value) ->  Value;
+convert_binary_values([{K, V}|T], Accum) when is_map(V) ->
+  convert_binary_values(
+    T,
+    lists:append(
+      Accum,
+      [{binary_to_list(K), convert_binary_values(maps:to_list(V), [])}]));
+convert_binary_values([{K, V}|T], Accum) when is_list(V) ->
+  convert_binary_values(
+    T,
+    lists:append(
+      Accum,
+      [{binary_to_list(K), convert_binary_values(V, [])}]));
+convert_binary_values([{}|T],Accum) ->
+  convert_binary_values(T, lists:append(Accum, [{}]));
+convert_binary_values([{K, V}|T], Accum) when is_binary(V) ->
+  convert_binary_values(T, lists:append(Accum, [{binary_to_list(K), binary_to_list(V)}]));
+convert_binary_values([{K, V}|T], Accum) ->
+  convert_binary_values(T, lists:append(Accum, [{binary_to_list(K), V}]));
+convert_binary_values([H|T], Accum) when is_map(H) ->
+  convert_binary_values(T, lists:append(Accum, convert_binary_values(maps:to_list(H), [])));
+convert_binary_values([H|T], Accum) when is_binary(H) ->
+  convert_binary_values(T, lists:append(Accum, [binary_to_list(H)]));
+convert_binary_values([H|T], Accum) when is_integer(H) ->
+  convert_binary_values(T, lists:append(Accum, [H]));
+convert_binary_values([H|T], Accum) when is_atom(H) ->
+  convert_binary_values(T, lists:append(Accum, [H]));
+convert_binary_values([H|T], Accum) ->
+  convert_binary_values(T, lists:append(Accum, convert_binary_values(H, []))).
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_sign.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_sign.erl
new file mode 100644
index 0000000000..b238f81cee
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_sign.erl
@@ -0,0 +1,289 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @private
+%% @doc rabbitmq_aws request signing methods
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_sign).
+
+%% API
+-export([headers/1, request_hash/5]).
+
+%% Transitional step until we can require Erlang/OTP 22 and
+%% use crypto:mac/4 instead of crypto:hmac/3.
+-compile(nowarn_deprecated_function).
+
+%% Export all for unit tests
+-ifdef(TEST).
+-compile(export_all).
+-endif.
+
+-ignore_xref([{crypto, hmac, 3}]).
+
+-include("rabbitmq_aws.hrl").
+
+-define(ALGORITHM, "AWS4-HMAC-SHA256").
+-define(ISOFORMAT_BASIC, "~4.10.0b~2.10.0b~2.10.0bT~2.10.0b~2.10.0b~2.10.0bZ").
+
+-spec headers(request()) -> headers().
+%% @doc Create the signed request headers
+%% end
+headers(Request) ->
+  RequestTimestamp = local_time(),
+  PayloadHash = sha256(Request#request.body),
+  URI = rabbitmq_aws_urilib:parse(Request#request.uri),
+  {_, Host, _} = URI#uri.authority,
+  Headers = append_headers(RequestTimestamp,
+                           length(Request#request.body),
+                           PayloadHash,
+                           Host,
+                           Request#request.security_token,
+                           Request#request.headers),
+  RequestHash = request_hash(Request#request.method,
+                             URI#uri.path,
+                             URI#uri.query,
+                             Headers,
+                             Request#request.body),
+  AuthValue = authorization(Request#request.access_key,
+                            Request#request.secret_access_key,
+                            RequestTimestamp,
+                            Request#request.region,
+                            Request#request.service,
+                            Headers,
+                            RequestHash),
+  sort_headers(lists:merge([{"authorization", AuthValue}], Headers)).
+
+
+-spec amz_date(AMZTimestamp :: string()) -> string().
+%% @doc Extract the date from the AMZ timestamp format.
+%% @end
+amz_date(AMZTimestamp) ->
+  [RequestDate, _] = string:tokens(AMZTimestamp, "T"),
+  RequestDate.
+
+
+-spec append_headers(AMZDate :: string(),
+                     ContentLength :: integer(),
+                     PayloadHash :: string(),
+                     Hostname :: host(),
+                     SecurityToken :: security_token(),
+                     Headers :: headers()) -> list().
+%% @doc Append the headers that need to be signed to the headers passed in with
+%%      the request
+%% @end
+append_headers(AMZDate, ContentLength, PayloadHash, Hostname, SecurityToken, Headers) ->
+  Defaults = default_headers(AMZDate, ContentLength, PayloadHash, Hostname, SecurityToken),
+  Headers1 = [{string:to_lower(Key), Value} || {Key, Value} <- Headers],
+  Keys = lists:usort(lists:append([string:to_lower(Key) || {Key, _} <- Defaults],
+                                  [Key || {Key, _} <- Headers1])),
+  sort_headers([{Key, header_value(Key, Headers1, proplists:get_value(Key, Defaults))} || Key <- Keys]).
+
+
+-spec authorization(AccessKey :: access_key(),
+                    SecretAccessKey :: secret_access_key(),
+                    RequestTimestamp :: string(),
+                    Region :: region(),
+                    Service :: string(),
+                    Headers :: headers(),
+                    RequestHash :: string()) -> string().
+%% @doc Return the authorization header value
+%% @end
+authorization(AccessKey, SecretAccessKey, RequestTimestamp, Region, Service, Headers, RequestHash) ->
+  RequestDate = amz_date(RequestTimestamp),
+  Scope = scope(RequestDate, Region, Service),
+  Credentials = ?ALGORITHM ++ " Credential=" ++ AccessKey ++ "/" ++ Scope,
+  SignedHeaders = "SignedHeaders=" ++ signed_headers(Headers),
+  StringToSign = string_to_sign(RequestTimestamp, RequestDate, Region, Service, RequestHash),
+  SigningKey = signing_key(SecretAccessKey, RequestDate, Region, Service),
+  Signature = string:join(["Signature", signature(StringToSign, SigningKey)], "="),
+  string:join([Credentials, SignedHeaders, Signature], ", ").
+
+
+-spec default_headers(RequestTimestamp :: string(),
+                      ContentLength :: integer(),
+                      PayloadHash :: string(),
+                      Hostname :: host(),
+                      SecurityToken :: security_token()) -> headers().
+%% @doc build the base headers that are merged in with the headers for every
+%%      request.
+%% @end
+default_headers(RequestTimestamp, ContentLength, PayloadHash, Hostname, undefined) ->
+  [{"content-length", integer_to_list(ContentLength)},
+   {"date", RequestTimestamp},
+   {"host", Hostname},
+   {"x-amz-content-sha256", PayloadHash}];
+default_headers(RequestTimestamp, ContentLength, PayloadHash, Hostname, SecurityToken) ->
+  [{"content-length", integer_to_list(ContentLength)},
+   {"date", RequestTimestamp},
+   {"host", Hostname},
+   {"x-amz-content-sha256", PayloadHash},
+   {"x-amz-security-token", SecurityToken}].
+
+
+-spec canonical_headers(Headers :: headers()) -> string().
+%% @doc Convert the headers list to a line-feed delimited string in the AWZ
+%%      canonical headers format.
+%% @end
+canonical_headers(Headers) ->
+  canonical_headers(sort_headers(Headers), []).
+
+-spec canonical_headers(Headers :: headers(), CanonicalHeaders :: list()) -> string().
+%% @doc Convert the headers list to a line-feed delimited string in the AWZ
+%%      canonical headers format.
+%% @end
+canonical_headers([], CanonicalHeaders) ->
+  lists:flatten(CanonicalHeaders);
+canonical_headers([{Key, Value}|T], CanonicalHeaders) ->
+  Header = string:join([string:to_lower(Key), Value], ":") ++ "\n",
+  canonical_headers(T, lists:append(CanonicalHeaders, [Header])).
+
+
+-spec credential_scope(RequestDate :: string(),
+                       Region :: region(),
+                       Service :: string()) -> string().
+%% @doc Return the credential scope string used in creating the request string to sign.
+%% @end
+credential_scope(RequestDate, Region, Service) ->
+  lists:flatten(string:join([RequestDate, Region, Service, "aws4_request"], "/")).
+
+
+-spec header_value(Key :: string(),
+                   Headers :: headers(),
+                   Default :: string()) -> string().
+%% @doc Return the the header value or the default value for the header if it
+%%      is not specified.
+%% @end
+header_value(Key, Headers, Default) ->
+  proplists:get_value(Key, Headers, proplists:get_value(string:to_lower(Key), Headers, Default)).
+
+
+-spec hmac_sign(Key :: string(), Message :: string()) -> string().
+%% @doc Return the SHA-256 hash for the specified value.
+%% @end
+hmac_sign(Key, Message) ->
+  SignedValue = crypto:hmac(sha256, Key, Message),
+  binary_to_list(SignedValue).
+
+
+-spec local_time() -> string().
+%% @doc Return the current timestamp in GMT formatted in ISO8601 basic format.
+%% @end
+local_time() ->
+  [LocalTime] = calendar:local_time_to_universal_time_dst(calendar:local_time()),
+  local_time(LocalTime).
+
+
+-spec local_time(calendar:datetime()) -> string().
+%% @doc Return the current timestamp in GMT formatted in ISO8601 basic format.
+%% @end
+local_time({{Y,M,D},{HH,MM,SS}}) ->
+  lists:flatten(io_lib:format(?ISOFORMAT_BASIC, [Y, M, D, HH, MM, SS])).
+
+
+-spec query_string(QueryArgs :: list()) -> string().
+%% @doc Return the sorted query string for the specified arguments.
+%% @end
+query_string(undefined) -> "";
+query_string(QueryArgs) ->
+  rabbitmq_aws_urilib:build_query_string(lists:keysort(1, QueryArgs)).
+
+
+-spec request_hash(Method :: method(),
+                   Path :: path(),
+                   QArgs :: query_args(),
+                   Headers :: headers(),
+                   Payload :: string()) -> string().
+%% @doc Create the request hash value
+%% @end
+request_hash(Method, Path, QArgs, Headers, Payload) ->
+  RawPath = case string:slice(Path, 0, 1) of
+    "/" -> Path;
+    _   -> "/" ++ Path
+  end,
+  EncodedPath = uri_string:recompose(#{path => RawPath}),
+  CanonicalRequest = string:join([string:to_upper(atom_to_list(Method)),
+                                  EncodedPath,
+                                  query_string(QArgs),
+                                  canonical_headers(Headers),
+                                  signed_headers(Headers),
+                                  sha256(Payload)], "\n"),
+  sha256(CanonicalRequest).
+
+
+-spec scope(AMZDate :: string(),
+            Region :: region(),
+            Service :: string()) -> string().
+%% @doc Create the Scope string
+%% @end
+scope(AMZDate, Region, Service) ->
+  string:join([AMZDate, Region, Service, "aws4_request"], "/").
+
+
+-spec sha256(Value :: string()) -> string().
+%% @doc Return the SHA-256 hash for the specified value.
+%% @end
+sha256(Value) ->
+  lists:flatten(io_lib:format("~64.16.0b",
+                              [binary:decode_unsigned(crypto:hash(sha256, Value))])).
+
+
+-spec signed_headers(Headers :: list()) -> string().
+%% @doc Return the signed headers string of delimited header key names
+%% @end
+signed_headers(Headers) ->
+  signed_headers(sort_headers(Headers), []).
+
+
+-spec signed_headers(Headers :: headers(), Values :: list()) -> string().
+%% @doc Return the signed headers string of delimited header key names
+%% @end
+signed_headers([], SignedHeaders) -> string:join(SignedHeaders, ";");
+signed_headers([{Key,_}|T], SignedHeaders) ->
+  signed_headers(T, SignedHeaders ++ [string:to_lower(Key)]).
+
+
+-spec signature(StringToSign :: string(),
+                SigningKey :: string()) -> string().
+%% @doc Create the request signature.
+%% @end
+signature(StringToSign, SigningKey) ->
+  SignedValue = crypto:hmac(sha256, SigningKey, StringToSign),
+  lists:flatten(io_lib:format("~64.16.0b", [binary:decode_unsigned(SignedValue)])).
+
+
+-spec signing_key(SecretKey :: secret_access_key(),
+                  AMZDate :: string(),
+                  Region :: region(),
+                  Service :: string()) -> string().
+%% @doc Create the signing key
+%% @end
+signing_key(SecretKey, AMZDate, Region, Service) ->
+  DateKey = hmac_sign("AWS4" ++ SecretKey, AMZDate),
+  RegionKey = hmac_sign(DateKey, Region),
+  ServiceKey = hmac_sign(RegionKey, Service),
+  hmac_sign(ServiceKey, "aws4_request").
+
+
+-spec string_to_sign(RequestTimestamp :: string(),
+                     RequestDate :: string(),
+                     Region :: region(),
+                     Service :: string(),
+                     RequestHash :: string()) -> string().
+%% @doc Return the string to sign when creating the signed request.
+%% @end
+string_to_sign(RequestTimestamp, RequestDate, Region, Service, RequestHash) ->
+  CredentialScope = credential_scope(RequestDate, Region, Service),
+  lists:flatten(string:join([
+    ?ALGORITHM,
+    RequestTimestamp,
+    CredentialScope,
+    RequestHash
+  ], "\n")).
+
+
+-spec sort_headers(Headers :: headers()) -> headers().
+%% @doc Case-insensitive sorting of the request headers
+%% @end
+sort_headers(Headers) ->
+  lists:sort(fun({A,_}, {B, _}) -> string:to_lower(A) =< string:to_lower(B) end, Headers).
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_sup.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_sup.erl
new file mode 100644
index 0000000000..6327b2029d
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_sup.erl
@@ -0,0 +1,20 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @doc rabbitmq_aws supervisor for the gen_server process
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0,
+         init/1]).
+
+-define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 5, Type, [I]}).
+
+start_link() ->
+  supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+  {ok, {{one_for_one, 5, 10}, [?CHILD(rabbitmq_aws, worker)]}}.
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_urilib.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_urilib.erl
new file mode 100644
index 0000000000..02435270ba
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_urilib.erl
@@ -0,0 +1,122 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016
+%% @doc urilib is a RFC-3986 URI Library for Erlang
+%%      https://github.com/gmr/urilib
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_urilib).
+
+-export([build/1,
+         build_query_string/1,
+         parse/1,
+         parse_userinfo/1,
+         parse_userinfo_result/1
+        ]).
+
+%% Export all for unit tests
+-ifdef(TEST).
+-compile(export_all).
+-endif.
+
+-include("rabbitmq_aws.hrl").
+
+-spec build(#uri{}) -> string().
+%% @doc Build a URI string
+%% @end
+build(URI) ->
+  {UserInfo, Host, Port} = URI#uri.authority,
+  UriMap = #{
+    scheme => to_list(URI#uri.scheme),
+    host => Host
+  },
+  UriMap1 = case UserInfo of
+    undefined         -> UriMap;
+    {User, undefined} -> maps:put(userinfo, User, UriMap);
+    {User, Password}  -> maps:put(userinfo, User ++ ":" ++ Password, UriMap);
+    Value             -> maps:put(userinfo, Value, UriMap)
+  end,
+  UriMap2 = case Port of
+    undefined -> UriMap1;
+    Value1    -> maps:put(port, Value1, UriMap1)
+  end,
+  UriMap3 = case URI#uri.path of
+    undefined -> maps:put(path, "", UriMap2);
+    Value2    ->
+      PrefixedPath = case string:slice(Value2, 0, 1) of
+        "/" -> Value2;
+        _   -> "/" ++ Value2
+      end,
+      maps:put(path, PrefixedPath, UriMap2)
+  end,
+  UriMap4 = case URI#uri.query of
+    undefined -> UriMap3;
+    ""        -> UriMap3;
+    Value3    -> maps:put(query, build_query_string(Value3), UriMap3)
+  end,
+  UriMap5 = case URI#uri.fragment of
+    undefined -> UriMap4;
+    Value4    -> maps:put(fragment, Value4, UriMap4)
+  end,
+  uri_string:recompose(UriMap5).
+
+-spec parse(string()) -> #uri{} | {error, any()}.
+%% @doc Parse a URI string returning a record with the parsed results
+%% @end
+parse(Value) ->
+  UriMap = uri_string:parse(Value),
+  Scheme = maps:get(scheme, UriMap, "https"),
+  Host = maps:get(host, UriMap),
+
+  DefaultPort = case Scheme of
+    "http"  -> 80;
+    "https" -> 443;
+    _       -> undefined
+  end,
+  Port = maps:get(port, UriMap, DefaultPort),
+  UserInfo = parse_userinfo(maps:get(userinfo, UriMap, undefined)),
+  Path = maps:get(path, UriMap),
+  Query = maps:get(query, UriMap, ""),
+  #uri{scheme = Scheme,
+       authority = {parse_userinfo(UserInfo), Host, Port},
+       path = Path,
+       query = uri_string:dissect_query(Query),
+       fragment = maps:get(fragment, UriMap, undefined)
+  }.
+
+
+-spec parse_userinfo(string() | undefined)
+    -> {username() | undefined, password() | undefined} | undefined.
+parse_userinfo(undefined) -> undefined;
+parse_userinfo([]) -> undefined;
+parse_userinfo({User, undefined}) -> {User, undefined};
+parse_userinfo({User, Password})  -> {User, Password};
+parse_userinfo(Value) ->
+  parse_userinfo_result(string:tokens(Value, ":")).
+
+
+-spec parse_userinfo_result(list())
+    -> {username() | undefined, password() | undefined} | undefined.
+parse_userinfo_result([User, Password]) -> {User, Password};
+parse_userinfo_result([User]) -> {User, undefined};
+parse_userinfo_result({User, undefined}) -> {User, undefined};
+parse_userinfo_result([]) -> undefined;
+parse_userinfo_result(User) -> {User, undefined}.
+
+%% @spec build_query(proplist()) -> string()
+%% @doc Build the query parameters string from a proplist
+%% @end
+%%
+
+-spec build_query_string([{any(), any()}]) -> string().
+
+build_query_string(Args) when is_list(Args) ->
+  Normalized = [{to_list(K), to_list(V)} || {K, V} <- Args],
+  uri_string:compose_query(Normalized).
+
+-spec to_list(Val :: integer() | list() | binary() | atom() | map()) -> list().
+to_list(Val) when is_list(Val)    -> Val;
+to_list(Val) when is_map(Val)     -> maps:to_list(Val);
+to_list(Val) when is_atom(Val)    -> atom_to_list(Val);
+to_list(Val) when is_binary(Val)  -> binary_to_list(Val);
+to_list(Val) when is_integer(Val) -> integer_to_list(Val).
diff --git a/deps/rabbitmq_aws/src/rabbitmq_aws_xml.erl b/deps/rabbitmq_aws/src/rabbitmq_aws_xml.erl
new file mode 100644
index 0000000000..fc3be5c642
--- /dev/null
+++ b/deps/rabbitmq_aws/src/rabbitmq_aws_xml.erl
@@ -0,0 +1,46 @@
+%% ====================================================================
+%% @author Gavin M. Roy <gavinmroy@gmail.com>
+%% @copyright 2016, Gavin M. Roy
+%% @doc Simple XML parser for AWS application/xml responses
+%% @end
+%% ====================================================================
+-module(rabbitmq_aws_xml).
+
+-export([parse/1]).
+
+-include_lib("xmerl/include/xmerl.hrl").
+
+-spec parse(Value :: string() | binary()) -> list().
+parse(Value) ->
+  {Element, _} = xmerl_scan:string(Value),
+  parse_node(Element).
+
+
+parse_node(#xmlElement{name=Name, content=Content}) ->
+  Value = parse_content(Content, []),
+  [{atom_to_list(Name), flatten_value(Value, Value)}].
+
+
+flatten_text([], Value) -> Value;
+flatten_text([{K,V}|T], Accum) when is_list(V) ->
+    flatten_text(T, lists:append([{K, V}], Accum));
+flatten_text([H | T], Accum) when is_list(H) ->
+    flatten_text(T, lists:append(T, Accum)).
+
+
+flatten_value([L], _) when is_list(L) -> L;
+flatten_value(L, _) when is_list(L) -> flatten_text(L, []).
+
+
+parse_content([], Value) -> Value;
+parse_content(#xmlElement{} = Element, Accum) ->
+  lists:append(parse_node(Element), Accum);
+parse_content(#xmlText{value=Value}, Accum) ->
+  case string:strip(Value) of
+    "" -> Accum;
+    "\n" -> Accum;
+    Stripped ->
+      lists:append([Stripped], Accum)
+  end;
+parse_content([H|T], Accum) ->
+  parse_content(T, parse_content(H, Accum)).