diff --git a/deps/rabbit/src/rabbit_auth_backend_internal.erl b/deps/rabbit/src/rabbit_auth_backend_internal.erl
index 93c1b9e26a..25c5852db0 100644
--- a/deps/rabbit/src/rabbit_auth_backend_internal.erl
+++ b/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.
diff --git a/deps/rabbit/src/rabbit_direct.erl b/deps/rabbit/src/rabbit_direct.erl
index d8d36c5b27..4ef2e6d6bd 100644
--- a/deps/rabbit/src/rabbit_direct.erl
+++ b/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 () ->
@@ -72,8 +72,19 @@ auth_fun({Username, Password}, VHost, ExtraAuthProps) ->
               'broker_not_found_on_node' |
               {'auth_failure', string()} | 'access_refused').
 
+%% Infos is a PropList which contains the content of the Proplist #amqp_adapter_info.additional_info
+%% among other credentials such as protocol, ssl information, etc.
+%% #amqp_adapter_info.additional_info may carry a credential called `authz_bakends` which has the
+%% content of the #user.authz_backends attribute. This means that we are propagating the outcome
+%% from the first successful authentication for the current user when opening an internal
+%% amqp connection. This is particularly relevant for protocol plugins such as AMQP 1.0 where
+%% users are authenticated in one context and later on an internal amqp connection is opened
+%% on a different context. In other words, we do not have anymore the initial credentials presented
+%% during the first authentication. However, we do have the outcome from such successful authentication.
+
 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 +125,13 @@ 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;
+        AuthzBackends -> AuthProps ++ AuthzBackends
+    end.
+
 extract_protocol(Infos) ->
     case proplists:get_value(protocol, Infos, undefined) of
         {Protocol, _Version} ->
diff --git a/deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_reader.erl b/deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_reader.erl
index 48b9770139..674ac8252f 100644
--- a/deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_reader.erl
+++ b/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,
diff --git a/deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_sup.erl b/deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_sup.erl
index dcbca00674..c5fc1f035f 100644
--- a/deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_sup.erl
+++ b/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,19 @@ init([]) ->
                 auto_shutdown => any_significant},
     {ok, {SupFlags, []}}.
 
-adapter_info(Sock) ->
-    amqp_connection:socket_adapter_info(Sock, {'AMQP', "1.0"}).
+
+%% For each AMQP 1.0 session opened, an internal direct AMQP 0-9-1 connection is opened too.
+%% This direct connection will authenticate the user again. Again because at this point
+%% the SASL handshake has already taken place and this user has already been authenticated.
+%% However, we do not have the credentials the user presented. For that reason, the
+%% #amqp_adapter_info.additional_info carries an extra property called authz_backends
+%% which is initialized from the #user.authz_backends attribute. In other words, we
+%% propagate the outcome from the first authentication attempt to the subsequent attempts.
+
+%% See rabbit_direct.erl to see how `authz_bakends` is propagated from
+% amqp_adapter_info.additional_info to the rabbit_access_control module
+
+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}]}.
diff --git a/deps/rabbitmq_auth_backend_oauth2/src/rabbit_auth_backend_oauth2.erl b/deps/rabbitmq_auth_backend_oauth2/src/rabbit_auth_backend_oauth2.erl
index 345b239df3..47912f029b 100644
--- a/deps/rabbitmq_auth_backend_oauth2/src/rabbit_auth_backend_oauth2.erl
+++ b/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
@@ -533,9 +536,23 @@ check_aud(Aud, ResourceServerId) ->
 
 get_scopes(#{?SCOPE_JWT_FIELD := Scope}) -> Scope.
 
+%% A token may be present in the password credential or in the rabbit_auth_backend_oauth2
+%% credential.  The former is the most common scenario for the first time authentication.
+%% However, there are scenarios where the same user (on the same connection) is authenticated
+%% more than once. When this scenario occurs, we extract the token from the credential
+%% called rabbit_auth_backend_oauth2 whose value is the Decoded token returned during the
+%% first authentication. 
+
 -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:
 %%
diff --git a/deps/rabbitmq_auth_backend_oauth2/test/unit_SUITE.erl b/deps/rabbitmq_auth_backend_oauth2/test/unit_SUITE.erl
index f12b272b77..aa76cd3e00 100644
--- a/deps/rabbitmq_auth_backend_oauth2/test/unit_SUITE.erl
+++ b/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}}}],