You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2026/09/03 16:39:13 UTC
(couchdb) branch iam-auth-couch_gun created (now c6255bb48)
This is an automated email from the ASF dual-hosted git repository.
nickva pushed a change to branch iam-auth-couch_gun
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at c6255bb48 Use couch_gun for auth ibm
This branch includes the following new commits:
new c6255bb48 Use couch_gun for auth ibm
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(couchdb) 01/01: Use couch_gun for auth ibm
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nickva pushed a commit to branch iam-auth-couch_gun
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit c6255bb484d70a115e90a27b95f7674ff4217173
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Thu Sep 3 12:36:43 2026 -0400
Use couch_gun for auth ibm
Copied couch_gun from https://github.com/apache/couchdb/pull/6098
Don't bother starting a pool as we'd keep the connection open for an hour and
at those time intervals it doesn't matter that much
---
src/couch/src/couch_gun.erl | 439 +++++++++++++++++++++
.../src/couch_replicator_auth_ibm.erl | 186 +++------
2 files changed, 496 insertions(+), 129 deletions(-)
diff --git a/src/couch/src/couch_gun.erl b/src/couch/src/couch_gun.erl
new file mode 100644
index 000000000..e207283ce
--- /dev/null
+++ b/src/couch/src/couch_gun.erl
@@ -0,0 +1,439 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+% Helper module to use gun instead of other http clients we had.
+%
+% Functions:
+%
+% * req/3,4,5: For basic synchronous requests
+%
+% * parse_url/1: Helper parser to turn urls with possible auth bits
+% embedded into a gun uri map
+%
+% * open/3, close/1: Open/close connections
+%
+% * send/5,6: Send requests on opened connections. Can take {Fun, State} for
+% a streaming body.
+%
+% * await/3: Wait for response a send.
+%
+% * headers/1, method/1, basic_auth/2: convert existing calling conventions
+% to gun's format. These are helpers to avoid modifying all the call sites
+% right off the bat. For example, the codebase expects headers to be
+% strings and methods to be atom so we transform them here accordingly.
+%
+% * norm_error/1: shorten/normalize gun's error reason
+
+-module(couch_gun).
+
+-export([
+ req/3,
+ req/4,
+ req/5,
+ parse_url/1,
+ open/3,
+ close/1,
+ send/5,
+ send/6,
+ await/3,
+ headers/1,
+ method/1,
+ basic_auth/2,
+ norm_error/1
+]).
+
+-define(DEFAULT_TIMEOUT, 30000).
+
+req(Method, Url, Headers) ->
+ req(Method, Url, Headers, <<>>, #{}).
+
+req(Method, Url, Headers, Body) ->
+ req(Method, Url, Headers, Body, #{}).
+
+req(Method, Url, Headers, Body, #{} = Opts) when is_atom(Method), is_list(Url) ->
+ case parse_url(Url) of
+ {ok, #{transport := Transport, host := Host, port := Port} = Parsed} ->
+ #{path := Path, userinfo := UserInfo} = Parsed,
+ Timeout = maps:get(timeout, Opts, ?DEFAULT_TIMEOUT),
+ Headers1 = auth_headers(headers(Headers), UserInfo, Opts),
+ OpenOpts = maps:with([tls_opts, tcp_opts, gun_opts], Opts),
+ case open(Host, Port, OpenOpts#{transport => Transport}) of
+ {ok, Pid} ->
+ try
+ case gun:await_up(Pid, Timeout) of
+ {ok, _} ->
+ Ref = send(Pid, Method, Path, Headers1, Body),
+ await(Pid, Ref, Timeout);
+ {error, Reason} ->
+ {error, norm_error(Reason)}
+ end
+ after
+ close(Pid)
+ end;
+ {error, Reason} ->
+ {error, norm_error(Reason)}
+ end;
+ {error, _} = Error ->
+ Error
+ end.
+
+parse_url("http://" ++ Rest) ->
+ parse_auth(tcp, 80, Rest);
+parse_url("https://" ++ Rest) ->
+ parse_auth(tls, 443, Rest);
+parse_url(_) ->
+ {error, invalid_uri}.
+
+parse_auth(Transport, DefaultPort, Rest) ->
+ {Auth, Path} =
+ case lists:splitwith(fun(C) -> C /= $/ andalso C /= $? end, Rest) of
+ {A, ""} -> {A, "/"};
+ {A, "?" ++ _ = Query} -> {A, "/" ++ Query};
+ {A, P} -> {A, P}
+ end,
+ {UserInfo, HostPort} =
+ case string:split(Auth, "@", trailing) of
+ [Creds, HP] -> {Creds, HP};
+ [HP] -> {undefined, HP}
+ end,
+ case parse_host_port(HostPort, DefaultPort) of
+ {error, _} = Error ->
+ Error;
+ {Host, Port} ->
+ {ok, #{
+ transport => Transport,
+ host => Host,
+ port => Port,
+ path => Path,
+ userinfo => UserInfo
+ }}
+ end.
+
+parse_host_port("", _DefaultPort) ->
+ {error, invalid_uri};
+parse_host_port("[" ++ Rest, DefaultPort) ->
+ % ipv6 with brackets [...]
+ case string:split(Rest, "]") of
+ ["", _] ->
+ {error, invalid_uri};
+ [Host, ""] ->
+ {Host, DefaultPort};
+ [Host, ":" ++ PortStr] ->
+ case string:to_integer(PortStr) of
+ {Port, ""} when is_integer(Port) -> {Host, Port};
+ _ -> {error, invalid_uri}
+ end;
+ _ ->
+ {error, invalid_uri}
+ end;
+parse_host_port(HostPort, DefaultPort) ->
+ case string:split(HostPort, ":", trailing) of
+ ["", _] ->
+ {error, invalid_uri};
+ [Host, PortStr] ->
+ case string:to_integer(PortStr) of
+ {Port, ""} when is_integer(Port) -> {Host, Port};
+ _ -> {error, invalid_uri}
+ end;
+ [Host] ->
+ {Host, DefaultPort}
+ end.
+
+% Spawn an http/1.1 gun connection process. Opts is a map that can have
+% these fields:
+%
+% transport - tcp | tls (default is tcp)
+% tls_opts - tls client opts, default is []
+% tcp_opts - gen_tcp options
+% gun_opts - Other gun options (see gun:open/3 docs)
+%
+% Host could be a string or address tuple. We don't wait for the connection to
+% be up a caller may do that with gun:wait_up/2 to get connection errors
+% earlier than during the first send.
+%
+open(Host, Port, #{} = Opts) ->
+ Transport = maps:get(transport, Opts, tcp),
+ OpenOpts0 = #{transport => Transport, protocols => [http], retry => 0},
+ OpenOpts1 =
+ case Opts of
+ #{tcp_opts := TcpOpts} -> OpenOpts0#{tcp_opts => TcpOpts};
+ #{} -> OpenOpts0
+ end,
+ OpenOpts2 =
+ case Transport of
+ tls -> OpenOpts1#{tls_opts => maps:get(tls_opts, Opts, [])};
+ tcp -> OpenOpts1
+ end,
+ OpenOpts = maps:merge(OpenOpts2, maps:get(gun_opts, Opts, #{})),
+ gun:open(host(Host), Port, OpenOpts).
+
+close(Pid) when is_pid(Pid) ->
+ try
+ gun:close(Pid)
+ catch
+ _:_ -> ok
+ end,
+ ok.
+
+% Send a request and get back a stream ref. Body may be a {Fun, State} tuple.
+% Then Fun(State) should return {ok, Data, NewState} and then return eof at the
+% end. Data will be sent chunked unless a content-length header is set.
+send(Pid, Method, Path, Headers, Body) ->
+ send(Pid, Method, Path, Headers, Body, #{}).
+
+send(Pid, Method, Path, Headers, {Fun, State}, ReqOpts) when is_function(Fun, 1) ->
+ SRef = gun:headers(Pid, method(Method), Path, headers(Headers), ReqOpts),
+ ok = send_body(Pid, SRef, Fun, State),
+ SRef;
+send(Pid, Method, Path, Headers, Body, ReqOpts) when is_pid(Pid) ->
+ gun:request(Pid, method(Method), Path, headers(Headers), body(Body), ReqOpts).
+
+body([]) ->
+ <<>>;
+body(Body) ->
+ Body.
+
+send_body(Pid, SRef, Fun, State) ->
+ case Fun(State) of
+ {ok, Data, State1} ->
+ send_body(Pid, SRef, Fun, State1, Data);
+ eof ->
+ ok = gun:data(Pid, SRef, fin, <<>>)
+ end.
+
+send_body(Pid, SRef, Fun, State, Data0) ->
+ case Fun(State) of
+ {ok, Data, State1} ->
+ % Send pending data before sending the next.
+ % We're doing one chunk at a time here
+ ok = gun:data(Pid, SRef, nofin, Data0),
+ send_body(Pid, SRef, Fun, State1, Data);
+ eof ->
+ ok = gun:data(Pid, SRef, fin, Data0)
+ end.
+
+% Wait for a response. First wait for status + headers then body. 1xx info
+% responses are skipped and we don't care about trailers either. If we got a
+% bad connection and didn't find out until calling send and await we'll get the
+% error here.
+await(Pid, SRef, Timeout) when is_pid(Pid) ->
+ MRef = monitor(process, Pid),
+ try await_headers(Pid, SRef, Timeout, MRef) of
+ {response, fin, Code, RespHeaders} ->
+ {ok, Code, RespHeaders, <<>>};
+ {response, nofin, Code, RespHeaders} ->
+ case gun:await_body(Pid, SRef, Timeout, MRef) of
+ {ok, RespBody} -> {ok, Code, RespHeaders, RespBody};
+ {ok, RespBody, _Trailers} -> {ok, Code, RespHeaders, RespBody};
+ {error, Reason} -> {error, norm_error(Reason)}
+ end;
+ {error, Reason} ->
+ {error, norm_error(Reason)}
+ after
+ demonitor(MRef, [flush])
+ end.
+
+await_headers(Pid, SRef, Timeout, MRef) ->
+ case gun:await(Pid, SRef, Timeout, MRef) of
+ {inform, _Status, _Headers} -> await_headers(Pid, SRef, Timeout, MRef);
+ Other -> Other
+ end.
+
+% Transform our request headers into gun's lowercase binary shape. Previous
+% http client accepted special atom headers like {basic_auth, {User, Pass}} and
+% {cookie, Value}, {content_type, Type}, {content_length, Len}. We handle those
+% here to avoid modifying all the call sites.
+
+headers(Headers) ->
+ [header(H) || H <- Headers].
+
+header({basic_auth, {User, Pass}}) ->
+ basic_auth(User, Pass);
+header({cookie, Cookie}) ->
+ {~"cookie", to_bin(Cookie)};
+header({content_type, Value}) ->
+ {~"content-type", to_bin(Value)};
+header({content_length, Value}) ->
+ {~"content-length", to_bin(Value)};
+header({Name, Value}) ->
+ {string:lowercase(to_bin(Name)), to_bin(Value)}.
+
+method(Method) when is_atom(Method) ->
+ string:uppercase(atom_to_binary(Method, utf8)).
+
+basic_auth(User, Pass) ->
+ UserPass = base64:encode(iolist_to_binary([User, $:, Pass])),
+ {~"authorization", <<"Basic ", UserPass/binary>>}.
+
+% If headers already have authorization set use that, otherwise take from the
+% userinfo field from the url
+auth_headers(Headers, UserInfo, Opts) ->
+ case lists:keymember(~"authorization", 1, Headers) of
+ true ->
+ Headers;
+ false ->
+ case {Opts, UserInfo} of
+ {#{basic_auth := {User, Pass}}, _} ->
+ [basic_auth(User, Pass) | Headers];
+ {#{}, undefined} ->
+ Headers;
+ {#{}, UserInfo} ->
+ case string:split(UserInfo, ":") of
+ [User, Pass] -> [basic_auth(User, Pass) | Headers];
+ [User] -> [basic_auth(User, "") | Headers]
+ end
+ end
+ end.
+
+to_bin(V) when is_atom(V) ->
+ atom_to_binary(V, utf8);
+to_bin(V) ->
+ iolist_to_binary(V).
+
+% gun expects IPs as address tuples
+host(Host) when is_list(Host) ->
+ case inet:parse_strict_address(Host) of
+ {ok, Ip} -> Ip;
+ {error, _} -> Host
+ end;
+host(Host) ->
+ Host.
+
+% To simplify error handling return {error, Reason} to make it easier for
+% callers to handle it instead of the multi-level nested error shapes from gun.
+norm_error({stream_error, Reason}) ->
+ norm_error(Reason);
+norm_error({connection_error, Reason}) ->
+ norm_error(Reason);
+norm_error({down, {shutdown, Reason}}) ->
+ norm_error(Reason);
+norm_error({down, Reason}) ->
+ norm_error(Reason);
+norm_error({shutdown, Reason}) ->
+ norm_error(Reason);
+norm_error(Reason) ->
+ Reason.
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+parse_url_test() ->
+ ?assertEqual(
+ {ok, #{transport => tcp, host => "h", port => 80, path => "/", userinfo => undefined}},
+ parse_url("http://h")
+ ),
+ ?assertEqual(
+ {ok, #{
+ transport => tcp, host => "h", port => 5984, path => "/db?a=b", userinfo => undefined
+ }},
+ parse_url("http://h:5984/db?a=b")
+ ),
+ ?assertEqual(
+ {ok, #{transport => tls, host => "h", port => 443, path => "/", userinfo => undefined}},
+ parse_url("https://h")
+ ),
+ ?assertEqual(
+ {ok, #{
+ transport => tcp, host => "127.0.0.1", port => 80, path => "/", userinfo => undefined
+ }},
+ parse_url("http://127.0.0.1")
+ ),
+ ?assertEqual(
+ {ok, #{transport => tcp, host => "::1", port => 5984, path => "/db", userinfo => undefined}},
+ parse_url("http://[::1]:5984/db")
+ ),
+ ?assertEqual(
+ {ok, #{transport => tcp, host => "::1", port => 80, path => "/", userinfo => undefined}},
+ parse_url("http://[::1]")
+ ),
+ ?assertEqual(
+ {ok, #{transport => tcp, host => "h", port => 80, path => "/", userinfo => "u:p"}},
+ parse_url("http://u:p@h")
+ ),
+ ?assertEqual(
+ {ok, #{
+ transport => tcp,
+ host => "h",
+ port => 15984,
+ path => "/_dbs_info?startkey=\"db1\"&endkey=\"db2\"",
+ userinfo => undefined
+ }},
+ parse_url("http://h:15984/_dbs_info?startkey=\"db1\"&endkey=\"db2\"")
+ ),
+ ?assertEqual(
+ {ok, #{transport => tcp, host => "h", port => 80, path => "/?q=1", userinfo => undefined}},
+ parse_url("http://h?q=1")
+ ),
+ ?assertEqual({error, invalid_uri}, parse_url("a potato")),
+ ?assertEqual({error, invalid_uri}, parse_url("ftp://h/")),
+ ?assertEqual({error, invalid_uri}, parse_url("http://")),
+ ?assertEqual({error, invalid_uri}, parse_url("http://:80/")),
+ ?assertEqual({error, invalid_uri}, parse_url("http://h:x/")),
+ ?assertEqual({error, invalid_uri}, parse_url("http://[::1")),
+ ?assertEqual({error, invalid_uri}, parse_url("http://[]:80/")).
+
+host_test() ->
+ ?assertEqual("cdb.example.com", host("cdb.example.com")),
+ ?assertEqual({127, 0, 0, 1}, host("127.0.0.1")),
+ ?assertEqual({0, 0, 0, 0, 0, 0, 0, 1}, host("::1")),
+ ?assertEqual({1, 2, 3, 4}, host({1, 2, 3, 4})).
+
+headers_test() ->
+ ?assertEqual([], headers([])),
+ ?assertEqual(
+ [{~"content-type", ~"application/json"}],
+ headers([{"Content-Type", "application/json"}])
+ ),
+ ?assertEqual(
+ [{~"x-foo", ~"1"}, {~"accept", ~"*/*"}],
+ headers([{'X-Foo', "1"}, {~"Accept", ~"*/*"}])
+ ),
+ ?assertEqual(
+ [basic_auth("u", "p"), {~"cookie", ~"k=v"}],
+ headers([{basic_auth, {"u", "p"}}, {cookie, "k=v"}])
+ ),
+ ?assertEqual(
+ [{~"content-type", ~"text/plain"}, {~"content-length", ~"3"}],
+ headers([{content_type, "text/plain"}, {content_length, "3"}])
+ ).
+
+method_test() ->
+ ?assertEqual(~"GET", method(get)),
+ ?assertEqual(~"COPY", method(copy)),
+ ?assertEqual(~"DELETE", method('Delete')).
+
+basic_auth_test() ->
+ ?assertEqual(
+ {~"authorization", <<"Basic ", (base64:encode(~"u:p"))/binary>>},
+ basic_auth("u", "p")
+ ),
+ ?assertEqual(basic_auth("u", "p"), basic_auth(~"u", ~"p")).
+
+auth_headers_test() ->
+ Auth = basic_auth("u", "p"),
+ Override = [{~"authorization", ~"Bearer dabears"}],
+ ?assertEqual([], auth_headers([], undefined, #{})),
+ ?assertEqual([Auth], auth_headers([], undefined, #{basic_auth => {"u", "p"}})),
+ ?assertEqual([Auth], auth_headers([], "u:p", #{})),
+ ?assertEqual([basic_auth("u", "")], auth_headers([], "u", #{})),
+ ?assertEqual([Auth], auth_headers([], "x:y", #{basic_auth => {"u", "p"}})),
+ ?assertEqual(Override, auth_headers(Override, "x:y", #{basic_auth => {"u", "p"}})).
+
+norm_error_test() ->
+ ?assertEqual(econnrefused, norm_error({down, {shutdown, econnrefused}})),
+ ?assertEqual(closed, norm_error({stream_error, closed})),
+ ?assertEqual(closed, norm_error({connection_error, closed})),
+ ?assertEqual(timeout, norm_error(timeout)),
+ ?assertEqual(normal, norm_error({down, normal})).
+
+-endif.
diff --git a/src/couch_replicator/src/couch_replicator_auth_ibm.erl b/src/couch_replicator/src/couch_replicator_auth_ibm.erl
index 40f66e7c4..5f3a84b91 100644
--- a/src/couch_replicator/src/couch_replicator_auth_ibm.erl
+++ b/src/couch_replicator/src/couch_replicator_auth_ibm.erl
@@ -67,7 +67,7 @@
-record(state, {
mac_key,
- token_uri_map
+ token_url
}).
-record(public_entry, {
@@ -154,17 +154,17 @@ update_last_used(#worker_state{} = WorkerState) ->
%% gen_server callbacks.
init(_) ->
- case token_uri_map() of
- {ok, TokenURIMap} ->
+ case token_url() of
+ {ok, TokenUrl} ->
?PUBLIC = ets:new(?PUBLIC, [protected, {keypos, #public_entry.api_key_mac}, named_table]),
?PRIVATE = ets:new(?PRIVATE, [
private, {keypos, #private_entry.api_key_mac}, named_table
]),
ok = config:listen_for_changes(?MODULE, nil),
- start_gun(#state{
+ {ok, #state{
mac_key = crypto:strong_rand_bytes(32),
- token_uri_map = TokenURIMap
- });
+ token_url = TokenUrl
+ }};
{error, Reason} ->
{error, Reason}
end.
@@ -249,14 +249,6 @@ handle_info({expire_api_key_entry, APIKeyMAC}, State) ->
reply_all(Entry, {error, expired_api_key_entry})
end,
{noreply, State};
-handle_info({'DOWN', AcquireMRef, process, _AcquirePid, {error, Reason}}, #state{} = State) ->
- case match_on_acquire_mref(AcquireMRef) of
- [#private_entry{} = Entry] ->
- reply_and_reset(Entry, {error, Reason});
- [] ->
- ok
- end,
- {noreply, State};
handle_info(
{'DOWN', AcquireMRef, process, _AcquirePid, {ok, StatusCode, _ResponseHeaders, ResponseBody}},
#state{} = State
@@ -327,28 +319,32 @@ handle_info(
ok
end,
{noreply, State};
+handle_info({'DOWN', AcquireMRef, process, _AcquirePid, Reason}, #state{} = State) ->
+ %% acquire process exited with {error, Reason} or died some other way
+ %% we don't care how, we always clean up after it
+ case match_on_acquire_mref(AcquireMRef) of
+ [#private_entry{} = Entry] ->
+ reply_and_reset(Entry, acquire_error(Reason));
+ [] ->
+ ok
+ end,
+ {noreply, State};
handle_info(restart_config_listener, State) ->
ok = config:listen_for_changes(?MODULE, nil),
{noreply, State};
-handle_info(token_url_change, State0) ->
- case token_uri_map() of
- {ok, TokenURIMap} ->
- stop_gun(State0),
- case start_gun(State0#state{token_uri_map = TokenURIMap}) of
- {ok, State1} ->
- {noreply, State1};
- {error, Reason} ->
- {stop, Reason, State0}
- end;
+handle_info(token_url_change, State) ->
+ case token_url() of
+ {ok, TokenUrl} ->
+ %% picked up by the next request, nothing to reconnect
+ {noreply, State#state{token_url = TokenUrl}};
{error, Reason} ->
- {stop, Reason, State0}
+ {stop, Reason, State}
end;
handle_info(Msg, State) ->
couch_log:warning("~p: unexpected info ~p", [?MODULE, Msg]),
{noreply, State}.
-terminate(_Reason, State) ->
- stop_gun(State),
+terminate(_Reason, _State) ->
ets:foldl(
fun(#private_entry{} = Entry, Acc) ->
cancel_timer(Entry#private_entry.refresh_ref),
@@ -394,65 +390,33 @@ extract_api_key(#httpdb{auth_props = AuthProps}) ->
{error, missing_api_key}
end.
-acquire_token(APIKey, #state{} = State) ->
+acquire_token(APIKey, #state{token_url = Url}) ->
Headers = [
- {~"content-type", ~"application/x-www-form-urlencoded"},
- host_header(State)
+ {~"content-type", ~"application/x-www-form-urlencoded"}
],
Body = mochiweb_util:urlencode([
{~"grant_type", ~"urn:ibm:params:oauth:grant-type:apikey"},
{~"response_type", ~"cloud_iam"},
{~"apikey", APIKey}
]),
- #{path := Path} = State#state.token_uri_map,
- {_Pid, Ref} = spawn_monitor(
- post(fix_path(Path), Headers, Body)
- ),
+ Timeout = config:get_integer("ibm", "request_timeout", 30000),
+ {_Pid, Ref} = spawn_monitor(fun() ->
+ exit(
+ try
+ post(Url, Headers, Body, Timeout)
+ catch
+ Tag:Err -> {error, {Tag, Err}}
+ end
+ )
+ end),
Ref.
-post(Path, ReqHeaders, ReqBody) ->
- post(Path, ReqHeaders, ReqBody, 5).
-
-post(Path, ReqHeaders, ReqBody, RemainingTries) ->
- fun() ->
- case gun_pool:post(Path, ReqHeaders, ReqBody) of
- {async, PoolStreamRef} ->
- Timeout = config:get_integer("ibm", "request_timeout", 30000),
- case gun_pool:await(PoolStreamRef, Timeout) of
- {response, fin, StatusCode, _RespHeaders} ->
- exit({error, {unexpected_status_code, StatusCode}});
- {response, nofin, StatusCode, RespHeaders} ->
- case gun_pool:await_body(PoolStreamRef, Timeout) of
- {ok, RespBody} ->
- exit({ok, StatusCode, RespHeaders, RespBody});
- {error, Reason} ->
- exit({error, Reason})
- end
- end;
- {error, no_connection_available, _Reason} when RemainingTries > 0 ->
- timer:sleep(1000),
- post(Path, ReqHeaders, ReqBody, RemainingTries - 1);
- {error, _Type, Reason} ->
- exit({error, Reason})
- end
- end.
-
-host_header(#state{} = State) ->
- #{host := Host} = State#state.token_uri_map,
- Port = maps:get(port, State#state.token_uri_map, undefined),
- case Port of
- undefined ->
- {~"host", Host};
- _ when is_integer(Port) ->
- {~"host", [Host, $:, integer_to_list(Port)]}
- end.
-
-fix_path([]) ->
- ~"/";
-fix_path(Path) when is_binary(Path), byte_size(Path) > 0 ->
- Path;
-fix_path(Path) when is_list(Path), length(Path) > 0 ->
- Path.
+post(Url, ReqHeaders, ReqBody, Timeout) ->
+ Opts = #{
+ timeout => Timeout,
+ tls_opts => [{cacerts, couch_replicator_utils:cacert_get()}]
+ },
+ couch_gun:req(post, Url, ReqHeaders, ReqBody, Opts).
decode_iam_response(ResponseBody) ->
try jiffy:decode(ResponseBody, [return_maps]) of
@@ -501,13 +465,13 @@ api_key_uuid(EncodedToken) when is_binary(EncodedToken) ->
~""
end.
--spec token_uri_map() -> {ok, uri_string:uri_map()} | {error, term()}.
-token_uri_map() ->
- URI = config:get("ibm", "token_url", "https://iam.cloud.ibm.com/identity/token"),
- case uri_string:parse(URI) of
- #{} = URIMap ->
- {ok, URIMap};
- {error, Reason, _Details} ->
+-spec token_url() -> {ok, string()} | {error, term()}.
+token_url() ->
+ Url = config:get("ibm", "token_url", "https://iam.cloud.ibm.com/identity/token"),
+ case couch_gun:parse_url(Url) of
+ {ok, #{}} ->
+ {ok, Url};
+ {error, Reason} ->
{error, Reason}
end.
@@ -517,6 +481,11 @@ token_timeout() ->
match_on_acquire_mref(AcquireMRef) ->
ets:match_object(?PRIVATE, #private_entry{acquire_mref = AcquireMRef, _ = '_'}).
+acquire_error({error, _} = Error) ->
+ Error;
+acquire_error(Reason) ->
+ {error, Reason}.
+
-spec reply_and_reset(Entry :: #private_entry{}, Reply :: term()) -> ok.
reply_and_reset(#private_entry{} = Entry, Reply) ->
ets:insert(?PRIVATE, Entry#private_entry{
@@ -537,35 +506,6 @@ cancel_timer(undefined) ->
cancel_timer(TimerRef) when is_reference(TimerRef) ->
erlang:cancel_timer(TimerRef).
-start_gun(#state{} = State) ->
- #{host := Host, scheme := Scheme} = State#state.token_uri_map,
- Port = port(State#state.token_uri_map),
- ConnOptions = #{
- transport => transport(Scheme), tls_opts => [{cacerts, couch_replicator_utils:cacert_get()}]
- },
- case gun_pool:start_pool(Host, Port, #{size => 5, conn_opts => ConnOptions}) of
- {ok, PoolPid} ->
- gun_pool:await_up(PoolPid),
- {ok, State};
- {error, Reason} ->
- {error, Reason}
- end.
-
-stop_gun(#state{} = State) ->
- #{host := Host, scheme := Scheme} = State#state.token_uri_map,
- Port = port(State#state.token_uri_map),
- gun_pool:stop_pool(Host, Port, #{transport => transport(Scheme)}).
-
-port(#{port := Port}) ->
- Port;
-port(#{scheme := "https"}) ->
- 443.
-
-transport("http") ->
- tcp;
-transport("https") ->
- tls.
-
mac(Key, Data) when is_binary(Key), is_binary(Data) ->
crypto:mac(hmac, sha256, Key, Data).
@@ -710,24 +650,12 @@ handle_response_test() ->
cleanup_test() ->
?assertEqual(ok, cleanup(#worker_state{})).
-port_test() ->
- ?assertEqual(1080, port(#{port => 1080})),
- ?assertEqual(443, port(#{scheme => "https"})),
- ?assertError(function_clause, port(#{})).
-
-token_uri_map_test() ->
+token_url_test() ->
try
meck:expect(config, get, fun("ibm", "token_url", _) -> "https://nohost.invalid" end),
- ?assertMatch(
- {ok, #{
- scheme := "https",
- path := [],
- host := "nohost.invalid"
- }},
- token_uri_map()
- ),
- meck:expect(config, get, fun("ibm", "token_url", _) -> "https:// " end),
- ?assertEqual({error, invalid_uri}, token_uri_map())
+ ?assertEqual({ok, "https://nohost.invalid"}, token_url()),
+ meck:expect(config, get, fun("ibm", "token_url", _) -> "https://" end),
+ ?assertEqual({error, invalid_uri}, token_url())
after
meck:unload()
end.