Fix issue #6909

Use the outcome from first authentication
stored in the #user.authz_backends to authenticate
subsequent attempts which occur when a session is
opened.
In particular, during the first authentication attempt
which occurs during the sasl handshake, the amqp 1.0
plugins reads and validates JWT token present in the
password field.
When a new AMQP 1.0 session is opened, the plugin creates
an internal AMQP connection which triggers a second/nth
authentication. For this second/nth authentication, the
plugin propagates as Authentication Credentials the outcome
from the first authentication which is stored in the
`#user.authz_backends`.
The Oauth2 backend first attempts to authenticate using
the password credentials else it uses the credential with the
key `rabbit_auth_backend_oauth2` which has a function which
returns the decoded token

deps/rabbit/src/rabbit_auth_backend_internal.erl

@@ -92,7 +92,11 @@ user_login_authentication(Username, AuthProps) ->
                           false
                   end
               end);
-        false -> exit({unknown_auth_props, Username, AuthProps})
+        false ->
+            case proplists:get_value(rabbit_auth_backend_internal, AuthProps, undefined) of
+                undefined -> exit({unknown_auth_props, Username, AuthProps});
+                _ -> internal_check_user_login(Username, fun(_) -> true end)
+            end
     end.
 
 state_can_expire() -> false.

deps/rabbit/src/rabbit_direct.erl

@@ -52,8 +52,8 @@ list() ->
 auth_fun({none, _}, _VHost, _ExtraAuthProps) ->
     fun () -> {ok, rabbit_auth_backend_dummy:user()} end;
 
-auth_fun({Username, none}, _VHost, _ExtraAuthProps) ->
-    fun () -> rabbit_access_control:check_user_login(Username, []) end;
+auth_fun({Username, none}, _VHost, ExtraAuthProps) ->
+    fun () -> rabbit_access_control:check_user_login(Username, [] ++ ExtraAuthProps) end;
 
 auth_fun({Username, Password}, VHost, ExtraAuthProps) ->
     fun () ->
@@ -73,7 +73,8 @@ auth_fun({Username, Password}, VHost, ExtraAuthProps) ->
               {'auth_failure', string()} | 'access_refused').
 
 connect(Creds, VHost, Protocol, Pid, Infos) ->
-    ExtraAuthProps = extract_extra_auth_props(Creds, VHost, Pid, Infos),
+    ExtraAuthProps = append_authz_backends(extract_extra_auth_props(Creds, VHost, Pid, Infos), Infos),
+
     AuthFun = auth_fun(Creds, VHost, ExtraAuthProps),
     case rabbit_boot_state:has_reached_and_is_active(core_started) of
         true  ->
@@ -114,6 +115,12 @@ extract_extra_auth_props(Creds, VHost, Pid, Infos) ->
             maybe_call_connection_info_module(Protocol, Creds, VHost, Pid, Infos)
     end.
 
+append_authz_backends(AuthProps, Infos) ->
+    case proplists:get_value(authz_backends, Infos, undefined) of
+        undefined -> AuthProps;
+        Authz_backends -> AuthProps ++ Authz_backends
+    end.
+
 extract_protocol(Infos) ->
     case proplists:get_value(protocol, Infos, undefined) of
         {Protocol, _Version} ->

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_reader.erl

@@ -710,6 +710,7 @@ send_to_new_1_0_session(Channel, Frame, State) ->
                                     user      = User},
         proxy_socket = ProxySocket} = State,
     %% Note: the equivalent, start_channel is in channel_sup_sup
