feat(auth): support HTTP authn return ACL rules

apps/emqx_auth/src/emqx_authz/emqx_authz_rule.erl

@@ -73,7 +73,8 @@
     permission/0,
     who_condition/0,
     action_condition/0,
-    topic_condition/0
+    topic_condition/0,
+    rule/0
 ]).
 
 -type action_precompile() ::

apps/emqx_auth/src/emqx_authz/emqx_authz_rule_raw.erl

@@ -21,7 +21,7 @@
 
 -module(emqx_authz_rule_raw).
 
--export([parse_rule/1, format_rule/1]).
+-export([parse_rule/1, parse_and_compile_rules/1, format_rule/1]).
 
 -include("emqx_authz.hrl").
 
@@ -31,6 +31,27 @@
 %% API
 %%--------------------------------------------------------------------
 
+%% @doc Parse and compile raw ACL rules.
+%% If any bad rule is found, `{bad_acl_rule, ..}' is thrown.
+-spec parse_and_compile_rules([rule_raw()]) -> [emqx_authz_rule:rule()].
+parse_and_compile_rules(Rules) ->
+    lists:map(
+        fun(Rule) ->
+            case parse_rule(Rule) of
+                {ok, {Permission, Action, Topics}} ->
+                    try
+                        emqx_authz_rule:compile({Permission, all, Action, Topics})
+                    catch
+                        throw:Reason ->
+                            throw({bad_acl_rule, Reason})
+                    end;
+                {error, Reason} ->
+                    throw({bad_acl_rule, Reason})
+            end
+        end,
+        Rules
+    ).
+
 -spec parse_rule(rule_raw()) ->
     {ok, {
         emqx_authz_rule:permission(),

apps/emqx_auth_http/src/emqx_auth_http.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_auth_http, [
     {description, "EMQX External HTTP API Authentication and Authorization"},
-    {vsn, "0.2.2"},
+    {vsn, "0.3.0"},
     {registered, []},
     {mod, {emqx_auth_http_app, []}},
     {applications, [

apps/emqx_auth_http/src/emqx_authn_http.erl

@@ -201,19 +201,7 @@ handle_response(Headers, Body) ->
     ContentType = proplists:get_value(<<"content-type">>, Headers),
     case safely_parse_body(ContentType, Body) of
         {ok, NBody} ->
-            case maps:get(<<"result">>, NBody, <<"ignore">>) of
-                <<"allow">> ->
-                    IsSuperuser = emqx_authn_utils:is_superuser(NBody),
-                    Attrs = emqx_authn_utils:client_attrs(NBody),
-                    Result = maps:merge(IsSuperuser, Attrs),
-                    {ok, Result};
-                <<"deny">> ->
-                    {error, not_authorized};
-                <<"ignore">> ->
-                    ignore;
-                _ ->
-                    ignore
-            end;
+            body_to_auth_data(NBody);
         {error, Reason} ->
             ?TRACE_AUTHN_PROVIDER(
                 error,
@@ -223,6 +211,91 @@ handle_response(Headers, Body) ->
             ignore
     end.
 
+body_to_auth_data(Body) ->
+    case maps:get(<<"result">>, Body, <<"ignore">>) of
+        <<"allow">> ->
+            IsSuperuser = emqx_authn_utils:is_superuser(Body),
+            Attrs = emqx_authn_utils:client_attrs(Body),
+            try
+                ExpireAt = expire_at(Body),
+                ACL = acl(ExpireAt, Body),
+                Result = merge_maps([ExpireAt, IsSuperuser, ACL, Attrs]),
+                {ok, Result}
+            catch
+                throw:{bad_acl_rule, Reason} ->
+                    %% it's a invalid token, so ok to log
+                    ?TRACE_AUTHN_PROVIDER("bad_acl_rule", Reason#{http_body => Body}),
+                    {error, bad_username_or_password};
+                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([]) -> #{};
+merge_maps([Map | Maps]) -> maps:merge(Map, merge_maps(Maps)).
+
+%% Return either an empty map, or a map with `expire_at` at millisecond precision
+%% Millisecond precision timestamp is required by `auth_expire_at`
+%% emqx_channel:schedule_connection_expire/1
+expire_at(Body) ->
+    case expire_sec(Body) of
+        undefined ->
+            #{};
+        Sec ->
+            #{expire_at => erlang:convert_time_unit(Sec, second, millisecond)}
+    end.
+
+expire_sec(#{<<"expire_at">> := ExpireTime}) when is_integer(ExpireTime) ->
+    Now = erlang:system_time(second),
+    NowMs = erlang:convert_time_unit(Now, second, millisecond),
+    case ExpireTime < Now of
+        true ->
+            throw(#{
+                cause => "'expire_at' is in the past.",
+                system_time => Now,
+                expire_at => ExpireTime
+            });
+        false when ExpireTime > (NowMs div 2) ->
+            throw(#{
+                cause => "'expire_at' does not appear to be a Unix epoch time in seconds.",
+                system_time => Now,
+                expire_at => ExpireTime
+            });
+        false ->
+            ExpireTime
+    end;
+expire_sec(#{<<"expire_at">> := _}) ->
+    throw(#{cause => "'expire_at' is not an integer (Unix epoch time in seconds)."});
+expire_sec(_) ->
+    undefined.
+
+acl(#{expire_at := ExpireTimeMs}, #{<<"acl">> := Rules}) ->
+    #{
+        acl => #{
+            source_for_logging => http,
+            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 => #{
+            source_for_logging => http,
+            rules => emqx_authz_rule_raw:parse_and_compile_rules(Rules)
+        }
+    };
+acl(_, _) ->
+    #{}.
+
 safely_parse_body(ContentType, Body) ->
     try
         parse_body(ContentType, Body)

apps/emqx_auth_http/test/emqx_authn_http_SUITE.erl

@@ -23,6 +23,7 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("emqx/include/emqx_placeholder.hrl").
+-include_lib("emqx/include/emqx_mqtt.hrl").
 
 -define(PATH, [?CONF_NS_ATOM]).
 
@@ -49,6 +50,21 @@
     })
 ).
 
+-define(SERVER_RESPONSE_WITH_ACL_JSON(ACL),
+    emqx_utils_json:encode(#{
+        result => allow,
+        acl => ACL
+    })
+).
+
+-define(SERVER_RESPONSE_WITH_ACL_JSON(ACL, Expire),
+    emqx_utils_json:encode(#{
+        result => allow,
+        acl => ACL,
+        expire_at => Expire
+    })
+).
+
 -define(SERVER_RESPONSE_URLENCODE(Result, IsSuperuser),
     list_to_binary(
         "result=" ++
@@ -506,6 +522,129 @@ test_ignore_allow_deny({ExpectedValue, ServerResponse}) ->
             )
     end.
 
+t_acl(_Config) ->
+    ACL = acl_rules(),
+    Config = raw_http_auth_config(),
+    {ok, _} = emqx:update_config(
+        ?PATH,
+        {create_authenticator, ?GLOBAL, Config}
+    ),
+    ok = emqx_authn_http_test_server:set_handler(
+        fun(Req0, State) ->
+            Req = cowboy_req:reply(
+                200,
+                #{<<"content-type">> => <<"application/json">>},
+                ?SERVER_RESPONSE_WITH_ACL_JSON(ACL),
+                Req0
+            ),
+            {ok, Req, State}
+        end
+    ),
+    {ok, C} = emqtt:start_link(
+        [
+            {clean_start, true},
+            {proto_ver, v5},
+            {clientid, <<"clientid">>},
+            {username, <<"username">>},
+            {password, <<"password">>}
+        ]
+    ),
+    {ok, _} = emqtt:connect(C),
+    Cases = [
+        {allow, <<"http-authn-acl/#">>},
+        {deny, <<"http-authn-acl/1">>},
+        {deny, <<"t/#">>}
+    ],
+    try
+        lists:foreach(
+            fun(Case) ->
+                test_acl(Case, C)
+            end,
+            Cases
+        )
+    after
+        ok = emqtt:disconnect(C)
+    end.
+
+t_auth_expire(_Config) ->
+    ACL = acl_rules(),
+    Config = raw_http_auth_config(),
+    {ok, _} = emqx:update_config(
+        ?PATH,
+        {create_authenticator, ?GLOBAL, Config}
+    ),
+    ExpireSec = 3,
+    WaitTime = timer:seconds(ExpireSec + 1),
+    Tests = [
+        {<<"ok-to-connect-but-expire-on-pub">>, erlang:system_time(second) + ExpireSec, fun(C) ->
+            {ok, _} = emqtt:connect(C),
+            receive
+                {'DOWN', _Ref, process, C, Reason} ->
+                    ?assertMatch({disconnected, ?RC_NOT_AUTHORIZED, _}, Reason)
+            after WaitTime ->
+                error(timeout)
+            end
+        end},
+        {<<"past">>, erlang:system_time(second) - 1, fun(C) ->
+            ?assertMatch({error, {bad_username_or_password, _}}, emqtt:connect(C)),
+            receive
+                {'DOWN', _Ref, process, C, Reason} ->
+                    ?assertMatch({shutdown, bad_username_or_password}, Reason)
+            end
+        end},
+        {<<"invalid">>, erlang:system_time(millisecond), fun(C) ->
+            ?assertMatch({error, {bad_username_or_password, _}}, emqtt:connect(C)),
+            receive
+                {'DOWN', _Ref, process, C, Reason} ->
+                    ?assertMatch({shutdown, bad_username_or_password}, Reason)
+            end
+        end}
+    ],
+    ok = emqx_authn_http_test_server:set_handler(
+        fun(Req0, State) ->
+            QS = cowboy_req:parse_qs(Req0),
+            {_, Username} = lists:keyfind(<<"username">>, 1, QS),
+            {_, ExpireTime, _} = lists:keyfind(Username, 1, Tests),
+            Req = cowboy_req:reply(
+                200,
+                #{<<"content-type">> => <<"application/json">>},
+                ?SERVER_RESPONSE_WITH_ACL_JSON(ACL, ExpireTime),
+                Req0
+            ),
+            {ok, Req, State}
+        end
+    ),
+    lists:foreach(fun test_auth_expire/1, Tests).
+
+test_auth_expire({Username, _ExpireTime, TestFn}) ->
+    {ok, C} = emqtt:start_link(
+        [
+            {clean_start, true},
+            {proto_ver, v5},
+            {clientid, <<"clientid">>},
+            {username, Username},
+            {password, <<"password">>}
+        ]
+    ),
+    _ = monitor(process, C),
+    unlink(C),
+    try
+        TestFn(C)
+    after
+        [ok = emqtt:disconnect(C) || is_process_alive(C)]
+    end.
+
+test_acl({allow, Topic}, C) ->
+    ?assertMatch(
+        {ok, #{}, [0]},
+        emqtt:subscribe(C, Topic)
+    );
+test_acl({deny, Topic}, C) ->
+    ?assertMatch(
+        {ok, #{}, [?RC_NOT_AUTHORIZED]},
+        emqtt:subscribe(C, Topic)
+    ).
+
 %%------------------------------------------------------------------------------
 %% Helpers
 %%------------------------------------------------------------------------------
@@ -765,3 +904,27 @@ to_list(B) when is_binary(B) ->
     binary_to_list(B);
 to_list(L) when is_list(L) ->
     L.
+
+acl_rules() ->
+    [
+        #{
+            <<"permission">> => <<"allow">>,
+            <<"action">> => <<"pub">>,
+            <<"topics">> => [
+                <<"http-authn-acl/1">>
+            ]
+        },
+        #{
+            <<"permission">> => <<"allow">>,
+            <<"action">> => <<"sub">>,
+            <<"topics">> =>
+                [
+                    <<"eq http-authn-acl/#">>
+                ]
+        },
+        #{
+            <<"permission">> => <<"deny">>,
+            <<"action">> => <<"all">>,
+            <<"topics">> => [<<"#">>]
+        }
+    ].

apps/emqx_auth_jwt/src/emqx_auth_jwt.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_auth_jwt, [
     {description, "EMQX JWT Authentication and Authorization"},
-    {vsn, "0.3.1"},
+    {vsn, "0.3.2"},
     {registered, []},
     {mod, {emqx_auth_jwt_app, []}},
     {applications, [

apps/emqx_auth_jwt/src/emqx_authn_jwt.erl

@@ -402,20 +402,7 @@ binary_to_number(Bin) ->
 parse_rules(Rules) when is_map(Rules) ->
     Rules;
 parse_rules(Rules) when is_list(Rules) ->
-    lists:map(fun parse_rule/1, Rules).
-
-parse_rule(Rule) ->
-    case emqx_authz_rule_raw:parse_rule(Rule) of
-        {ok, {Permission, Action, Topics}} ->
-            try
-                emqx_authz_rule:compile({Permission, all, Action, Topics})
-            catch
-                throw:Reason ->
-                    throw({bad_acl_rule, Reason})
-            end;
-        {error, Reason} ->
-            throw({bad_acl_rule, Reason})
-    end.
+    emqx_authz_rule_raw:parse_and_compile_rules(Rules).
 
 merge_maps([]) -> #{};
 merge_maps([Map | Maps]) -> maps:merge(Map, merge_maps(Maps)).