feat(scram): supports ACL rules in `scram_restapi` backend

apps/emqx_auth_http/include/emqx_auth_http.hrl

@@ -31,4 +31,6 @@
 -define(AUTHN_TYPE, {?AUTHN_MECHANISM, ?AUTHN_BACKEND}).
 -define(AUTHN_TYPE_SCRAM, {?AUTHN_MECHANISM_SCRAM, ?AUTHN_BACKEND}).
 
+-define(AUTHN_DATA_FIELDS, [is_superuser, client_attrs, expire_at, acl]).
+
 -endif.

apps/emqx_auth_http/src/emqx_authn_http.erl

@@ -32,7 +32,9 @@
     with_validated_config/2,
     generate_request/2,
     request_for_log/2,
-    response_for_log/1
+    response_for_log/1,
+    extract_auth_data/2,
+    safely_parse_body/2
 ]).
 
 %%------------------------------------------------------------------------------
@@ -209,23 +211,28 @@ handle_response(Headers, Body) ->
     case safely_parse_body(ContentType, Body) of
         {ok, NBody} ->
             body_to_auth_data(NBody);
-        {error, Reason} ->
+        {error, _Reason} ->
-            ?TRACE_AUTHN_PROVIDER(
-                error,
-                "parse_http_response_failed",
-                #{content_type => ContentType, body => Body, reason => Reason}
-            ),
             ignore
     end.
 
 body_to_auth_data(Body) ->
     case maps:get(<<"result">>, Body, <<"ignore">>) of
         <<"allow">> ->
+            extract_auth_data(http, Body);
+        <<"deny">> ->
+            {error, not_authorized};
+        <<"ignore">> ->
+            ignore;
+        _ ->
+            ignore
+    end.
+
+extract_auth_data(Source, Body) ->
     IsSuperuser = emqx_authn_utils:is_superuser(Body),
     Attrs = emqx_authn_utils:client_attrs(Body),
     try
         ExpireAt = expire_at(Body),
-                ACL = acl(ExpireAt, Body),
+        ACL = acl(ExpireAt, Source, Body),
         Result = merge_maps([ExpireAt, IsSuperuser, ACL, Attrs]),
         {ok, Result}
     catch
@@ -236,13 +243,6 @@ body_to_auth_data(Body) ->
         throw:Reason ->
             ?TRACE_AUTHN_PROVIDER("bad_response_body", Reason#{http_body => Body}),
             {error, bad_username_or_password}
-            end;
-        <<"deny">> ->
-            {error, not_authorized};
-        <<"ignore">> ->
-            ignore;
-        _ ->
-            ignore
     end.
 
 merge_maps([]) -> #{};
@@ -283,40 +283,43 @@ expire_sec(#{<<"expire_at">> := _}) ->
 expire_sec(_) ->
     undefined.
 
-acl(#{expire_at := ExpireTimeMs}, #{<<"acl">> := Rules}) ->
+acl(#{expire_at := ExpireTimeMs}, Source, #{<<"acl">> := Rules}) ->
     #{
         acl => #{
-            source_for_logging => http,
+            source_for_logging => Source,
             rules => emqx_authz_rule_raw:parse_and_compile_rules(Rules),
             %% It's seconds level precision (like JWT) for authz
             %% see emqx_authz_client_info:check/1
             expire => erlang:convert_time_unit(ExpireTimeMs, millisecond, second)
         }
     };
-acl(_NoExpire, #{<<"acl">> := Rules}) ->
+acl(_NoExpire, Source, #{<<"acl">> := Rules}) ->
     #{
         acl => #{
-            source_for_logging => http,
+            source_for_logging => Source,
             rules => emqx_authz_rule_raw:parse_and_compile_rules(Rules)
         }
     };
-acl(_, _) ->
+acl(_, _, _) ->
     #{}.
 
 safely_parse_body(ContentType, Body) ->
     try
         parse_body(ContentType, Body)
     catch
-        _Class:_Reason ->
+        _Class:Reason ->
+            ?TRACE_AUTHN_PROVIDER(
+                error,
+                "parse_http_response_failed",
+                #{content_type => ContentType, body => Body, reason => Reason}
+            ),
             {error, invalid_body}
     end.
 
 parse_body(<<"application/json", _/binary>>, Body) ->
     {ok, emqx_utils_json:decode(Body, [return_maps])};
 parse_body(<<"application/x-www-form-urlencoded", _/binary>>, Body) ->
-    Flags = [<<"result">>, <<"is_superuser">>],
+    NBody = maps:from_list(cow_qs:parse_qs(Body)),
-    RawMap = maps:from_list(cow_qs:parse_qs(Body)),
-    NBody = maps:with(Flags, RawMap),
     {ok, NBody};
 parse_body(ContentType, _) ->
     {error, {unsupported_content_type, ContentType}}.

apps/emqx_auth_http/src/emqx_authn_scram_restapi.erl

@@ -10,8 +10,11 @@
 
 -module(emqx_authn_scram_restapi).
 
--include_lib("emqx_auth/include/emqx_authn.hrl").
+-feature(maybe_expr, enable).
+
+-include("emqx_auth_http.hrl").
 -include_lib("emqx/include/logger.hrl").
+-include_lib("emqx_auth/include/emqx_authn.hrl").
 
 -behaviour(emqx_authn_provider).
 
@@ -28,10 +31,6 @@
     <<"salt">>
 ]).
 