+
     case rabbit_amqp1_0_session_sup_sup:start_session(
            %% NB subtract fixed frame header size
            ChanSupSup, {amqp10_framing, Sock, Channel,

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_sup.erl

@@ -29,7 +29,7 @@
 
 %%----------------------------------------------------------------------------
 start_link({amqp10_framing, Sock, Channel, FrameMax, ReaderPid,
-            Username, VHost, Collector, ProxySocket}) ->
+            User, VHost, Collector, ProxySocket}) ->
     {ok, SupPid} = supervisor:start_link(?MODULE, []),
     {ok, WriterPid} =
         supervisor:start_child(
@@ -61,8 +61,8 @@ start_link({amqp10_framing, Sock, Channel, FrameMax, ReaderPid,
                id => channel,
                start =>
                    {rabbit_amqp1_0_session_process, start_link, [
-                       {Channel, ReaderPid, WriterPid, Username, VHost, FrameMax,
-                           adapter_info(SocketForAdapterInfo), Collector}
+                       {Channel, ReaderPid, WriterPid, User, VHost, FrameMax,
+                           adapter_info(User, SocketForAdapterInfo), Collector}
                    ]},
                restart => transient,
                significant => true,
@@ -86,5 +86,7 @@ init([]) ->
                 auto_shutdown => any_significant},
     {ok, {SupFlags, []}}.
 
-adapter_info(Sock) ->
-    amqp_connection:socket_adapter_info(Sock, {'AMQP', "1.0"}).
+adapter_info(User, Sock) ->
+    AdapterInfo = amqp_connection:socket_adapter_info(Sock, {'AMQP', "1.0"}),
+    AdapterInfo#amqp_adapter_info{additional_info =
+        AdapterInfo#amqp_adapter_info.additional_info ++ [{authz_backends, User#user.authz_backends}]}.

deps/rabbitmq_auth_backend_oauth2/src/rabbit_auth_backend_oauth2.erl

@@ -167,7 +167,7 @@ validate_token_expiry(#{<<"exp">> := Exp}) when is_integer(Exp) ->
     end;
 validate_token_expiry(#{}) -> ok.
 
--spec check_token(binary()) ->
+-spec check_token(binary() | map()) ->
           {'ok', map()} |
           {'error', term() }|
           {'refused',
@@ -175,6 +175,9 @@ validate_token_expiry(#{}) -> ok.
            {'error', term()} |
            {'invalid_aud', term()}}.
 
+check_token(DecodedToken) when is_map(DecodedToken) ->
+    {ok, DecodedToken};
+
 check_token(Token) ->
     Settings = application:get_all_env(?APP),
     case uaa_jwt:decode_and_verify(Token) of
@@ -535,7 +538,14 @@ get_scopes(#{?SCOPE_JWT_FIELD := Scope}) -> Scope.
 
 -spec token_from_context(map()) -> binary() | undefined.
 token_from_context(AuthProps) ->
-    maps:get(password, AuthProps, undefined).
+    case maps:get(password, AuthProps, undefined) of
+        undefined ->
+            case maps:get(rabbit_auth_backend_oauth2, AuthProps, undefined) of
+                undefined -> undefined;
+                Impl -> Impl()
+            end;
+        Token -> Token
+    end.
 
 %% Decoded tokens look like this:
 %%

deps/rabbitmq_auth_backend_oauth2/test/unit_SUITE.erl

@@ -12,6 +12,7 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 
+
 all() ->
     [
         test_own_scope,
@@ -19,6 +20,7 @@ all() ->
         test_validate_payload,
         test_validate_payload_when_verify_aud_false,
         test_successful_access_with_a_token,
+        test_successful_access_with_a_parsed_token,
         test_successful_access_with_a_token_that_has_tag_scopes,
         test_unsuccessful_access_with_a_bogus_token,
         test_restricted_vhost_access_with_a_valid_token,
@@ -629,6 +631,22 @@ test_successful_access_with_a_token(_) ->
 
     assert_topic_access_granted(User, VHost, <<"bar">>, read, #{routing_key => <<"#/foo">>}).
 
+test_successful_access_with_a_parsed_token(_) ->
+    Jwk = ?UTIL_MOD:fixture_jwk(),
+    UaaEnv = [{signing_keys, #{<<"token-key">> => {map, Jwk}}}],
+    application:set_env(rabbitmq_auth_backend_oauth2, key_config, UaaEnv),
+    application:set_env(rabbitmq_auth_backend_oauth2, resource_server_id, <<"rabbitmq">>),
+
+    VHost    = <<"vhost">>,
+    Username = <<"username">>,
+    Token    = ?UTIL_MOD:sign_token_hs(?UTIL_MOD:token_with_sub(?UTIL_MOD:fixture_token(), Username), Jwk),
+    {ok, #auth_user{impl = Impl} } =
+        rabbit_auth_backend_oauth2:user_login_authentication(Username, [{password, Token}]),
+
+    {ok, _ } =
+        rabbit_auth_backend_oauth2:user_login_authentication(Username, [{rabbit_auth_backend_oauth2, Impl}]).
+
+
 test_successful_access_with_a_token_that_has_tag_scopes(_) ->
     Jwk = ?UTIL_MOD:fixture_jwk(),
     UaaEnv = [{signing_keys, #{<<"token-key">> => {map, Jwk}}}],