From 19183b7f9bfc6366dcdcb1ae65013f10f016ba3d Mon Sep 17 00:00:00 2001 From: Robert Newson Date: Mon, 29 Jun 2026 13:27:32 +0100 Subject: [PATCH] couch replication auth plugin for IBM IAM with refresh --- rel/overlay/etc/default.ini | 3 +- .../src/couch_replicator_auth_ibm.erl | 472 ++++++++++++++++++ src/docs/src/config/replicator.rst | 5 +- src/docs/src/replication/replicator.rst | 22 + 4 files changed, 499 insertions(+), 3 deletions(-) create mode 100644 src/couch_replicator/src/couch_replicator_auth_ibm.erl diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini index 0e0eaba6bde..ceffa06a6ae 100644 --- a/rel/overlay/etc/default.ini +++ b/rel/overlay/etc/default.ini @@ -808,8 +808,9 @@ partitioned||* = true ; particular endpoint (source or target). Normally couch_replicator_auth_noop ; would be used at the end of the list as a "catch-all". It doesn't do anything ; and effectively implements the previous behavior of using basic auth. -; There are currently two plugins available: +; There are currently three plugins available: ; couch_replicator_auth_session - use _session cookie authentication +; couch_replicator_auth_ibm - use IBM's IAM service for authentication ; couch_replicator_auth_noop - use basic authentication (previous default) ; Currently, the new _session cookie authentication is tried first, before ; falling back to the old basic authentication default: diff --git a/src/couch_replicator/src/couch_replicator_auth_ibm.erl b/src/couch_replicator/src/couch_replicator_auth_ibm.erl new file mode 100644 index 00000000000..2f96e56f881 --- /dev/null +++ b/src/couch_replicator/src/couch_replicator_auth_ibm.erl @@ -0,0 +1,472 @@ +% 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. + +% This module allows a replication source or target to use an IAM api key for authentication. +% +% Features; +% +% Automatic refreshing of time-limited token before expiration +% Deduplication - only one token will be acquired for each distinct IAM api key +% +% Implementation details +% +% As api keys are sensitive, the only copy of api keys is held in a private ETS table +% owned by this module's gen_server. An opaque reference is returned to clients (this is +% a message authentication code where the key is a non-persisted value generated by the +% gen_server) + +-module(couch_replicator_auth_ibm). + +-behaviour(couch_replicator_auth). +-behaviour(gen_server). + +-export([ + sup_initialize/0, + sup_cleanup/1, + initialize/1, + update_headers/2, + handle_response/3, + cleanup/1 +]). + +-export([ + init/1, + handle_call/3, + handle_cast/2, + handle_info/2, + terminate/2 +]). + +-include_lib("couch_replicator/include/couch_replicator_api_wrap.hrl"). + +-define(EARLY_REFRESH_MS, 300000). +-define(MIN_REFRESH_MS, 10000). +-define(PUBLIC, couch_replicator_auth_ibm_public). +-define(PRIVATE, couch_replicator_auth_ibm_private). + +-record(state, { + token_url, + mac_key, + gun_pid, + gun_mref +}). + +-record(public_entry, { + api_key_mac, + token +}). + +-record(private_entry, { + api_key, + api_key_uuid, + api_key_mac, + gun_stream_ref, + gun_status_code, + gun_body = [], + refresh_ref, + expires_ref, + waiters = [] +}). + +%% callbacks + +sup_initialize() -> + application:ensure_all_started(gun), + {ok, _} = gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +sup_cleanup(_) -> + ok = gen_server:stop(?MODULE). + +initialize(#httpdb{} = HttpDb) -> + case extract_api_key(HttpDb) of + {ok, APIKey} -> + case gen_server:call(?MODULE, {register, APIKey}) of + {ok, APIKeyMAC} -> + {ok, HttpDb, APIKeyMAC}; + {error, Reason} -> + {error, Reason} + end; + {error, _} -> + ignore + end. + +update_headers(APIKeyMAC, Headers) when is_list(Headers) -> + case get_token(APIKeyMAC) of + {ok, Token} -> + {[{~"Authorization", <<"Bearer ", Token/binary>>} | Headers], APIKeyMAC}; + {error, _Reason} -> + {Headers, APIKeyMAC} + end. + +handle_response(APIKeyMAC, _StatusCode, _Headers) -> + {continue, APIKeyMAC}. + +cleanup(_APIKeyMAC) -> + ok. + +get_token(APIKeyMAC) -> + case ets:lookup(?PUBLIC, APIKeyMAC) of + [#public_entry{token = Token}] when is_binary(Token) -> + {ok, Token}; + [#public_entry{}] -> + gen_server:call(?MODULE, {get_token, APIKeyMAC}, token_timeout()); + [] -> + {error, no_token} + end. + +%% gen_server callbacks. + +init(_) -> + 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 + ]), + start_gun(#state{ + mac_key = crypto:strong_rand_bytes(32), + token_url = TokenURL + }); + {error, Reason} -> + {error, Reason} + end. + +handle_call({register, APIKey}, _From, State) -> + case ets:match_object(?PRIVATE, #private_entry{api_key = APIKey, _ = '_'}) of + [#private_entry{} = Entry] -> + {reply, {ok, Entry#private_entry.api_key_mac}, State}; + [] -> + GunStreamRef = acquire_token(APIKey, State), + APIKeyMAC = mac(State#state.mac_key, APIKey), + true = ets:insert_new(?PUBLIC, #public_entry{ + api_key_mac = APIKeyMAC + }), + true = ets:insert_new(?PRIVATE, #private_entry{ + api_key = APIKey, + api_key_mac = APIKeyMAC, + gun_stream_ref = GunStreamRef + }), + {reply, {ok, APIKeyMAC}, State} + end; +handle_call({get_token, APIKeyMAC}, From, State) -> + case ets:lookup(?PUBLIC, APIKeyMAC) of + [] -> + {reply, {error, no_such_api_key}, State}; + [#public_entry{token = Token}] when Token /= undefined -> + {reply, {ok, Token}, State}; + [#public_entry{}] -> + [#private_entry{} = Entry] = ets:lookup(?PRIVATE, APIKeyMAC), + ets:insert(?PRIVATE, Entry#private_entry{waiters = [From | Entry#private_entry.waiters]}), + case Entry of + #private_entry{gun_stream_ref = GunStreamRef} = Entry when + GunStreamRef /= undefined + -> + ok; + #private_entry{} = Entry -> + self() ! {refresh_token, APIKeyMAC} + end, + {noreply, State} + end; +handle_call(_Msg, _From, State) -> + {reply, {error, unexpected_msg}, State}. + +handle_cast(_Msg, State) -> + {noreply, State}. + +handle_info({refresh_token, APIKeyMAC}, State) -> + case ets:lookup(?PRIVATE, APIKeyMAC) of + [] -> + ok; + [#private_entry{gun_stream_ref = undefined} = Entry] -> + couch_log:notice("~p: refreshing api key ~s", [ + ?MODULE, Entry#private_entry.api_key_uuid + ]), + GunStreamRef = acquire_token(Entry#private_entry.api_key, State), + ets:insert(?PRIVATE, Entry#private_entry{gun_stream_ref = GunStreamRef}); + [#private_entry{}] -> + ok + end, + {noreply, State}; +handle_info({expire_api_key, APIKeyMAC}, State) -> + case ets:lookup(?PRIVATE, APIKeyMAC) of + [] -> + ok; + [#private_entry{} = Entry] -> + couch_log:warning("~p: removing expired api key ~s", [ + ?MODULE, Entry#private_entry.api_key_uuid + ]), + ets:delete(?PUBLIC, APIKeyMAC), + ets:delete(?PRIVATE, APIKeyMAC), + cancel_timer(Entry#private_entry.refresh_ref), + cancel_timer(Entry#private_entry.expires_ref) + end, + {noreply, State}; +handle_info( + {gun_response, GunPid, GunStreamRef, nofin, StatusCode, _Headers}, + #state{gun_pid = GunPid} = State +) -> + case match_on_gun_stream_ref(GunStreamRef) of + [#private_entry{} = Entry] -> + ets:insert(?PRIVATE, Entry#private_entry{gun_status_code = StatusCode}); + _ -> + ok + end, + {noreply, State}; +handle_info({gun_data, GunPid, GunStreamRef, nofin, Data}, #state{gun_pid = GunPid} = State) -> + case match_on_gun_stream_ref(GunStreamRef) of + [#private_entry{} = Entry] -> + ets:insert(?PRIVATE, Entry#private_entry{ + gun_body = [Data | Entry#private_entry.gun_body] + }); + _ -> + ok + end, + {noreply, State}; +handle_info({gun_data, GunPid, GunStreamRef, fin, Data}, #state{gun_pid = GunPid} = State) -> + case match_on_gun_stream_ref(GunStreamRef) of + [#private_entry{} = Entry] -> + ResponseBody = [Data | Entry#private_entry.gun_body], + case Entry#private_entry.gun_status_code of + 200 -> + case decode_iam_response(ResponseBody) of + {ok, Token, ExpiresInMs} -> + UUID = api_key_uuid(Token), + couch_log:notice("~p: refreshed api key ~s", [ + ?MODULE, UUID + ]), + cancel_timer(Entry#private_entry.refresh_ref), + cancel_timer(Entry#private_entry.expires_ref), + RefreshRef = erlang:send_after( + max(?MIN_REFRESH_MS, ExpiresInMs - ?EARLY_REFRESH_MS), + self(), + {refresh_token, Entry#private_entry.api_key_mac} + ), + ExpiresRef = erlang:send_after( + ExpiresInMs, + self(), + {expire_api_key, Entry#private_entry.api_key_mac} + ), + true = ets:insert(?PUBLIC, #public_entry{ + api_key_mac = Entry#private_entry.api_key_mac, + token = Token + }), + reply_and_reset( + Entry#private_entry{ + api_key_uuid = UUID, + refresh_ref = RefreshRef, + expires_ref = ExpiresRef + }, + {ok, Token} + ); + {error, Reason} -> + couch_log:notice("~p: failed to refresh api key ~s: ~p", [ + ?MODULE, Entry#private_entry.api_key_uuid, Reason + ]), + reply_and_reset(Entry, {error, Reason}) + end; + StatusCode -> + ErrorMessage = extract_error_message(ResponseBody), + couch_log:notice("~p: failed to refresh api key ~s: ~p", [ + ?MODULE, Entry#private_entry.api_key_uuid, ErrorMessage + ]), + case StatusCode of + 500 -> + erlang:send_after( + ?MIN_REFRESH_MS, + self(), + {refresh_token, Entry#private_entry.api_key_mac} + ); + _ -> + ok + end, + reply_and_reset(Entry, {error, ErrorMessage}) + end; + [] -> + ok + end, + {noreply, State}; +handle_info( + {'DOWN', GunMRef, process, GunPid, Reason}, #state{gun_pid = GunPid, gun_mref = GunMRef} = State +) -> + couch_log:warning("~p: gun process crashed for reason: ~p", [?MODULE, Reason]), + handle_info(restart_gun, State#state{gun_pid = undefined, gun_mref = undefined}); +handle_info({gun_up, GunPid, _Protocol}, #state{gun_pid = GunPid} = State) -> + {noreply, State}; +handle_info({gun_down, GunPid, _Protocol, closed, []}, #state{gun_pid = GunPid} = State) -> + {noreply, State}; +handle_info({gun_down, GunPid, _Protocol, Reason, KilledStreams}, #state{gun_pid = GunPid} = State) -> + couch_log:warning("~p: gun connection down for reason: ~p", [?MODULE, Reason]), + lists:foreach( + fun(GunStreamRef) -> + case match_on_gun_stream_ref(GunStreamRef) of + [#private_entry{} = Entry] -> + reply_and_reset(Entry, {error, Reason}); + [] -> + ok + end + end, + KilledStreams + ), + {noreply, State}; +handle_info(restart_gun, State) -> + case start_gun(State) of + {ok, NewState} -> + {noreply, NewState}; + {error, Reason} -> + couch_log:warning("~p: gun restart failed for reason: ~p", [?MODULE, Reason]), + erlang:send_after(5000, self(), restart_gun), + {noreply, State} + end; +handle_info(Msg, State) -> + couch_log:warning("~p: unexpected info ~p", [?MODULE, Msg]), + {noreply, State}. + +terminate(_Reason, State) -> + ok = gun:close(State#state.gun_pid), + ets:foldl( + fun(#private_entry{} = Entry, Acc) -> + cancel_timer(Entry#private_entry.refresh_ref), + cancel_timer(Entry#private_entry.expires_ref), + lists:foreach( + fun(W) -> gen_server:reply(W, {error, terminated}) end, + Entry#private_entry.waiters + ), + Acc + end, + ok, + ?PRIVATE + ). + +%% private functions. + +extract_api_key(#httpdb{auth_props = AuthProps}) -> + case proplists:get_value(~"ibm", AuthProps) of + {IBMProps} when is_list(IBMProps) -> + case proplists:get_value(~"api_key", IBMProps) of + APIKey when is_binary(APIKey), byte_size(APIKey) > 0 -> + {ok, APIKey}; + _ -> + {error, missing_api_key} + end; + _ -> + {error, missing_api_key} + end. + +acquire_token(APIKey, #state{} = State) -> + Headers = [ + {~"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} = uri_string:parse(State#state.token_url), + gun:post(State#state.gun_pid, Path, Headers, Body). + +decode_iam_response(ResponseBody) -> + try jiffy:decode(ResponseBody, [return_maps]) of + Decoded when is_map(Decoded) -> + Token = maps:get(~"access_token", Decoded, undefined), + ExpiresInSecs = maps:get(~"expires_in", Decoded, undefined), + if + is_binary(Token) andalso is_integer(ExpiresInSecs) -> + {ok, Token, ExpiresInSecs * 1000}; + true -> + {error, malformed_iam_response} + end; + _ -> + {error, malformed_iam_response} + catch + _:_ -> + {error, malformed_iam_response} + end. + +extract_error_message(ResponseBody) -> + try jiffy:decode(ResponseBody, [return_maps]) of + #{~"errorCode" := ErrorCode, ~"errorMessage" := ErrorMessage} -> + {ErrorCode, ErrorMessage}; + _ -> + malformed_iam_response + catch + _:_ -> + malformed_iam_response + end. + +api_key_uuid(EncodedToken) when is_binary(EncodedToken) -> + [_Header, Payload, _Signature] = binary:split(EncodedToken, ~".", [global]), + case b64url:decode(Payload) of + {error, _Reason} -> + ~""; + Decoded when is_binary(Decoded) -> + case jiffy:decode(Decoded, [return_maps]) of + #{~"apikey_uuid" := UUID} -> + UUID; + _ -> + ~"" + end + end. + +token_url() -> + URI = config:get("ibm", "token_url", "https://iam.cloud.ibm.com/identity/token"), + case uri_string:parse(URI) of + #{scheme := "https"} -> + {ok, URI}; + #{} -> + {error, not_https}; + {error, Reason, _Details} -> + {error, Reason} + end. + +token_timeout() -> + config:get_integer("ibm", "token_timeout", 30000). + +match_on_gun_stream_ref(GunStreamRef) -> + ets:match_object(?PRIVATE, #private_entry{gun_stream_ref = GunStreamRef, _ = '_'}). + +reply_and_reset(#private_entry{} = Entry, Reply) -> + ets:insert(?PRIVATE, Entry#private_entry{ + gun_stream_ref = undefined, + gun_status_code = undefined, + gun_body = [], + waiters = [] + }), + lists:foreach( + fun(W) -> gen_server:reply(W, Reply) end, + Entry#private_entry.waiters + ). + +cancel_timer(undefined) -> + ok; +cancel_timer(TimerRef) when is_reference(TimerRef) -> + erlang:cancel_timer(TimerRef). + +start_gun(#state{} = State) -> + #{host := Host} = ParsedUrl = uri_string:parse(State#state.token_url), + Options = #{tls_opts => [{cacerts, couch_replicator_utils:cacert_get()}]}, + {ok, GunPid} = gun:open(Host, port(ParsedUrl), Options), + case gun:await_up(GunPid) of + {ok, _Protocol} -> + couch_log:notice("~p: gun connection up", [?MODULE]), + MRef = monitor(process, GunPid), + {ok, State#state{gun_pid = GunPid, gun_mref = MRef}}; + {error, Reason} -> + {error, Reason} + end. + +port(#{port := Port}) -> + Port; +port(#{scheme := "https"}) -> + 443. + +mac(Key, Data) when is_binary(Key), is_binary(Data) -> + crypto:mac(hmac, sha256, Key, Data). diff --git a/src/docs/src/config/replicator.rst b/src/docs/src/config/replicator.rst index ceb3eebbaa2..b86d018d6c7 100644 --- a/src/docs/src/config/replicator.rst +++ b/src/docs/src/config/replicator.rst @@ -370,9 +370,10 @@ Replicator Database Configuration List of replicator client authentication plugins. Plugins will be tried in order and the first to initialize successfully will - be used. By default there are two plugins available: + be used. By default there are three plugins available: `couch_replicator_auth_session` implementing session (cookie) - authentication, and `couch_replicator_auth_noop` implementing basic + authentication, `couch_replicator_auth_ibm` integrating + with IBM's IAM service and `couch_replicator_auth_noop` implementing basic authentication. For backwards compatibility, the no-op plugin should be used at the end of the plugin list:: diff --git a/src/docs/src/replication/replicator.rst b/src/docs/src/replication/replicator.rst index 3d8b6bf7bab..a3c0c883562 100644 --- a/src/docs/src/replication/replicator.rst +++ b/src/docs/src/replication/replicator.rst @@ -809,6 +809,28 @@ they are used. If they are not, then URL userinfo is checked. If credentials are found there, then those credentials are used, otherwise basic auth header is used. +Using an IBM IAM apikey +======================= + +If you've enabled the optional `couch_replicator_auth_ibm_auth` +replicator authentication plugin you can specify an IBM IAM apikey in +your source or target as follows; + + .. code-block:: javascript + + { + "target": { + "url": "http://someurl.com/mydb", + "auth": { + "ibm": { + "apikey": "$apikey" + } + } + }, + ... + } + + Replicate Winning Revisions Only ================================