couch replication auth plugin for IBM IAM with refresh

[rnewson]

Overview

Teach the CouchDB replicator how to acquire an access token from an apikey from IBM's IAM authentication service.

Note that an access token has an expiration (currently one hour but subject to change).

The plugin fetches a new token 5 minutes before the currently held one expires.

Testing recommendations

will be covered by tests

Related Issues or Pull Requests

N/A

Checklist

• This is my own work, I did not use AI, LLM's or similar technology
• Code is written and works correctly
• Changes are covered by tests
• Any new configurable parameters are documented in rel/overlay/etc/default.ini
• Documentation changes were made in the src/docs folder
• Documentation changes were backported (separated PR) to affected branches

[nickva]

This is a good start. But I think we can simplify it by using the same pattern as in https://github.com/apache/couchdb/blob/main/src/couch_replicator/src/couch_replicator_auth_session.erl with a gen_server, also use a first type synchronous token fetch in init.

For ibrowse requests don't use the default lb pool, replicator has it's own connection pool so we'd be using that instead (see https://github.com/apache/couchdb/blob/main/src/couch_replicator/src/couch_replicator_auth_session.erl).

Also wondering how easy it would be to have this not just IBM-specific. We have JWT token auth for our server side. Maybe this can be adapted somehow to work in a more generic way so it could work with Azure, AWS, other IAM-like auth systems. But, perhaps configuring it then would be tricky and keeping the plugins separate just to be able to configure the token url, then maybe separate plugins is just easier.

For config and naming maybe we should use some indication this for a replicator auth plugin not some generic IBM integration (thinking of "ibm_iam" section).

[nickva]

It might be better to do a first time synchronous token fetch on init. So when the job starts we'd know right away it has an invalid creds

[rnewson]

I had that earlier but it entailed duplication of the 'get token' ibrowse code (one synchronous, the other not). so I simplified to one version.

[nickva]

That's why I think we can just use a similar structure as the session plugin and always use the synchronous version.

[rnewson]

I do want to think of a more generic thing, and we should maybe make it possible for an auth plugin to have application like semantics (so they can start a supervision tree etc). in that version things become simpler. the auth plugin would launch a gen_server that makes a public ETS table. each replication job would insert the apikey when it starts and then fetch the token in update_headers as needed. the gen_server could refresh all the tokens slightly in advance of their expiration, handle retries on timeout or 500's, etc.

[nickva]

I tried adding a supervisor-level API for replicator auth plugins #6070

[nickva]

Looks very nice. Using gun is great and like the shared ets and using sigils.

One thing that might be possible to simplify not handling all the gun async messages in the gen_server, it's easy to miss some and get into a blocked or stuck state. I think we could still avoid blocking the gen_server with a gun_await/await_body inline if we spawn a separate process to handle the connection and just return the result/error in a single DOWN message (kind of like how we have opener processes for couch_server / indexes).

IBM refresh time is an hour so even for 100 different keys it's only ~30 second intervals and most likely users will just use a few keys only.

So maybe something like

{_, Ref} = spawn_monitor(fun() -> 
  exit(try fetch(URIMap, APIKey, Timeout) catch T:E -> {error, {T,E}} end)
end)

fetch(...)->  
   gun:post(...)
   case gun:await(...) ->
      {response, nofin, 200} ->
           {ok, Body} = gun:await_body(...)
           decode(Body);
       ...

  end.

[nickva]

Is there a gun_response with fin to handle? Got headers but no body, maybe some redirect or 500 case?

[rnewson]