--define(OPTIONAL_USER_INFO_KEYS, [
-    <<"is_superuser">>
-]).
-
 %%------------------------------------------------------------------------------
 %% APIs
 %%------------------------------------------------------------------------------
@@ -78,7 +77,9 @@ authenticate(
             reason => Reason
         })
     end,
-    emqx_utils_scram:authenticate(AuthMethod, AuthData, AuthCache, RetrieveFun, OnErrFun, State);
+    emqx_utils_scram:authenticate(
+        AuthMethod, AuthData, AuthCache, State, RetrieveFun, OnErrFun, ?AUTHN_DATA_FIELDS
+    );
 authenticate(_Credential, _State) ->
     ignore.
 
@@ -119,16 +120,11 @@ retrieve(
 
 handle_response(Headers, Body) ->
     ContentType = proplists:get_value(<<"content-type">>, Headers),
-    case safely_parse_body(ContentType, Body) of
+    maybe
-        {ok, NBody} ->
+        {ok, NBody} ?= emqx_authn_http:safely_parse_body(ContentType, Body),
-            body_to_user_info(NBody);
+        {ok, UserInfo} ?= body_to_user_info(NBody),
-        {error, Reason} = Error ->
+        {ok, AuthData} ?= emqx_authn_http:extract_auth_data(scram_restapi, NBody),
-            ?TRACE_AUTHN_PROVIDER(
+        {ok, maps:merge(AuthData, UserInfo)}
-                error,
-                "parse_scram_restapi_response_failed",
-                #{content_type => ContentType, body => Body, reason => Reason}
-            ),
-            Error
     end.
 
 body_to_user_info(Body) ->
@@ -137,26 +133,16 @@ body_to_user_info(Body) ->
         true ->
             case safely_convert_hex(Required0) of
                 {ok, Required} ->
-                    UserInfo0 = maps:merge(Required, maps:with(?OPTIONAL_USER_INFO_KEYS, Body)),
+                    {ok, emqx_utils_maps:safe_atom_key_map(Required)};
-                    UserInfo1 = emqx_utils_maps:safe_atom_key_map(UserInfo0),
-                    UserInfo = maps:merge(#{is_superuser => false}, UserInfo1),
-                    {ok, UserInfo};
                 Error ->
+                    ?TRACE_AUTHN_PROVIDER("decode_keys_failed", #{http_body => Body}),
                     Error
             end;
         _ ->
-            ?TRACE_AUTHN_PROVIDER("bad_response_body", #{http_body => Body}),
+            ?TRACE_AUTHN_PROVIDER("missing_requried_keys", #{http_body => Body}),
             {error, bad_response}
     end.
 
-safely_parse_body(ContentType, Body) ->
-    try
-        parse_body(ContentType, Body)
-    catch
-        _Class:_Reason ->
-            {error, invalid_body}
-    end.
-
 safely_convert_hex(Required) ->
     try
         {ok,
@@ -171,15 +157,5 @@ safely_convert_hex(Required) ->
             {error, Reason}
     end.
 
-parse_body(<<"application/json", _/binary>>, Body) ->
-    {ok, emqx_utils_json:decode(Body, [return_maps])};
-parse_body(<<"application/x-www-form-urlencoded", _/binary>>, Body) ->
-    Flags = ?REQUIRED_USER_INFO_KEYS ++ ?OPTIONAL_USER_INFO_KEYS,
-    RawMap = maps:from_list(cow_qs:parse_qs(Body)),
-    NBody = maps:with(Flags, RawMap),
-    {ok, NBody};
-parse_body(ContentType, _) ->
-    {error, {unsupported_content_type, ContentType}}.
-
 merge_scram_conf(Conf, State) ->
     maps:merge(maps:with([algorithm, iteration_count], Conf), State).

apps/emqx_auth_http/test/emqx_authn_scram_restapi_SUITE.erl

@@ -21,6 +21,9 @@
 -define(ALGORITHM_STR, <<"sha512">>).
 -define(ITERATION_COUNT, 4096).
 
+-define(T_ACL_USERNAME, <<"username">>).
+-define(T_ACL_PASSWORD, <<"password">>).
+
 -include_lib("emqx/include/emqx_placeholder.hrl").
 
 all() ->
@@ -120,59 +123,8 @@ t_authenticate(_Config) ->
 
     ok = emqx_config:put([mqtt, idle_timeout], 500),
 
-    {ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
+    {ok, Pid} = create_connection(Username, Password),
-
+    emqx_authn_mqtt_test_client:stop(Pid).
-    ClientFirstMessage = esasl_scram:client_first_message(Username),
-
-    ConnectPacket = ?CONNECT_PACKET(
-        #mqtt_packet_connect{
-            proto_ver = ?MQTT_PROTO_V5,
-            properties = #{
-                'Authentication-Method' => <<"SCRAM-SHA-512">>,
-                'Authentication-Data' => ClientFirstMessage
-            }
-        }
-    ),
-
-    ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
-
-    %% Intentional sleep to trigger idle timeout for the connection not yet authenticated
-    ok = ct:sleep(1000),
-
-    ?AUTH_PACKET(
-        ?RC_CONTINUE_AUTHENTICATION,
-        #{'Authentication-Data' := ServerFirstMessage}
-    ) = receive_packet(),
-
-    {continue, ClientFinalMessage, ClientCache} =
-        esasl_scram:check_server_first_message(
-            ServerFirstMessage,
-            #{
-                client_first_message => ClientFirstMessage,
-                password => Password,
-                algorithm => ?ALGORITHM
-            }
-        ),
-
-    AuthContinuePacket = ?AUTH_PACKET(
-        ?RC_CONTINUE_AUTHENTICATION,
-        #{
-            'Authentication-Method' => <<"SCRAM-SHA-512">>,
-            'Authentication-Data' => ClientFinalMessage
-        }
-    ),
-
-    ok = emqx_authn_mqtt_test_client:send(Pid, AuthContinuePacket),
-
-    ?CONNACK_PACKET(
-        ?RC_SUCCESS,
-        _,
-        #{'Authentication-Data' := ServerFinalMessage}
-    ) = receive_packet(),
-
-    ok = esasl_scram:check_server_final_message(
-        ServerFinalMessage, ClientCache#{algorithm => ?ALGORITHM}
-    ).
 
 t_authenticate_bad_props(_Config) ->
     Username = <<"u">>,
@@ -316,6 +268,47 @@ t_destroy(_Config) ->
         _
     ) = receive_packet().
 
+t_acl(_Config) ->
+    init_auth(),
+
+    ACL = emqx_authn_http_SUITE:acl_rules(),
+    set_user_handler(?T_ACL_USERNAME, ?T_ACL_PASSWORD, #{acl => ACL}),
+    {ok, Pid} = create_connection(?T_ACL_USERNAME, ?T_ACL_PASSWORD),
+
+    Cases = [
+        {allow, <<"http-authn-acl/#">>},
+        {deny, <<"http-authn-acl/1">>},
+        {deny, <<"t/#">>}
+    ],
+
+    try
+        lists:foreach(
+            fun(Case) ->
+                test_acl(Case, Pid)
+            end,
+            Cases
+        )
+    after
+        ok = emqx_authn_mqtt_test_client:stop(Pid)
+    end.
+
+t_auth_expire(_Config) ->
+    init_auth(),
+
+    ExpireSec = 3,
+    WaitTime = timer:seconds(ExpireSec + 1),
+    ACL = emqx_authn_http_SUITE:acl_rules(),
+
+    set_user_handler(?T_ACL_USERNAME, ?T_ACL_PASSWORD, #{
+        acl => ACL,
+        expire_at =>
+            erlang:system_time(second) + ExpireSec
+    }),
+    {ok, Pid} = create_connection(?T_ACL_USERNAME, ?T_ACL_PASSWORD),
+
+    timer:sleep(WaitTime),
+    ?assertEqual(false, erlang:is_process_alive(Pid)).
+
 t_is_superuser() ->
     State = init_auth(),
     ok = test_is_superuser(State, false),
@@ -326,7 +319,7 @@ test_is_superuser(State, ExpectedIsSuperuser) ->
     Username = <<"u">>,
     Password = <<"p">>,
 
-    set_user_handler(Username, Password, ExpectedIsSuperuser),
+    set_user_handler(Username, Password, #{is_superuser => ExpectedIsSuperuser}),
 
     ClientFirstMessage = esasl_scram:client_first_message(Username),
 
@@ -384,19 +377,20 @@ raw_config() ->
     }.
 
 set_user_handler(Username, Password) ->
-    set_user_handler(Username, Password, false).
+    set_user_handler(Username, Password, #{is_superuser => false}).
-set_user_handler(Username, Password, IsSuperuser) ->
+set_user_handler(Username, Password, Extra0) ->
     %% HTTP Server
     Handler = fun(Req0, State) ->
         #{
             username := Username
         } = cowboy_req:match_qs([username], Req0),
 
-        UserInfo = make_user_info(Password, ?ALGORITHM, ?ITERATION_COUNT, IsSuperuser),
+        UserInfo = make_user_info(Password, ?ALGORITHM, ?ITERATION_COUNT),
+        Extra = maps:merge(#{is_superuser => false}, Extra0),
         Req = cowboy_req:reply(
             200,
             #{<<"content-type">> => <<"application/json">>},
-            emqx_utils_json:encode(UserInfo),
+            emqx_utils_json:encode(maps:merge(Extra, UserInfo)),
             Req0
         ),
         {ok, Req, State}
@@ -415,7 +409,7 @@ init_auth(Config) ->
     {ok, [#{state := State}]} = emqx_authn_chains:list_authenticators(?GLOBAL),
     State.
 
-make_user_info(Password, Algorithm, IterationCount, IsSuperuser) ->
+make_user_info(Password, Algorithm, IterationCount) ->
     {StoredKey, ServerKey, Salt} = esasl_scram:generate_authentication_info(
         Password,
         #{
@@ -426,8 +420,7 @@ make_user_info(Password, Algorithm, IterationCount, IsSuperuser) ->
     #{
         stored_key => binary:encode_hex(StoredKey),
         server_key => binary:encode_hex(ServerKey),
-        salt => binary:encode_hex(Salt),
+        salt => binary:encode_hex(Salt)
-        is_superuser => IsSuperuser
     }.
 
 receive_packet() ->
@@ -438,3 +431,79 @@ receive_packet() ->
     after 1000 ->
         ct:fail("Deliver timeout")
     end.
+
+create_connection(Username, Password) ->
+    {ok, Pid} = emqx_authn_mqtt_test_client:start_link("127.0.0.1", 1883),
+
+    ClientFirstMessage = esasl_scram:client_first_message(Username),
+
+    ConnectPacket = ?CONNECT_PACKET(
+        #mqtt_packet_connect{
+            proto_ver = ?MQTT_PROTO_V5,
+            properties = #{
+                'Authentication-Method' => <<"SCRAM-SHA-512">>,
+                'Authentication-Data' => ClientFirstMessage
+            }
+        }
+    ),
+
+    ok = emqx_authn_mqtt_test_client:send(Pid, ConnectPacket),
+
+    %% Intentional sleep to trigger idle timeout for the connection not yet authenticated
+    ok = ct:sleep(1000),
+
+    ?AUTH_PACKET(
+        ?RC_CONTINUE_AUTHENTICATION,
+        #{'Authentication-Data' := ServerFirstMessage}
+    ) = receive_packet(),
+
+    {continue, ClientFinalMessage, ClientCache} =
+        esasl_scram:check_server_first_message(
+            ServerFirstMessage,
+            #{
+                client_first_message => ClientFirstMessage,
+                password => Password,
+                algorithm => ?ALGORITHM
+            }
+        ),
+
+    AuthContinuePacket = ?AUTH_PACKET(
+        ?RC_CONTINUE_AUTHENTICATION,
+        #{
+            'Authentication-Method' => <<"SCRAM-SHA-512">>,
+            'Authentication-Data' => ClientFinalMessage
+        }
+    ),
+
+    ok = emqx_authn_mqtt_test_client:send(Pid, AuthContinuePacket),
+
+    ?CONNACK_PACKET(
+        ?RC_SUCCESS,
+        _,
+        #{'Authentication-Data' := ServerFinalMessage}
+    ) = receive_packet(),
+
+    ok = esasl_scram:check_server_final_message(
+        ServerFinalMessage, ClientCache#{algorithm => ?ALGORITHM}
+    ),
+    {ok, Pid}.
+
+test_acl({allow, Topic}, C) ->
+    ?assertMatch(
+        [0],
+        send_subscribe(C, Topic)
+    );
+test_acl({deny, Topic}, C) ->
+    ?assertMatch(
+        [?RC_NOT_AUTHORIZED],
+        send_subscribe(C, Topic)
+    ).
+
+send_subscribe(Client, Topic) ->
+    TopicOpts = #{nl => 0, rap => 0, rh => 0, qos => 0},
+    Packet = ?SUBSCRIBE_PACKET(1, [{Topic, TopicOpts}]),
+    emqx_authn_mqtt_test_client:send(Client, Packet),
+    timer:sleep(200),
+
+    ?SUBACK_PACKET(1, ReasonCode) = receive_packet(),
+    ReasonCode.

apps/emqx_auth_mnesia/src/emqx_authn_scram_mnesia.erl

@@ -141,7 +141,9 @@ authenticate(
             reason => Reason
         })
     end,
