yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(authorization): moves authorization configuration items from zone to root

This commit is contained in:
zhanghongtong 2021-08-19 10:27:32 +08:00 committed by Rory Z
parent 87881621bb
commit 5652917af6
6 changed files with 93 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
apps/emqx/etc/emqx.conf
View File

@ -88,6 +88,48 @@ broker {
perf.trie_compaction = true perf.trie_compaction = true
} }
authorization {
## Behaviour after not matching a rule.
##
## @doc authorization.no_match
## ValueType: allow | deny
## Default: allow
no_match: allow
## The action when authorization check reject current operation
##
## @doc authorization.deny_action
## ValueType: ignore | disconnect
## Default: ignore
deny_action: ignore
## Whether to enable Authorization cache.
##
## If enabled, Authorization roles for each client will be cached in the memory
##
## @doc authorization.cache.enable
## ValueType: Boolean
## Default: true
cache.enable: true
## The maximum count of Authorization entries can be cached for a client.
##
## @doc authorization.cache.max_size
## ValueType: Integer
## Range: [0, 1048576]
## Default: 32
cache.max_size: 32
## The time after which an Authorization cache entry will be deleted
##
## @doc authorization.cache.ttl
## ValueType: Duration
## Default: 1m
cache.ttl: 1m
}
##================================================================== ##==================================================================
## Zones and Listeners ## Zones and Listeners
##================================================================== ##==================================================================
@ -114,7 +156,6 @@ broker {
## - `auth.*` ## - `auth.*`
## - `stats.*` ## - `stats.*`
## - `mqtt.*` ## - `mqtt.*`
## - `authorization.*`
## - `flapping_detect.*` ## - `flapping_detect.*`
## - `force_shutdown.*` ## - `force_shutdown.*`
## - `conn_congestion.*` ## - `conn_congestion.*`
@ -396,47 +437,6 @@ zones.default {
} }
authorization {
## Enable Authorization check.
##
## @doc zones.<name>.authorization.enable
## ValueType: Boolean
## Default: true
enable = true
## The action when authorization check reject current operation
##
## @doc zones.<name>.authorization.deny_action
## ValueType: ignore | disconnect
## Default: ignore
deny_action = ignore
## Whether to enable Authorization cache.
##
## If enabled, Authorization roles for each client will be cached in the memory
##
## @doc zones.<name>.authorization.cache.enable
## ValueType: Boolean
## Default: true
cache.enable = true
## The maximum count of Authorization entries can be cached for a client.
##
## @doc zones.<name>.authorization.cache.max_size
## ValueType: Integer
## Range: [0, 1048576]
## Default: 32
cache.max_size = 32
## The time after which an Authorization cache entry will be deleted
##
## @doc zones.<name>.authorization.cache.ttl
## ValueType: Duration
## Default: 1m
cache.ttl = 1m
}
flapping_detect { flapping_detect {
## Enable Flapping Detection. ## Enable Flapping Detection.
## ##
@ -1158,7 +1158,6 @@ zones.default {
#This is an example zone which has less "strict" settings. #This is an example zone which has less "strict" settings.
#It's useful to clients connecting the broker from trusted networks. #It's useful to clients connecting the broker from trusted networks.
zones.internal { zones.internal {
authorization.enable = true
auth.enable = false auth.enable = false
listeners.mqtt_internal { listeners.mqtt_internal {
type = tcp type = tcp

13
apps/emqx/src/emqx_access_control.erl
View File

@ -39,23 +39,24 @@ authenticate(Credential) ->
%% @doc Check Authorization %% @doc Check Authorization
-spec authorize(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic()) -spec authorize(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic())
-> allow | deny. -> allow | deny.
authorize(ClientInfo = #{zone := Zone}, PubSub, Topic) -> authorize(ClientInfo, PubSub, Topic) ->
case emqx_authz_cache:is_enabled(Zone) of case emqx_authz_cache:is_enabled() of
true -> check_authorization_cache(ClientInfo, PubSub, Topic); true -> check_authorization_cache(ClientInfo, PubSub, Topic);
false -> do_authorize(ClientInfo, PubSub, Topic) false -> do_authorize(ClientInfo, PubSub, Topic)
end. end.
check_authorization_cache(ClientInfo = #{zone := Zone}, PubSub, Topic) -> check_authorization_cache(ClientInfo, PubSub, Topic) ->
case emqx_authz_cache:get_authz_cache(Zone, PubSub, Topic) of case emqx_authz_cache:get_authz_cache(PubSub, Topic) of
not_found -> not_found ->
AuthzResult = do_authorize(ClientInfo, PubSub, Topic), AuthzResult = do_authorize(ClientInfo, PubSub, Topic),
emqx_authz_cache:put_authz_cache(Zone, PubSub, Topic, AuthzResult), emqx_authz_cache:put_authz_cache(PubSub, Topic, AuthzResult),
AuthzResult; AuthzResult;
AuthzResult -> AuthzResult AuthzResult -> AuthzResult
end. end.
do_authorize(ClientInfo, PubSub, Topic) -> do_authorize(ClientInfo, PubSub, Topic) ->
case run_hooks('client.authorize', [ClientInfo, PubSub, Topic], allow) of NoMatch = emqx:get_config([authorization, no_match], allow),
case run_hooks('client.authorize', [ClientInfo, PubSub, Topic], NoMatch) of
allow -> allow; allow -> allow;
_Other -> deny _Other -> deny
end. end.

64
apps/emqx/src/emqx_authz_cache.erl
View File

@ -18,15 +18,15 @@
-include("emqx.hrl"). -include("emqx.hrl").
-export([ list_authz_cache/1 -export([ list_authz_cache/0
, get_authz_cache/3 , get_authz_cache/2
, put_authz_cache/4 , put_authz_cache/3
, cleanup_authz_cache/1 , cleanup_authz_cache/0
, empty_authz_cache/0 , empty_authz_cache/0
, dump_authz_cache/0 , dump_authz_cache/0
, get_cache_max_size/1 , get_cache_max_size/0
, get_cache_ttl/1 , get_cache_ttl/0
, is_enabled/1 , is_enabled/0
, drain_cache/0 , drain_cache/0
]). ]).
@ -50,45 +50,45 @@ cache_k(PubSub, Topic)-> {PubSub, Topic}.
cache_v(AuthzResult)-> {AuthzResult, time_now()}. cache_v(AuthzResult)-> {AuthzResult, time_now()}.
drain_k() -> {?MODULE, drain_timestamp}. drain_k() -> {?MODULE, drain_timestamp}.
-spec(is_enabled(atom()) -> boolean()). -spec(is_enabled() -> boolean()).
is_enabled(Zone) -> is_enabled() ->
emqx_config:get_zone_conf(Zone, [authorization, cache, enable]). emqx:get_config([authorization, cache, enable], false).
-spec(get_cache_max_size(atom()) -> integer()). -spec(get_cache_max_size() -> integer()).
get_cache_max_size(Zone) -> get_cache_max_size() ->
emqx_config:get_zone_conf(Zone, [authorization, cache, max_size]). emqx:get_config([authorization, cache, max_size]).
-spec(get_cache_ttl(atom()) -> integer()). -spec(get_cache_ttl() -> integer()).
get_cache_ttl(Zone) -> get_cache_ttl() ->
emqx_config:get_zone_conf(Zone, [authorization, cache, ttl]). emqx:get_config([authorization, cache, ttl]).
-spec(list_authz_cache(atom()) -> [authz_cache_entry()]). -spec(list_authz_cache() -> [authz_cache_entry()]).
list_authz_cache(Zone) -> list_authz_cache() ->
cleanup_authz_cache(Zone), cleanup_authz_cache(),
map_authz_cache(fun(Cache) -> Cache end). map_authz_cache(fun(Cache) -> Cache end).
%% We'll cleanup the cache before replacing an expired authz. %% We'll cleanup the cache before replacing an expired authz.
-spec get_authz_cache(atom(), emqx_types:pubsub(), emqx_topic:topic()) -> -spec get_authz_cache(emqx_types:pubsub(), emqx_topic:topic()) ->
authz_result() | not_found. authz_result() | not_found.
get_authz_cache(Zone, PubSub, Topic) -> get_authz_cache(PubSub, Topic) ->
case erlang:get(cache_k(PubSub, Topic)) of case erlang:get(cache_k(PubSub, Topic)) of
undefined -> not_found; undefined -> not_found;
{AuthzResult, CachedAt} -> {AuthzResult, CachedAt} ->
if_expired(get_cache_ttl(Zone), CachedAt, if_expired(get_cache_ttl(), CachedAt,
fun(false) -> fun(false) ->
AuthzResult; AuthzResult;
(true) -> (true) ->
cleanup_authz_cache(Zone), cleanup_authz_cache(),
not_found not_found
end) end)
end. end.
%% If the cache get full, and also the latest one %% If the cache get full, and also the latest one
%% is expired, then delete all the cache entries %% is expired, then delete all the cache entries
-spec put_authz_cache(atom(), emqx_types:pubsub(), emqx_topic:topic(), authz_result()) -spec put_authz_cache(emqx_types:pubsub(), emqx_topic:topic(), authz_result())
-> ok. -> ok.
put_authz_cache(Zone, PubSub, Topic, AuthzResult) -> put_authz_cache(PubSub, Topic, AuthzResult) ->
MaxSize = get_cache_max_size(Zone), true = (MaxSize =/= 0), MaxSize = get_cache_max_size(), true = (MaxSize =/= 0),
Size = get_cache_size(), Size = get_cache_size(),
case Size < MaxSize of case Size < MaxSize of
true -> true ->
@ -96,7 +96,7 @@ put_authz_cache(Zone, PubSub, Topic, AuthzResult) ->
false -> false ->
NewestK = get_newest_key(), NewestK = get_newest_key(),
{_AuthzResult, CachedAt} = erlang:get(NewestK), {_AuthzResult, CachedAt} = erlang:get(NewestK),
if_expired(get_cache_ttl(Zone), CachedAt, if_expired(get_cache_ttl(), CachedAt,
fun(true) -> fun(true) ->
% all cache expired, cleanup first % all cache expired, cleanup first
empty_authz_cache(), empty_authz_cache(),
@ -123,10 +123,10 @@ evict_authz_cache() ->
decr_cache_size(). decr_cache_size().
%% cleanup all the expired cache entries %% cleanup all the expired cache entries
-spec(cleanup_authz_cache(atom()) -> ok). -spec(cleanup_authz_cache() -> ok).
cleanup_authz_cache(Zone) -> cleanup_authz_cache() ->
keys_queue_set( keys_queue_set(
cleanup_authz(get_cache_ttl(Zone), keys_queue_get())). cleanup_authz(get_cache_ttl(), keys_queue_get())).
get_oldest_key() -> get_oldest_key() ->
keys_queue_pick(queue_front()). keys_queue_pick(queue_front()).
@ -143,8 +143,8 @@ dump_authz_cache() ->
map_authz_cache(fun(Cache) -> Cache end). map_authz_cache(fun(Cache) -> Cache end).
map_authz_cache(Fun) -> map_authz_cache(Fun) ->
[Fun(R) || R = {{SubPub, _T}, _Authz} <- get(), SubPub =:= publish [Fun(R) || R = {{SubPub, _T}, _Authz} <- erlang:get(),
orelse SubPub =:= subscribe]. SubPub =:= publish orelse SubPub =:= subscribe].
foreach_authz_cache(Fun) -> foreach_authz_cache(Fun) ->
_ = map_authz_cache(Fun), _ = map_authz_cache(Fun),
ok. ok.

25
apps/emqx/src/emqx_channel.erl
View File

@ -425,7 +425,7 @@ handle_in(?PUBCOMP_PACKET(PacketId, _ReasonCode), Channel = #channel{session = S
end; end;
handle_in(Packet = ?SUBSCRIBE_PACKET(PacketId, Properties, TopicFilters), handle_in(Packet = ?SUBSCRIBE_PACKET(PacketId, Properties, TopicFilters),
Channel = #channel{clientinfo = ClientInfo = #{zone := Zone}}) -> Channel = #channel{clientinfo = ClientInfo}) ->
case emqx_packet:check(Packet) of case emqx_packet:check(Packet) of
ok -> ok ->
TopicFilters0 = parse_topic_filters(TopicFilters), TopicFilters0 = parse_topic_filters(TopicFilters),
@ -434,7 +434,7 @@ handle_in(Packet = ?SUBSCRIBE_PACKET(PacketId, Properties, TopicFilters),
HasAuthzDeny = lists:any(fun({_TopicFilter, ReasonCode}) -> HasAuthzDeny = lists:any(fun({_TopicFilter, ReasonCode}) ->
ReasonCode =:= ?RC_NOT_AUTHORIZED ReasonCode =:= ?RC_NOT_AUTHORIZED
end, TupleTopicFilters0), end, TupleTopicFilters0),
DenyAction = emqx_config:get_zone_conf(Zone, [authorization, deny_action]), DenyAction = emqx:get_config([authorization, deny_action], ignore),
case DenyAction =:= disconnect andalso HasAuthzDeny of case DenyAction =:= disconnect andalso HasAuthzDeny of
true -> handle_out(disconnect, ?RC_NOT_AUTHORIZED, Channel); true -> handle_out(disconnect, ?RC_NOT_AUTHORIZED, Channel);
false -> false ->
@ -536,8 +536,7 @@ process_connect(AckProps, Channel = #channel{conninfo = ConnInfo,
%% Process Publish %% Process Publish
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
process_publish(Packet = ?PUBLISH_PACKET(QoS, Topic, PacketId), process_publish(Packet = ?PUBLISH_PACKET(QoS, Topic, PacketId), Channel) ->
Channel = #channel{clientinfo = #{zone := Zone}}) ->
case pipeline([fun check_quota_exceeded/2, case pipeline([fun check_quota_exceeded/2,
fun process_alias/2, fun process_alias/2,
fun check_pub_alias/2, fun check_pub_alias/2,
@ -550,7 +549,7 @@ process_publish(Packet = ?PUBLISH_PACKET(QoS, Topic, PacketId),
{error, Rc = ?RC_NOT_AUTHORIZED, NChannel} -> {error, Rc = ?RC_NOT_AUTHORIZED, NChannel} ->
?LOG(warning, "Cannot publish message to ~s due to ~s.", ?LOG(warning, "Cannot publish message to ~s due to ~s.",
[Topic, emqx_reason_codes:text(Rc)]), [Topic, emqx_reason_codes:text(Rc)]),
case emqx_config:get_zone_conf(Zone, [authorization, deny_action]) of case emqx:get_config([authorization, deny_action], ignore) of
ignore -> ignore ->
case QoS of case QoS of
?QOS_0 -> {ok, NChannel}; ?QOS_0 -> {ok, NChannel};
@ -955,9 +954,8 @@ handle_call({takeover, 'end'}, Channel = #channel{session = Session,
AllPendings = lists:append(Delivers, Pendings), AllPendings = lists:append(Delivers, Pendings),
disconnect_and_shutdown(takeovered, AllPendings, Channel); disconnect_and_shutdown(takeovered, AllPendings, Channel);
handle_call(list_authz_cache, #channel{clientinfo = #{zone := Zone}} handle_call(list_authz_cache, Channel) ->
= Channel) -> {reply, emqx_authz_cache:list_authz_cache(), Channel};
{reply, emqx_authz_cache:list_authz_cache(Zone), Channel};
handle_call({quota, Policy}, Channel) -> handle_call({quota, Policy}, Channel) ->
Zone = info(zone, Channel), Zone = info(zone, Channel),
@ -1420,8 +1418,7 @@ check_pub_alias(_Packet, _Channel) -> ok.
check_pub_authz(#mqtt_packet{variable = #mqtt_packet_publish{topic_name = Topic}}, check_pub_authz(#mqtt_packet{variable = #mqtt_packet_publish{topic_name = Topic}},
#channel{clientinfo = ClientInfo}) -> #channel{clientinfo = ClientInfo}) ->
case is_authz_enabled(ClientInfo) andalso case emqx_access_control:authorize(ClientInfo, publish, Topic) of
emqx_access_control:authorize(ClientInfo, publish, Topic) of
false -> ok; false -> ok;
allow -> ok; allow -> ok;
deny -> {error, ?RC_NOT_AUTHORIZED} deny -> {error, ?RC_NOT_AUTHORIZED}
@ -1454,8 +1451,7 @@ check_sub_authzs([], _Channel, Acc) ->
lists:reverse(Acc). lists:reverse(Acc).
check_sub_authz(TopicFilter, #channel{clientinfo = ClientInfo}) -> check_sub_authz(TopicFilter, #channel{clientinfo = ClientInfo}) ->
case is_authz_enabled(ClientInfo) andalso case emqx_access_control:authorize(ClientInfo, subscribe, TopicFilter) of
emqx_access_control:authorize(ClientInfo, subscribe, TopicFilter) of
false -> allow; false -> allow;
Result -> Result Result -> Result
end. end.
@ -1621,11 +1617,6 @@ maybe_shutdown(Reason, Channel = #channel{conninfo = ConnInfo}) ->
_ -> shutdown(Reason, Channel) _ -> shutdown(Reason, Channel)
end. end.
%%--------------------------------------------------------------------
%% Is Authorization enabled?
is_authz_enabled(#{zone := Zone, is_superuser := IsSuperuser}) ->
(not IsSuperuser) andalso emqx_config:get_zone_conf(Zone, [authorization, enable]).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Parse Topic Filters %% Parse Topic Filters

9
apps/emqx/src/emqx_schema.erl
View File

@ -70,7 +70,7 @@
-export([conf_get/2, conf_get/3, keys/2, filter/1]). -export([conf_get/2, conf_get/3, keys/2, filter/1]).
-export([ssl/1]). -export([ssl/1]).
structs() -> ["zones", "listeners", "broker", "plugins", "sysmon", "alarm"]. structs() -> ["zones", "listeners", "broker", "plugins", "sysmon", "alarm", "authorization"].
fields("stats") -> fields("stats") ->
[ {"enable", t(boolean(), undefined, true)} [ {"enable", t(boolean(), undefined, true)}
@ -80,10 +80,10 @@ fields("auth") ->
[ {"enable", t(boolean(), undefined, false)} [ {"enable", t(boolean(), undefined, false)}
]; ];
fields("authorization_settings") -> fields("authorization") ->
[ {"enable", t(boolean(), undefined, true)} [ {"no_match", t(union(allow, deny), undefined, allow)}
, {"cache", ref("authorization_cache")}
, {"deny_action", t(union(ignore, disconnect), undefined, ignore)} , {"deny_action", t(union(ignore, disconnect), undefined, ignore)}
, {"cache", ref("authorization_cache")}
]; ];
fields("authorization_cache") -> fields("authorization_cache") ->
@ -129,7 +129,6 @@ fields("zones") ->
fields("zone_settings") -> fields("zone_settings") ->
[ {"mqtt", ref("mqtt")} [ {"mqtt", ref("mqtt")}
, {"authorization", ref("authorization_settings")}
, {"auth", ref("auth")} , {"auth", ref("auth")}
, {"stats", ref("stats")} , {"stats", ref("stats")}
, {"flapping_detect", ref("flapping_detect")} , {"flapping_detect", ref("flapping_detect")}

4
apps/emqx/test/emqx_authz_cache_SUITE.erl
View File

@ -26,7 +26,6 @@ all() -> emqx_ct:all(?MODULE).
init_per_suite(Config) -> init_per_suite(Config) ->
emqx_ct_helpers:boot_modules(all), emqx_ct_helpers:boot_modules(all),
emqx_ct_helpers:start_apps([]), emqx_ct_helpers:start_apps([]),
toggle_authz(true),
Config. Config.
end_per_suite(_Config) -> end_per_suite(_Config) ->
@ -78,6 +77,3 @@ t_drain_authz_cache(_) ->
{ok, _, _} = emqtt:subscribe(Client, <<"t2">>, 0), {ok, _, _} = emqtt:subscribe(Client, <<"t2">>, 0),
?assert(length(gen_server:call(ClientPid, list_authz_cache)) > 0), ?assert(length(gen_server:call(ClientPid, list_authz_cache)) > 0),
emqtt:stop(Client). emqtt:stop(Client).
toggle_authz(Bool) when is_boolean(Bool) ->
emqx_config:put_zone_conf(default, [authorization, enable], Bool).