hm, yes, perhaps. will add a clause in case (an http 204, say. IAM don't say they send it, but we should handle every possibility)

[nickva]

Noticed a few more things. As far as I can tell we account for most of the gun async clauses. However I think given how rare refreshes happen we could simplify this quite a bit by using a separate process for the request and not worry them at all.

acquire_proc(APIKeyMAC, APIKey, #state{token_uri_map = URIMap} = State) ->
    {_, Ref} = spawn_monitor(fun() ->
        exit(try acquire(URIMap, APIKey) catch Tag:Err -> {error, {Tag, Err}} end)
    end),
    ets:update_element(?PRIVATE, APIKeyMAC, {#private_entry.req, Ref}),
    State#state{reqs = Reqs#{Ref => APIKeyMac}}.

acquire(#{host := Host} = URIMap, APIKey) ->
    Timeout = timeout_for_this_job(),
    Opts = #{
        connect_timeout => Timeout,
        tls_opts => [{cacerts, couch_replicator_utils:cacert_get()}]
    },
    {ok, Pid} = gun:open(Host, port(URIMap), Opts),
    {ok, _Proto} = gun:await_up(ConnPid, Timeout),
    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}
    ]),
    StreamRef = gun:post(ConnPid, fix_path(maps:get(path, URIMap)), Headers, Body),
    case gun:await(ConnPid, StreamRef, Timeout) of
        {response, nofin, 200, _Headers} ->
            {ok, RespBody} = gun:await_body(ConnPid, StreamRef, Timeout),
            decode_iam_response(RespBody);
        {response, nofin, StatusCode, _Headers} ->
            {ok, RespBody} = gun:await_body(ConnPid, StreamRef, Timeout),
            {error, {http_error, StatusCode, get_error(RespBody)}};
        {response, fin, StatusCode, _Headers} ->
            {error, {http_error, StatusCode, no_body}};
        {error, Reason} ->
            {error, Reason}
    end.

We always exit with a success, or an error (include a timeout). Then we can skip handling all 10 or so gun async messages in the main gen_server and still work asynchronously. gun:await reference https://ninenines.eu/docs/en/gun/2.5/manual/gun.await/

The only clause we'd need it to handle in the main gen_server is the result in a DOWN:

handle_info({'DOWN', Ref, _, _, Result}, #state{reqs = Reqs} = State) ->
    case maps:take(Ref, Reqs) of
        {APIKeyMAC, Reqs1} ->
            handle_result(APIKeyMAC, Result),
            {noreply, State#state{reqs=Reqs1}};
        error ->
            {noreply, State}
    end.

handle_result(APIMacKey, Result) ->
    case ets:lookup(?PRIVATE, APIKeyMac) of
         [] -> % missing?
              ok;
         [#private_entry{} = Entry] ->
             case Result of
                      {ok, Token, Expiry} -> 
                          ....reset refreshes, timeout, insert token, reply to waiter
                       {error, Error} ->
                          .... retries?logs
                          reply_to_waiters_with_error(Error);
                        Other ->
                              ... maybe a timeout
                            reply_to_waiters_with_error({error, Error})
              end
end)

[nickva]

Will this always work? If either last_used or token_update_at might be undefined or a non-integer we might get unexpected states here. Monotonic times also always start with negative values. It should be fine to compare them as both of them came from the same monotonic clock and are initialized from it.

[rnewson]

last_used is initialized to current time on first entry but token_updated_at is undefined until we get the IAM response, but we don't set a timer to send a refresh_token event until we get that response. if that response is successful we set token_updated_at to the current time. if not, we don't set it, and so last_used > undefined is false for any value of last_used so we do nothing here, which is what we'd want.

not sure how to make it clearer, I agree it's subtle. I could use another atom as the default value for token_updated_at but it would never need to be referred to again in the code so that seems odd.

[nickva]

but we don't set a timer to send a refresh_token event until we get that response.

not sure how to make it clearer, I agree it's subtle. I could use another atom as the default value for token_updated_at but it would never need to be referred to again in the code so that seems odd.

What if we don't get a successful response but a 500? We reset and fire another refresh later, that comes to the same handler compares last_used > token_update_at (=undefined) so it's false, we "let it expire". It expires, job crashes and restarts. When registering it sees the existing entry so it goes back to refresh and it goes into a loop?

[nickva]

Wonder if something like this would make sense instead of the comparison

% refresh if never got a token (updated_at=undefined), or any callers are waiting,  or was used since last refresh
should_refresh(#private_entry{token_updated_at = undefined}) ->
    true;
should_refresh(#private_entry{waiters = [_ | _]}) ->
    true;
should_refresh(#private_entry{last_used = LastUsed, token_updated_at = TokenUpdatedAt}) ->
    LastUsed > TokenUpdatedAt

[nickva]

What happens to any in-flight stream refs in this case? If the process just goes away we'd still be left with stale stream refs referencing refreshes in process that will never finish.

We do:

                #private_entry{gun_stream_ref = GunStreamRef} when
                    GunStreamRef /= undefined
                ->

But we don't match and see if it's an active ref or something that went away and if match some ref (not necessarily) the current ref we'd fall through and wait for a refresh to finish. Perhaps we have a guarantees that when gun just never crashes or dies that way without sending us something back...but without that we'd want some way to always clean up stale entries thinger.

[nickva]

Added a few in-line comments mostly around lifecyle and cleanups. I think in some case we can get stale or always looping cases.

I still think this can be a lot simpler where we can use something like couch_gun from https://github.com/apache/couchdb/blob/gun-for-tests/src/couch/src/couch_gun.erl in a separate process not having to handle low level intermediate streaming responses like gun_error, headers only, data only, process up/down etc. That should be handled by gun somewhere in a separate process, and we'd just care about whether the request completed and returned with ok|error response.

With deduplication and 1 hour refresh period I think it's a decent tradeoff for simplicity. We can always add the extra async processing if this becomes a major bottleneck

[nickva]

It looks a bit better as the undefined comparison is gone however if in case of a 500 triggered refresh we'd still go into a loop and get stuck as last_used > token_updated_at will always stay false?