-    emqx_utils_scram:authenticate(AuthMethod, AuthData, AuthCache, RetrieveFun, OnErrFun, State);
+    emqx_utils_scram:authenticate(
+        AuthMethod, AuthData, AuthCache, State, RetrieveFun, OnErrFun, [is_superuser]
+    );
 authenticate(_Credential, _State) ->
     ignore.
 

apps/emqx_utils/src/emqx_utils_scram.erl

@@ -16,17 +16,17 @@
 
 -module(emqx_utils_scram).
 
--export([authenticate/6]).
+-export([authenticate/7]).
 
 %%------------------------------------------------------------------------------
 %% Authentication
 %%------------------------------------------------------------------------------
-authenticate(AuthMethod, AuthData, AuthCache, RetrieveFun, OnErrFun, Conf) ->
+authenticate(AuthMethod, AuthData, AuthCache, Conf, RetrieveFun, OnErrFun, ResultKeys) ->
     case ensure_auth_method(AuthMethod, AuthData, Conf) of
         true ->
             case AuthCache of
                 #{next_step := client_final} ->
-                    check_client_final_message(AuthData, AuthCache, Conf, OnErrFun);
+                    check_client_final_message(AuthData, AuthCache, Conf, OnErrFun, ResultKeys);
                 _ ->
                     check_client_first_message(AuthData, AuthCache, Conf, RetrieveFun, OnErrFun)
             end;
@@ -64,9 +64,7 @@ check_client_first_message(
             {error, not_authorized}
     end.
 
-check_client_final_message(
+check_client_final_message(Bin, Cache, #{algorithm := Alg}, OnErrFun, ResultKeys) ->
-    Bin, #{is_superuser := IsSuperuser} = Cache, #{algorithm := Alg}, OnErrFun
-) ->
     case
         esasl_scram:check_client_final_message(
             Bin,
@@ -74,7 +72,7 @@ check_client_final_message(
         )
     of
         {ok, ServerFinalMessage} ->
-            {ok, #{is_superuser => IsSuperuser}, ServerFinalMessage};
+            {ok, maps:with(ResultKeys, Cache), ServerFinalMessage};
         {error, Reason} ->
             OnErrFun("check_client_final_message_error", Reason),
             {error, not_authorized}