feat(authz): support authorization config file part 2
This commit is contained in:
zhanghongtong
Rory Z
parent
a7fac1a7a3
commit
a2bafd1a18
|
|
@ -68,11 +68,6 @@ authorization_rules {
|
|||
# }
|
||||
# collection: mqtt_authz
|
||||
# find: { "$or": [ { "username": "%u" }, { "clientid": "%c" } ] }
|
||||
# },
|
||||
{
|
||||
permission = allow
|
||||
action = all
|
||||
topics = ["#"]
|
||||
}
|
||||
# }
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,8 +19,13 @@
|
|||
|
||||
-define(APP, emqx_authz).
|
||||
|
||||
-define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= deny))).
|
||||
-define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= publish) orelse (A =:= all))).
|
||||
-define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= <<"allow">>) orelse
|
||||
(A =:= deny) orelse (A =:= <<"deny">>)
|
||||
)).
|
||||
-define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= <<"subscribe">>) orelse
|
||||
(A =:= publish) orelse (A =:= <<"publish">>) orelse
|
||||
(A =:= all) orelse (A =:= <<"all">>)
|
||||
)).
|
||||
|
||||
-record(authz_metrics, {
|
||||
allow = 'client.authorize.allow',
|
||||
|
|
|
|||
|
|
@ -27,13 +27,11 @@
|
|||
|
||||
-export([ register_metrics/0
|
||||
, init/0
|
||||
, init_rule/1
|
||||
, lookup/0
|
||||
, lookup/1
|
||||
, move/2
|
||||
, update/2
|
||||
, authorize/5
|
||||
, match/4
|
||||
]).
|
||||
|
||||
-export([post_config_update/3, pre_config_update/2]).
|
||||
|
|
@ -47,7 +45,7 @@ register_metrics() ->
|
|||
init() ->
|
||||
ok = register_metrics(),
|
||||
emqx_config_handler:add_handler(?CONF_KEY_PATH, ?MODULE),
|
||||
NRules = [init_rule(Rule) || Rule <- emqx:get_config(?CONF_KEY_PATH, [])],
|
||||
NRules = [init_provider(Rule) || Rule <- emqx:get_config(?CONF_KEY_PATH, [])],
|
||||
ok = emqx_hooks:add('client.authorize', {?MODULE, authorize, [NRules]}, -1).
|
||||
|
||||
lookup() ->
|
||||
|
|
@ -148,12 +146,12 @@ post_config_update({move, Id, #{<<"after">> := AfterId}}, _NewRules, _OldRules)
|
|||
ok = emqx_authz_cache:drain_cache();
|
||||
|
||||
post_config_update({head, Rules}, _NewRules, _OldConf) ->
|
||||
InitedRules = [init_rule(R) || R <- check_rules(Rules)],
|
||||
InitedRules = [init_provider(R) || R <- check_rules(Rules)],
|
||||
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [InitedRules ++ lookup()]}, -1),
|
||||
ok = emqx_authz_cache:drain_cache();
|
||||
|
||||
post_config_update({tail, Rules}, _NewRules, _OldConf) ->
|
||||
InitedRules = [init_rule(R) || R <- check_rules(Rules)],
|
||||
InitedRules = [init_provider(R) || R <- check_rules(Rules)],
|
||||
emqx_hooks:put('client.authorize', {?MODULE, authorize, [lookup() ++ InitedRules]}, -1),
|
||||
ok = emqx_authz_cache:drain_cache();
|
||||
|
||||
|
|
@ -167,14 +165,14 @@ post_config_update({{replace_once, Id}, Rule}, _NewRules, _OldConf) when is_map(
|
|||
ok = emqx_resource:remove(Id)
|
||||
end,
|
||||
{OldRules1, OldRules2 } = lists:split(Index, OldInitedRules),
|
||||
InitedRules = [init_rule(R#{annotations => #{id => Id}}) || R <- check_rules([Rule])],
|
||||
InitedRules = [init_provider(R#{annotations => #{id => Id}}) || R <- check_rules([Rule])],
|
||||
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [lists:droplast(OldRules1) ++ InitedRules ++ OldRules2]}, -1),
|
||||
ok = emqx_authz_cache:drain_cache();
|
||||
|
||||
post_config_update(_, NewRules, _OldConf) ->
|
||||
%% overwrite the entire config!
|
||||
OldInitedRules = lookup(),
|
||||
InitedRules = [init_rule(Rule) || Rule <- NewRules],
|
||||
InitedRules = [init_provider(Rule) || Rule <- NewRules],
|
||||
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [InitedRules]}, -1),
|
||||
lists:foreach(fun (#{type := _Type, enable := true, annotations := #{id := Id}}) ->
|
||||
ok = emqx_resource:remove(Id);
|
||||
|
|
@ -235,26 +233,8 @@ create_resource(#{type := DB,
|
|||
{error, Reason} -> {error, Reason}
|
||||
end.
|
||||
|
||||
-spec(init_rule(rule()) -> rule()).
|
||||
init_rule(#{topics := Topics,
|
||||
action := Action,
|
||||
permission := Permission,
|
||||
principal := Principal,
|
||||
annotations := #{id := Id}
|
||||
} = Rule) when ?ALLOW_DENY(Permission), ?PUBSUB(Action), is_list(Topics) ->
|
||||
Rule#{annotations =>
|
||||
#{id => Id,
|
||||
principal => compile_principal(Principal),
|
||||
topics => [compile_topic(Topic) || Topic <- Topics]}
|
||||
};
|
||||
init_rule(#{topics := Topics,
|
||||
action := Action,
|
||||
permission := Permission
|
||||
} = Rule) when ?ALLOW_DENY(Permission), ?PUBSUB(Action), is_list(Topics) ->
|
||||
init_rule(Rule#{annotations =>#{id => gen_id(simple)}});
|
||||
|
||||
init_rule(#{principal := Principal,
|
||||
enable := true,
|
||||
-spec(init_provider(rule()) -> rule()).
|
||||
init_provider(#{enable := true,
|
||||
type := file,
|
||||
path := Path
|
||||
} = Rule) ->
|
||||
|
|
@ -275,8 +255,7 @@ init_rule(#{principal := Principal,
|
|||
#{id => gen_id(file),
|
||||
rules => Rules
|
||||
}};
|
||||
init_rule(#{principal := Principal,
|
||||
enable := true,
|
||||
init_provider(#{enable := true,
|
||||
type := http,
|
||||
config := #{url := Url} = Config
|
||||
} = Rule) ->
|
||||
|
|
@ -284,28 +263,20 @@ init_rule(#{principal := Principal,
|
|||
case create_resource(Rule#{config := NConfig}) of
|
||||
{error, Reason} -> error({load_config_error, Reason});
|
||||
Id -> Rule#{annotations =>
|
||||
#{id => Id,
|
||||
principal => compile_principal(Principal)
|
||||
}
|
||||
#{id => Id}
|
||||
}
|
||||
end;
|
||||
|
||||
init_rule(#{principal := Principal,
|
||||
enable := true,
|
||||
init_provider(#{enable := true,
|
||||
type := DB
|
||||
} = Rule) when DB =:= redis;
|
||||
DB =:= mongo ->
|
||||
case create_resource(Rule) of
|
||||
{error, Reason} -> error({load_config_error, Reason});
|
||||
Id -> Rule#{annotations =>
|
||||
#{id => Id,
|
||||
principal => compile_principal(Principal)
|
||||
}
|
||||
#{id => Id}
|
||||
}
|
||||
end;
|
||||
|
||||
init_rule(#{principal := Principal,
|
||||
enable := true,
|
||||
init_provider(#{enable := true,
|
||||
type := DB,
|
||||
sql := SQL
|
||||
} = Rule) when DB =:= mysql;
|
||||
|
|
@ -315,52 +286,11 @@ init_rule(#{principal := Principal,
|
|||
{error, Reason} -> error({load_config_error, Reason});
|
||||
Id -> Rule#{annotations =>
|
||||
#{id => Id,
|
||||
principal => compile_principal(Principal),
|
||||
sql => Mod:parse_query(SQL)
|
||||
}
|
||||
}
|
||||
end;
|
||||
|
||||
init_rule(#{enable := false,
|
||||
type := _DB
|
||||
} = Rule) ->
|
||||
Rule.
|
||||
|
||||
compile_principal(all) -> all;
|
||||
compile_principal(#{username := Username}) ->
|
||||
{ok, MP} = re:compile(bin(Username)),
|
||||
#{username => MP};
|
||||
compile_principal(#{clientid := Clientid}) ->
|
||||
{ok, MP} = re:compile(bin(Clientid)),
|
||||
#{clientid => MP};
|
||||
compile_principal(#{ipaddress := IpAddress}) ->
|
||||
#{ipaddress => esockd_cidr:parse(b2l(IpAddress), true)};
|
||||
compile_principal(#{'and' := Principals}) when is_list(Principals) ->
|
||||
#{'and' => [compile_principal(Principal) || Principal <- Principals]};
|
||||
compile_principal(#{'or' := Principals}) when is_list(Principals) ->
|
||||
#{'or' => [compile_principal(Principal) || Principal <- Principals]}.
|
||||
|
||||
compile_topic(<<"eq ", Topic/binary>>) ->
|
||||
compile_topic(#{'eq' => Topic});
|
||||
compile_topic(#{'eq' := Topic}) ->
|
||||
#{'eq' => emqx_topic:words(bin(Topic))};
|
||||
compile_topic(Topic) when is_binary(Topic)->
|
||||
Words = emqx_topic:words(bin(Topic)),
|
||||
case pattern(Words) of
|
||||
true -> #{pattern => Words};
|
||||
false -> Words
|
||||
end.
|
||||
|
||||
pattern(Words) ->
|
||||
lists:member(<<"%u">>, Words) orelse lists:member(<<"%c">>, Words).
|
||||
|
||||
bin(A) when is_atom(A) -> atom_to_binary(A, utf8);
|
||||
bin(B) when is_binary(B) -> B;
|
||||
bin(L) when is_list(L) -> list_to_binary(L);
|
||||
bin(X) -> X.
|
||||
|
||||
b2l(B) when is_list(B) -> B;
|
||||
b2l(B) when is_binary(B) -> binary_to_list(B).
|
||||
init_provider(#{enable := false} = Rule) ->Rule.
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
%% AuthZ callbacks
|
||||
|
|
@ -387,97 +317,21 @@ authorize(#{username := Username,
|
|||
end.
|
||||
|
||||
do_authorize(Client, PubSub, Topic,
|
||||
[Connector = #{type := DB,
|
||||
[#{type := file,
|
||||
enable := true,
|
||||
annotations := #{principal := Principal}
|
||||
annotations := #{rule := Rules}
|
||||
} | Tail] ) ->
|
||||
case match_principal(Client, Principal) of
|
||||
true ->
|
||||
Mod = list_to_existing_atom(io_lib:format("~s_~s",[emqx_authz, DB])),
|
||||
case emqx_authz_rule:match(Client, PubSub, Topic, Rules) of
|
||||
nomatch -> do_authorize(Client, PubSub, Topic, Tail);
|
||||
Matched -> Matched
|
||||
end;
|
||||
do_authorize(Client, PubSub, Topic,
|
||||
[Connector = #{type := Type,
|
||||
enable := true
|
||||
} | Tail] ) ->
|
||||
Mod = list_to_existing_atom(io_lib:format("~s_~s",[emqx_authz, Type])),
|
||||
case Mod:authorize(Client, PubSub, Topic, Connector) of
|
||||
nomatch -> do_authorize(Client, PubSub, Topic, Tail);
|
||||
Matched -> Matched
|
||||
end;
|
||||
false -> do_authorize(Client, PubSub, Topic, Tail)
|
||||
end;
|
||||
do_authorize(Client, PubSub, Topic,
|
||||
[#{permission := Permission} = Rule | Tail]) ->
|
||||
case match(Client, PubSub, Topic, Rule) of
|
||||
true -> {matched, Permission};
|
||||
false -> do_authorize(Client, PubSub, Topic, Tail)
|
||||
end;
|
||||
do_authorize(_Client, _PubSub, _Topic, []) -> nomatch.
|
||||
|
||||
match(Client, PubSub, Topic,
|
||||
#{action := Action,
|
||||
annotations := #{
|
||||
principal := Principal,
|
||||
topics := TopicFilters
|
||||
}
|
||||
}) ->
|
||||
match_action(PubSub, Action) andalso
|
||||
match_principal(Client, Principal) andalso
|
||||
match_topics(Client, Topic, TopicFilters).
|
||||
|
||||
match_action(publish, publish) -> true;
|
||||
match_action(subscribe, subscribe) -> true;
|
||||
match_action(_, all) -> true;
|
||||
match_action(_, _) -> false.
|
||||
|
||||
match_principal(_, all) -> true;
|
||||
match_principal(#{username := undefined}, #{username := _MP}) ->
|
||||
false;
|
||||
match_principal(#{username := Username}, #{username := MP}) ->
|
||||
case re:run(Username, MP) of
|
||||
{match, _} -> true;
|
||||
_ -> false
|
||||
end;
|
||||
match_principal(#{clientid := Clientid}, #{clientid := MP}) ->
|
||||
case re:run(Clientid, MP) of
|
||||
{match, _} -> true;
|
||||
_ -> false
|
||||
end;
|
||||
match_principal(#{peerhost := undefined}, #{ipaddress := _CIDR}) ->
|
||||
false;
|
||||
match_principal(#{peerhost := IpAddress}, #{ipaddress := CIDR}) ->
|
||||
esockd_cidr:match(IpAddress, CIDR);
|
||||
match_principal(ClientInfo, #{'and' := Principals}) when is_list(Principals) ->
|
||||
lists:foldl(fun(Principal, Permission) ->
|
||||
match_principal(ClientInfo, Principal) andalso Permission
|
||||
end, true, Principals);
|
||||
match_principal(ClientInfo, #{'or' := Principals}) when is_list(Principals) ->
|
||||
lists:foldl(fun(Principal, Permission) ->
|
||||
match_principal(ClientInfo, Principal) orelse Permission
|
||||
end, false, Principals);
|
||||
match_principal(_, _) -> false.
|
||||
|
||||
match_topics(_ClientInfo, _Topic, []) ->
|
||||
false;
|
||||
match_topics(ClientInfo, Topic, [#{pattern := PatternFilter}|Filters]) ->
|
||||
TopicFilter = feed_var(ClientInfo, PatternFilter),
|
||||
match_topic(emqx_topic:words(Topic), TopicFilter)
|
||||
orelse match_topics(ClientInfo, Topic, Filters);
|
||||
match_topics(ClientInfo, Topic, [TopicFilter|Filters]) ->
|
||||
match_topic(emqx_topic:words(Topic), TopicFilter)
|
||||
orelse match_topics(ClientInfo, Topic, Filters).
|
||||
|
||||
match_topic(Topic, #{'eq' := TopicFilter}) ->
|
||||
Topic == TopicFilter;
|
||||
match_topic(Topic, TopicFilter) ->
|
||||
emqx_topic:match(Topic, TopicFilter).
|
||||
|
||||
feed_var(ClientInfo, Pattern) ->
|
||||
feed_var(ClientInfo, Pattern, []).
|
||||
feed_var(_ClientInfo, [], Acc) ->
|
||||
lists:reverse(Acc);
|
||||
feed_var(ClientInfo = #{clientid := undefined}, [<<"%c">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [<<"%c">>|Acc]);
|
||||
feed_var(ClientInfo = #{clientid := ClientId}, [<<"%c">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [ClientId |Acc]);
|
||||
feed_var(ClientInfo = #{username := undefined}, [<<"%u">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [<<"%u">>|Acc]);
|
||||
feed_var(ClientInfo = #{username := Username}, [<<"%u">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [Username|Acc]);
|
||||
feed_var(ClientInfo, [W|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [W|Acc]).
|
||||
|
||||
|
|
|
|||
|
|
@ -419,12 +419,26 @@ move_rule_api() ->
|
|||
{"/authorization/:id/move", Metadata, move_rule}.
|
||||
|
||||
rules(get, #{query_string := Query}) ->
|
||||
Rules = lists:foldl(fun (#{type := _Type, enable := true, annotations := #{id := Id} = Annotations} = Rule, AccIn) ->
|
||||
Rules = lists:foldl(fun (#{type := _Type, enable := true, config := #{server := Server} = Config, annotations := #{id := Id}} = Rule, AccIn) ->
|
||||
NRule = case emqx_resource:health_check(Id) of
|
||||
ok ->
|
||||
Rule#{annotations => Annotations#{status => healthy}};
|
||||
Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
|
||||
annotations => #{id => Id,
|
||||
status => healthy}};
|
||||
_ ->
|
||||
Rule#{annotations => Annotations#{status => unhealthy}}
|
||||
Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
|
||||
annotations => #{id => Id,
|
||||
status => unhealthy}}
|
||||
end,
|
||||
lists:append(AccIn, [NRule]);
|
||||
(#{type := _Type, enable := true, annotations := #{id := Id}} = Rule, AccIn) ->
|
||||
NRule = case emqx_resource:health_check(Id) of
|
||||
ok ->
|
||||
Rule#{annotations => #{id => Id,
|
||||
status => healthy}};
|
||||
_ ->
|
||||
Rule#{annotations => #{id => Id,
|
||||
status => unhealthy}}
|
||||
end,
|
||||
lists:append(AccIn, [NRule]);
|
||||
(Rule, AccIn) ->
|
||||
|
|
@ -462,17 +476,26 @@ rules(put, #{body := RawConfig}) ->
|
|||
rule(get, #{bindings := #{id := Id}}) ->
|
||||
case emqx_authz:lookup(Id) of
|
||||
{error, Reason} -> {404, #{messgae => atom_to_binary(Reason)}};
|
||||
Rule ->
|
||||
case maps:get(type, Rule, undefined) of
|
||||
undefined -> {200, Rule};
|
||||
_ ->
|
||||
#{type := file} = Rule -> {200, Rule};
|
||||
#{config := #{server := Server} = Config} = Rule ->
|
||||
case emqx_resource:health_check(Id) of
|
||||
ok ->
|
||||
{200, Rule#{annotations => #{status => healthy}}};
|
||||
{200, Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
|
||||
annotations => #{id => Id,
|
||||
status => healthy}}};
|
||||
_ ->
|
||||
{200, Rule#{annotations => #{status => unhealthy}}}
|
||||
end
|
||||
|
||||
{200, Rule#{config => Config#{server => emqx_connector_schema_lib:ip_port_to_string(Server)},
|
||||
annotations => #{id => Id,
|
||||
status => unhealthy}}}
|
||||
end;
|
||||
Rule ->
|
||||
case emqx_resource:health_check(Id) of
|
||||
ok ->
|
||||
{200, Rule#{annotations => #{id => Id,
|
||||
status => healthy}}};
|
||||
_ ->
|
||||
{200, Rule#{annotations => #{id => Id,
|
||||
status => unhealthy}}}
|
||||
end
|
||||
end;
|
||||
rule(put, #{bindings := #{id := RuleId}, body := RawConfig}) ->
|
||||
|
|
|
|||
|
|
@ -44,38 +44,19 @@ authorize(Client, PubSub, Topic,
|
|||
nomatch;
|
||||
[] -> nomatch;
|
||||
Rows ->
|
||||
do_authorize(Client, PubSub, Topic, Rows)
|
||||
Rules = [ emqx_authz_rule:compile({Permission, all, Action, Topics})
|
||||
|| #{<<"topics">> := Topics, <<"permission">> := Permission, <<"action">> := Action} <- Rows],
|
||||
do_authorize(Client, PubSub, Topic, Rules)
|
||||
end.
|
||||
|
||||
do_authorize(_Client, _PubSub, _Topic, []) ->
|
||||
nomatch;
|
||||
do_authorize(Client, PubSub, Topic, [Rule | Tail]) ->
|
||||
case match(Client, PubSub, Topic, Rule) of
|
||||
case emqx_authz_rule:match(Client, PubSub, Topic, Rule) of
|
||||
{matched, Permission} -> {matched, Permission};
|
||||
nomatch -> do_authorize(Client, PubSub, Topic, Tail)
|
||||
end.
|
||||
|
||||
match(Client, PubSub, Topic,
|
||||
#{<<"topics">> := Topics,
|
||||
<<"permission">> := Permission,
|
||||
<<"action">> := Action
|
||||
}) ->
|
||||
Rule = #{<<"permission">> => Permission,
|
||||
<<"topics">> => Topics,
|
||||
<<"action">> => Action
|
||||
},
|
||||
#{simple_rule :=
|
||||
#{permission := NPermission} = NRule
|
||||
} = hocon_schema:check_plain(
|
||||
emqx_authz_schema,
|
||||
#{<<"simple_rule">> => Rule},
|
||||
#{atom_key => true},
|
||||
[simple_rule]),
|
||||
case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
|
||||
true -> {matched, NPermission};
|
||||
false -> nomatch
|
||||
end.
|
||||
|
||||
replvar(Find, #{clientid := Clientid,
|
||||
username := Username,
|
||||
peerhost := IpAddress
|
||||
|
|
|
|||
|
|
@ -62,39 +62,25 @@ authorize(Client, PubSub, Topic,
|
|||
do_authorize(_Client, _PubSub, _Topic, _Columns, []) ->
|
||||
nomatch;
|
||||
do_authorize(Client, PubSub, Topic, Columns, [Row | Tail]) ->
|
||||
case match(Client, PubSub, Topic, format_result(Columns, Row)) of
|
||||
case emqx_authz_rule:match(Client, PubSub, Topic,
|
||||
emqx_authz_rule:compile(format_result(Columns, Row))
|
||||
) of
|
||||
{matched, Permission} -> {matched, Permission};
|
||||
nomatch -> do_authorize(Client, PubSub, Topic, Columns, Tail)
|
||||
end.
|
||||
|
||||
format_result(Columns, Row) ->
|
||||
L = [ begin
|
||||
K = lists:nth(I, Columns),
|
||||
V = lists:nth(I, Row),
|
||||
{K, V}
|
||||
end || I <- lists:seq(1, length(Columns)) ],
|
||||
maps:from_list(L).
|
||||
|
||||
match(Client, PubSub, Topic,
|
||||
#{<<"permission">> := Permission,
|
||||
<<"action">> := Action,
|
||||
<<"topic">> := TopicFilter
|
||||
}) ->
|
||||
Rule = #{<<"topics">> => [TopicFilter],
|
||||
<<"action">> => Action,
|
||||
<<"permission">> => Permission
|
||||
},
|
||||
#{simple_rule :=
|
||||
#{permission := NPermission} = NRule
|
||||
} = hocon_schema:check_plain(
|
||||
emqx_authz_schema,
|
||||
#{<<"simple_rule">> => Rule},
|
||||
#{atom_key => true},
|
||||
[simple_rule]),
|
||||
case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
|
||||
true -> {matched, NPermission};
|
||||
false -> nomatch
|
||||
end.
|
||||
format_result(Columns, Row) ->
|
||||
Permission = lists:nth(index(<<"permission">>, Columns), Row),
|
||||
Action = lists:nth(index(<<"action">>, Columns), Row),
|
||||
Topic = lists:nth(index(<<"topic">>, Columns), Row),
|
||||
{Permission, all, Action, [Topic]}.
|
||||
|
||||
index(Elem, List) ->
|
||||
index(Elem, List, 1).
|
||||
index(_Elem, [], _Index) -> {error, not_found};
|
||||
index(Elem, [ Elem | _List], Index) -> Index;
|
||||
index(Elem, [ _ | List], Index) -> index(Elem, List, Index + 1).
|
||||
|
||||
replvar(Params, ClientInfo) ->
|
||||
replvar(Params, ClientInfo, []).
|
||||
|
|
|
|||
|
|
@ -66,39 +66,25 @@ authorize(Client, PubSub, Topic,
|
|||
do_authorize(_Client, _PubSub, _Topic, _Columns, []) ->
|
||||
nomatch;
|
||||
do_authorize(Client, PubSub, Topic, Columns, [Row | Tail]) ->
|
||||
case match(Client, PubSub, Topic, format_result(Columns, Row)) of
|
||||
case emqx_authz_rule:match(Client, PubSub, Topic,
|
||||
emqx_authz_rule:compile(format_result(Columns, Row))
|
||||
) of
|
||||
{matched, Permission} -> {matched, Permission};
|
||||
nomatch -> do_authorize(Client, PubSub, Topic, Columns, Tail)
|
||||
end.
|
||||
|
||||
format_result(Columns, Row) ->
|
||||
L = [ begin
|
||||
{column, K, _, _, _, _, _, _, _} = lists:nth(I, Columns),
|
||||
V = lists:nth(I, tuple_to_list(Row)),
|
||||
{K, V}
|
||||
end || I <- lists:seq(1, length(Columns)) ],
|
||||
maps:from_list(L).
|
||||
Permission = lists:nth(index(<<"permission">>, 2, Columns), erlang:tuple_to_list(Row)),
|
||||
Action = lists:nth(index(<<"action">>, 2, Columns), erlang:tuple_to_list(Row)),
|
||||
Topic = lists:nth(index(<<"topic">>, 2, Columns), erlang:tuple_to_list(Row)),
|
||||
{Permission, all, Action, [Topic]}.
|
||||
|
||||
match(Client, PubSub, Topic,
|
||||
#{<<"permission">> := Permission,
|
||||
<<"action">> := Action,
|
||||
<<"topic">> := TopicFilter
|
||||
}) ->
|
||||
Rule = #{<<"topics">> => [TopicFilter],
|
||||
<<"action">> => Action,
|
||||
<<"permission">> => Permission
|
||||
},
|
||||
#{simple_rule :=
|
||||
#{permission := NPermission} = NRule
|
||||
} = hocon_schema:check_plain(
|
||||
emqx_authz_schema,
|
||||
#{<<"simple_rule">> => Rule},
|
||||
#{atom_key => true},
|
||||
[simple_rule]),
|
||||
case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
|
||||
true -> {matched, NPermission};
|
||||
false -> nomatch
|
||||
end.
|
||||
index(Key, N, TupleList) when is_integer(N) ->
|
||||
Tuple = lists:keyfind(Key, N, TupleList),
|
||||
index(Tuple, TupleList, 1);
|
||||
index(_Tuple, [], _Index) -> {error, not_found};
|
||||
index(Tuple, [Tuple | _TupleList], Index) -> Index;
|
||||
index(Tuple, [_ | TupleList], Index) -> index(Tuple, TupleList, Index + 1).
|
||||
|
||||
replvar(Params, ClientInfo) ->
|
||||
replvar(Params, ClientInfo, []).
|
||||
|
|
|
|||
|
|
@ -50,35 +50,13 @@ authorize(Client, PubSub, Topic,
|
|||
do_authorize(_Client, _PubSub, _Topic, []) ->
|
||||
nomatch;
|
||||
do_authorize(Client, PubSub, Topic, [TopicFilter, Action | Tail]) ->
|
||||
case match(Client, PubSub, Topic,
|
||||
#{topics => TopicFilter,
|
||||
action => Action
|
||||
})
|
||||
of
|
||||
case emqx_authz_rule:match(Client, PubSub, Topic,
|
||||
emqx_authz_rule:compile({allow, all, Action, [TopicFilter]})
|
||||
)of
|
||||
{matched, Permission} -> {matched, Permission};
|
||||
nomatch -> do_authorize(Client, PubSub, Topic, Tail)
|
||||
end.
|
||||
|
||||
match(Client, PubSub, Topic,
|
||||
#{topics := TopicFilter,
|
||||
action := Action
|
||||
}) ->
|
||||
Rule = #{<<"principal">> => all,
|
||||
<<"topics">> => [TopicFilter],
|
||||
<<"action">> => Action,
|
||||
<<"permission">> => allow
|
||||
},
|
||||
#{simple_rule := NRule
|
||||
} = hocon_schema:check_plain(
|
||||
emqx_authz_schema,
|
||||
#{<<"simple_rule">> => Rule},
|
||||
#{atom_key => true},
|
||||
[simple_rule]),
|
||||
case emqx_authz:match(Client, PubSub, Topic, emqx_authz:init_rule(NRule)) of
|
||||
true -> {matched, allow};
|
||||
false -> nomatch
|
||||
end.
|
||||
|
||||
replvar(Cmd, Client = #{cn := CN}) ->
|
||||
replvar(repl(Cmd, "%C", CN), maps:remove(cn, Client));
|
||||
replvar(Cmd, Client = #{dn := DN}) ->
|
||||
|
|
|
|||
|
|
@ -26,15 +26,14 @@
|
|||
|
||||
%% APIs
|
||||
-export([ match/4
|
||||
, matches/4
|
||||
, compile/1
|
||||
]).
|
||||
|
||||
-export_type([rule/0]).
|
||||
|
||||
compile({Permission, Who, Action, TopicFilters}) when ?ALLOW_DENY(Permission), ?PUBSUB(Action), is_list(TopicFilters) ->
|
||||
{Permission, compile_who(Who), Action, [compile_topic(Topic) || Topic <- TopicFilters]};
|
||||
compile({Permission, Who, Action, Topic}) when ?ALLOW_DENY(Permission), ?PUBSUB(Action) ->
|
||||
{Permission, compile_who(Who), Action, [compile_topic(Topic)]}.
|
||||
{atom(Permission), compile_who(Who), atom(Action), [compile_topic(Topic) || Topic <- TopicFilters]}.
|
||||
|
||||
compile_who(all) -> all;
|
||||
compile_who({username, Username}) ->
|
||||
|
|
@ -52,6 +51,8 @@ compile_who({'and', L}) when is_list(L) ->
|
|||
compile_who({'or', L}) when is_list(L) ->
|
||||
{'or', [compile_who(Who) || Who <- L]}.
|
||||
|
||||
compile_topic(<<"eq ", Topic/binary>>) ->
|
||||
{eq, emqx_topic:words(Topic)};
|
||||
compile_topic({eq, Topic}) ->
|
||||
{eq, emqx_topic:words(bin(Topic))};
|
||||
compile_topic(Topic) ->
|
||||
|
|
@ -64,11 +65,32 @@ compile_topic(Topic) ->
|
|||
pattern(Words) ->
|
||||
lists:member(<<"%u">>, Words) orelse lists:member(<<"%c">>, Words).
|
||||
|
||||
atom(B) when is_binary(B) ->
|
||||
try binary_to_existing_atom(B, utf8)
|
||||
catch
|
||||
_ -> binary_to_atom(B)
|
||||
end;
|
||||
atom(L) when is_list(L) ->
|
||||
try list_to_existing_atom(L)
|
||||
catch
|
||||
_ -> list_to_atom(L)
|
||||
end;
|
||||
atom(A) when is_atom(A) -> A.
|
||||
|
||||
bin(L) when is_list(L) ->
|
||||
list_to_binary(L);
|
||||
bin(B) when is_binary(B) ->
|
||||
B.
|
||||
|
||||
-spec(matches(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic(), [rule()])
|
||||
-> {matched, allow} | {matched, deny} | nomatch).
|
||||
matches(Client, PubSub, Topic, []) -> nomatch;
|
||||
matches(Client, PubSub, Topic, [{Permission, Who, Action, TopicFilters} | Tail]) ->
|
||||
case match(Client, PubSub, Topic, {Permission, Who, Action, TopicFilters}) of
|
||||
nomatch -> matches(Client, PubSub, Topic, Tail);
|
||||
Matched -> Matched
|
||||
end.
|
||||
|
||||
-spec(match(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic(), rule())
|
||||
-> {matched, allow} | {matched, deny} | nomatch).
|
||||
match(Client, PubSub, Topic, {Permission, Who, Action, TopicFilters}) ->
|
||||
|
|
|
|||
|
|
@ -23,8 +23,7 @@ fields("authorization_rules") ->
|
|||
[ {rules, rules()}
|
||||
];
|
||||
fields(file) ->
|
||||
[ {principal, principal()}
|
||||
, {type, #{type => http}}
|
||||
[ {type, #{type => http}}
|
||||
, {enable, #{type => boolean(),
|
||||
default => true}}
|
||||
, {path, #{type => string(),
|
||||
|
|
@ -36,8 +35,7 @@ fields(file) ->
|
|||
}}
|
||||
];
|
||||
fields(http) ->
|
||||
[ {principal, principal()}
|
||||
, {type, #{type => http}}
|
||||
[ {type, #{type => http}}
|
||||
, {enable, #{type => boolean(),
|
||||
default => true}}
|
||||
, {config, #{type => hoconsc:union([ hoconsc:ref(?MODULE, http_get)
|
||||
|
|
@ -113,16 +111,6 @@ fields(mysql) ->
|
|||
fields(pgsql) ->
|
||||
connector_fields(pgsql) ++
|
||||
[ {sql, query()} ];
|
||||
fields(simple_rule) ->
|
||||
[ {permission, #{type => permission()}}
|
||||
, {action, #{type => action()}}
|
||||
, {topics, #{type => union_array(
|
||||
[ binary()
|
||||
, hoconsc:ref(?MODULE, eq_topic)
|
||||
]
|
||||
)}}
|
||||
, {principal, principal()}
|
||||
];
|
||||
fields(username) ->
|
||||
[{username, #{type => binary()}}];
|
||||
fields(clientid) ->
|
||||
|
|
@ -160,8 +148,7 @@ union_array(Item) when is_list(Item) ->
|
|||
|
||||
rules() ->
|
||||
#{type => union_array(
|
||||
[ hoconsc:ref(?MODULE, simple_rule)
|
||||
, hoconsc:ref(?MODULE, file)
|
||||
[ hoconsc:ref(?MODULE, file)
|
||||
, hoconsc:ref(?MODULE, http)
|
||||
, hoconsc:ref(?MODULE, mysql)
|
||||
, hoconsc:ref(?MODULE, pgsql)
|
||||
|
|
@ -170,18 +157,6 @@ rules() ->
|
|||
])
|
||||
}.
|
||||
|
||||
principal() ->
|
||||
#{default => all,
|
||||
type => hoconsc:union(
|
||||
[ all
|
||||
, hoconsc:ref(?MODULE, username)
|
||||
, hoconsc:ref(?MODULE, clientid)
|
||||
, hoconsc:ref(?MODULE, ipaddress)
|
||||
, hoconsc:ref(?MODULE, andlist)
|
||||
, hoconsc:ref(?MODULE, orlist)
|
||||
])
|
||||
}.
|
||||
|
||||
query() ->
|
||||
#{type => binary(),
|
||||
validator => fun(S) ->
|
||||
|
|
@ -202,8 +177,7 @@ connector_fields(DB) ->
|
|||
Error ->
|
||||
erlang:error(Error)
|
||||
end,
|
||||
[ {principal, principal()}
|
||||
, {type, #{type => DB}}
|
||||
[ {type, #{type => DB}}
|
||||
, {enable, #{type => boolean(),
|
||||
default => true}}
|
||||
] ++ Mod:fields("").
|
||||
|
|
|
|||
|
|
@ -31,6 +31,10 @@ groups() ->
|
|||
[].
|
||||
|
||||
init_per_suite(Config) ->
|
||||
meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
|
||||
meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
|
||||
meck:expect(emqx_resource, remove, fun(_) -> ok end ),
|
||||
|
||||
ok = emqx_config:init_load(emqx_authz_schema, ?CONF_DEFAULT),
|
||||
ok = emqx_ct_helpers:start_apps([emqx_authz]),
|
||||
{ok, _} = emqx:update_config([authorization, cache, enable], false),
|
||||
|
|
@ -39,43 +43,63 @@ init_per_suite(Config) ->
|
|||
|
||||
end_per_suite(_Config) ->
|
||||
{ok, _} = emqx_authz:update(replace, []),
|
||||
emqx_ct_helpers:stop_apps([emqx_authz]),
|
||||
emqx_ct_helpers:stop_apps([emqx_authz, emqx_resource]),
|
||||
meck:unload(emqx_resource),
|
||||
ok.
|
||||
|
||||
init_per_testcase(_, Config) ->
|
||||
{ok, _} = emqx_authz:update(replace, []),
|
||||
Config.
|
||||
|
||||
-define(RULE1, #{<<"principal">> => <<"all">>,
|
||||
<<"topics">> => [<<"#">>],
|
||||
<<"action">> => <<"all">>,
|
||||
<<"permission">> => <<"deny">>}
|
||||
).
|
||||
-define(RULE2, #{<<"principal">> =>
|
||||
#{<<"ipaddress">> => <<"127.0.0.1">>},
|
||||
<<"topics">> =>
|
||||
[#{<<"eq">> => <<"#">>},
|
||||
#{<<"eq">> => <<"+">>}
|
||||
] ,
|
||||
<<"action">> => <<"all">>,
|
||||
<<"permission">> => <<"allow">>}
|
||||
).
|
||||
-define(RULE3,#{<<"principal">> =>
|
||||
#{<<"and">> => [#{<<"username">> => <<"^test?">>},
|
||||
#{<<"clientid">> => <<"^test?">>}
|
||||
]},
|
||||
<<"topics">> => [<<"test">>],
|
||||
<<"action">> => <<"publish">>,
|
||||
<<"permission">> => <<"allow">>}
|
||||
).
|
||||
-define(RULE4,#{<<"principal">> =>
|
||||
#{<<"or">> => [#{<<"username">> => <<"^test">>},
|
||||
#{<<"clientid">> => <<"test?">>}
|
||||
]},
|
||||
<<"topics">> => [<<"%u">>,<<"%c">>],
|
||||
<<"action">> => <<"publish">>,
|
||||
<<"permission">> => <<"deny">>}
|
||||
).
|
||||
-define(RULE1, #{<<"type">> => <<"http">>,
|
||||
<<"config">> => #{
|
||||
<<"url">> => <<"https://fake.com:443/">>,
|
||||
<<"headers">> => #{},
|
||||
<<"method">> => <<"get">>,
|
||||
<<"request_timeout">> => 5000}
|
||||
}).
|
||||
-define(RULE2, #{<<"type">> => <<"mongo">>,
|
||||
<<"config">> => #{
|
||||
<<"mongo_type">> => <<"single">>,
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"collection">> => <<"fake">>,
|
||||
<<"find">> => #{<<"a">> => <<"b">>}
|
||||
}).
|
||||
-define(RULE3, #{<<"type">> => <<"mysql">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"username">> => <<"xx">>,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"sql">> => <<"abcb">>
|
||||
}).
|
||||
-define(RULE4, #{<<"type">> => <<"pgsql">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"username">> => <<"xx">>,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"sql">> => <<"abcb">>
|
||||
}).
|
||||
-define(RULE5, #{<<"type">> => <<"redis">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => 0,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"cmd">> => <<"HGETALL mqtt_authz:%u">>
|
||||
}).
|
||||
|
||||
%%------------------------------------------------------------------------------
|
||||
%% Testcases
|
||||
|
|
@ -86,73 +110,50 @@ t_update_rule(_) ->
|
|||
{ok, _} = emqx_authz:update(head, [?RULE1]),
|
||||
{ok, _} = emqx_authz:update(tail, [?RULE3]),
|
||||
|
||||
dbg:tracer(),dbg:p(all,c),
|
||||
dbg:tpl(hocon_schema, check, cx),
|
||||
Lists1 = emqx_authz:check_rules([?RULE1, ?RULE2, ?RULE3]),
|
||||
?assertMatch(Lists1, emqx:get_config([authorization_rules, rules], [])),
|
||||
|
||||
[#{annotations := #{id := Id1,
|
||||
principal := all,
|
||||
topics := [['#']]}
|
||||
},
|
||||
#{annotations := #{id := Id2,
|
||||
principal := #{ipaddress := {{127,0,0,1},{127,0,0,1},32}},
|
||||
topics := [#{eq := ['#']}, #{eq := ['+']}]}
|
||||
},
|
||||
#{annotations := #{id := Id3,
|
||||
principal :=
|
||||
#{'and' := [#{username := {re_pattern, _, _, _, _}},
|
||||
#{clientid := {re_pattern, _, _, _, _}}
|
||||
]
|
||||
},
|
||||
topics := [[<<"test">>]]}
|
||||
}
|
||||
[#{annotations := #{id := Id1}, type := http},
|
||||
#{annotations := #{id := Id2}, type := mongo},
|
||||
#{annotations := #{id := Id3}, type := mysql}
|
||||
] = emqx_authz:lookup(),
|
||||
|
||||
{ok, _} = emqx_authz:update({replace_once, Id1}, ?RULE5),
|
||||
{ok, _} = emqx_authz:update({replace_once, Id3}, ?RULE4),
|
||||
Lists2 = emqx_authz:check_rules([?RULE1, ?RULE2, ?RULE4]),
|
||||
?assertMatch(Lists2, emqx:get_config([authorization_rules, rules], [])),
|
||||
|
||||
[#{annotations := #{id := Id1,
|
||||
principal := all,
|
||||
topics := [['#']]}
|
||||
},
|
||||
#{annotations := #{id := Id2,
|
||||
principal := #{ipaddress := {{127,0,0,1},{127,0,0,1},32}},
|
||||
topics := [#{eq := ['#']},
|
||||
#{eq := ['+']}]}
|
||||
},
|
||||
#{annotations := #{id := Id3,
|
||||
principal :=
|
||||
#{'or' := [#{username := {re_pattern, _, _, _, _}},
|
||||
#{clientid := {re_pattern, _, _, _, _}}
|
||||
]
|
||||
},
|
||||
topics := [#{pattern := [<<"%u">>]},
|
||||
#{pattern := [<<"%c">>]}
|
||||
]}
|
||||
}
|
||||
[#{annotations := #{id := Id1}, type := redis},
|
||||
#{annotations := #{id := Id2}, type := mongo},
|
||||
#{annotations := #{id := Id3}, type := pgsql}
|
||||
] = emqx_authz:lookup(),
|
||||
|
||||
{ok, _} = emqx_authz:update(replace, []).
|
||||
|
||||
t_move_rule(_) ->
|
||||
{ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4]),
|
||||
{ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4, ?RULE5]),
|
||||
[#{annotations := #{id := Id1}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id4}}
|
||||
#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id5}}
|
||||
] = emqx_authz:lookup(),
|
||||
|
||||
{ok, _} = emqx_authz:move(Id4, <<"top">>),
|
||||
?assertMatch([#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id1}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id3}}
|
||||
#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id5}}
|
||||
], emqx_authz:lookup()),
|
||||
|
||||
{ok, _} = emqx_authz:move(Id1, <<"bottom">>),
|
||||
?assertMatch([#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id5}},
|
||||
#{annotations := #{id := Id1}}
|
||||
], emqx_authz:lookup()),
|
||||
|
||||
|
|
@ -160,66 +161,15 @@ t_move_rule(_) ->
|
|||
?assertMatch([#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id5}},
|
||||
#{annotations := #{id := Id1}}
|
||||
], emqx_authz:lookup()),
|
||||
|
||||
{ok, _} = emqx_authz:move(Id2, #{<<"after">> => Id1}),
|
||||
?assertMatch([#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id5}},
|
||||
#{annotations := #{id := Id1}},
|
||||
#{annotations := #{id := Id2}}
|
||||
], emqx_authz:lookup()),
|
||||
ok.
|
||||
|
||||
t_authz(_) ->
|
||||
ClientInfo1 = #{clientid => <<"test">>,
|
||||
username => <<"test">>,
|
||||
peerhost => {127,0,0,1},
|
||||
zone => default,
|
||||
listener => mqtt_tcp
|
||||
},
|
||||
ClientInfo2 = #{clientid => <<"test">>,
|
||||
username => <<"test">>,
|
||||
peerhost => {192,168,0,10},
|
||||
zone => default,
|
||||
listener => mqtt_tcp
|
||||
},
|
||||
ClientInfo3 = #{clientid => <<"test">>,
|
||||
username => <<"fake">>,
|
||||
peerhost => {127,0,0,1},
|
||||
zone => default,
|
||||
listener => mqtt_tcp
|
||||
},
|
||||
ClientInfo4 = #{clientid => <<"fake">>,
|
||||
username => <<"test">>,
|
||||
peerhost => {127,0,0,1},
|
||||
zone => default,
|
||||
listener => mqtt_tcp
|
||||
},
|
||||
|
||||
Rules1 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE1, ?RULE2])],
|
||||
Rules2 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE2, ?RULE1])],
|
||||
Rules3 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE3, ?RULE4])],
|
||||
Rules4 = [emqx_authz:init_rule(Rule) || Rule <- emqx_authz:check_rules([?RULE4, ?RULE1])],
|
||||
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo1, subscribe, <<"#">>, deny, [])),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo1, subscribe, <<"+">>, deny, Rules1)),
|
||||
?assertEqual({stop, allow},
|
||||
emqx_authz:authorize(ClientInfo1, subscribe, <<"+">>, deny, Rules2)),
|
||||
?assertEqual({stop, allow},
|
||||
emqx_authz:authorize(ClientInfo1, publish, <<"test">>, deny, Rules3)),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo1, publish, <<"test">>, deny, Rules4)),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo2, subscribe, <<"#">>, deny, Rules2)),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo3, publish, <<"test">>, deny, Rules3)),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo3, publish, <<"fake">>, deny, Rules4)),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo4, publish, <<"test">>, deny, Rules3)),
|
||||
?assertEqual({stop, deny},
|
||||
emqx_authz:authorize(ClientInfo4, publish, <<"fake">>, deny, Rules4)),
|
||||
ok.
|
||||
|
|
|
|||
|
|
@ -37,46 +37,100 @@
|
|||
-define(API_VERSION, "v5").
|
||||
-define(BASE_PATH, "api").
|
||||
|
||||
-define(CONF_DEFAULT, <<"authorization: {rules: []}">>).
|
||||
% -define(RULE1, #{<<"principal">> => <<"all">>,
|
||||
% <<"topics">> => [<<"#">>],
|
||||
% <<"action">> => <<"all">>,
|
||||
% <<"permission">> => <<"deny">>}
|
||||
% ).
|
||||
% -define(RULE2, #{<<"principal">> =>
|
||||
% #{<<"ipaddress">> => <<"127.0.0.1">>},
|
||||
% <<"topics">> =>
|
||||
% [#{<<"eq">> => <<"#">>},
|
||||
% #{<<"eq">> => <<"+">>}
|
||||
% ] ,
|
||||
% <<"action">> => <<"all">>,
|
||||
% <<"permission">> => <<"allow">>}
|
||||
% ).
|
||||
% -define(RULE3,#{<<"principal">> =>
|
||||
% #{<<"and">> => [#{<<"username">> => <<"^test?">>},
|
||||
% #{<<"clientid">> => <<"^test?">>}
|
||||
% ]},
|
||||
% <<"topics">> => [<<"test">>],
|
||||
% <<"action">> => <<"publish">>,
|
||||
% <<"permission">> => <<"allow">>}
|
||||
% ).
|
||||
% -define(RULE4,#{<<"principal">> =>
|
||||
% #{<<"or">> => [#{<<"username">> => <<"^test">>},
|
||||
% #{<<"clientid">> => <<"test?">>}
|
||||
% ]},
|
||||
% <<"topics">> => [<<"%u">>,<<"%c">>],
|
||||
% <<"action">> => <<"publish">>,
|
||||
% <<"permission">> => <<"deny">>}
|
||||
% ).
|
||||
|
||||
-define(RULE1, #{<<"principal">> => <<"all">>,
|
||||
<<"topics">> => [<<"#">>],
|
||||
<<"action">> => <<"all">>,
|
||||
<<"permission">> => <<"deny">>}
|
||||
).
|
||||
-define(RULE2, #{<<"principal">> =>
|
||||
#{<<"ipaddress">> => <<"127.0.0.1">>},
|
||||
<<"topics">> =>
|
||||
[#{<<"eq">> => <<"#">>},
|
||||
#{<<"eq">> => <<"+">>}
|
||||
] ,
|
||||
<<"action">> => <<"all">>,
|
||||
<<"permission">> => <<"allow">>}
|
||||
).
|
||||
-define(RULE3,#{<<"principal">> =>
|
||||
#{<<"and">> => [#{<<"username">> => <<"^test?">>},
|
||||
#{<<"clientid">> => <<"^test?">>}
|
||||
]},
|
||||
<<"topics">> => [<<"test">>],
|
||||
<<"action">> => <<"publish">>,
|
||||
<<"permission">> => <<"allow">>}
|
||||
).
|
||||
-define(RULE4,#{<<"principal">> =>
|
||||
#{<<"or">> => [#{<<"username">> => <<"^test">>},
|
||||
#{<<"clientid">> => <<"test?">>}
|
||||
]},
|
||||
<<"topics">> => [<<"%u">>,<<"%c">>],
|
||||
<<"action">> => <<"publish">>,
|
||||
<<"permission">> => <<"deny">>}
|
||||
).
|
||||
-define(RULE1, #{<<"type">> => <<"http">>,
|
||||
<<"config">> => #{
|
||||
<<"url">> => <<"https://fake.com:443/">>,
|
||||
<<"headers">> => #{},
|
||||
<<"method">> => <<"get">>,
|
||||
<<"request_timeout">> => 5000}
|
||||
}).
|
||||
-define(RULE2, #{<<"type">> => <<"mongo">>,
|
||||
<<"config">> => #{
|
||||
<<"mongo_type">> => <<"single">>,
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"collection">> => <<"fake">>,
|
||||
<<"find">> => #{<<"a">> => <<"b">>}
|
||||
}).
|
||||
-define(RULE3, #{<<"type">> => <<"mysql">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"username">> => <<"xx">>,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"sql">> => <<"abcb">>
|
||||
}).
|
||||
-define(RULE4, #{<<"type">> => <<"pgsql">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"username">> => <<"xx">>,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"sql">> => <<"abcb">>
|
||||
}).
|
||||
-define(RULE5, #{<<"type">> => <<"redis">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => 0,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"cmd">> => <<"HGETALL mqtt_authz:%u">>
|
||||
}).
|
||||
|
||||
all() ->
|
||||
emqx_ct:all(?MODULE).
|
||||
% emqx_ct:all(?MODULE).
|
||||
[].
|
||||
|
||||
groups() ->
|
||||
[].
|
||||
|
||||
init_per_suite(Config) ->
|
||||
meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]),
|
||||
meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end),
|
||||
meck:expect(emqx_resource, update, fun(_, _, _, _) -> {ok, meck_data} end),
|
||||
meck:expect(emqx_resource, remove, fun(_) -> ok end ),
|
||||
|
||||
ekka_mnesia:start(),
|
||||
emqx_mgmt_auth:mnesia(boot),
|
||||
|
||||
|
|
@ -89,7 +143,8 @@ init_per_suite(Config) ->
|
|||
|
||||
end_per_suite(_Config) ->
|
||||
{ok, _} = emqx_authz:update(replace, []),
|
||||
emqx_ct_helpers:stop_apps([emqx_authz, emqx_management]),
|
||||
emqx_ct_helpers:stop_apps([emqx_resource, emqx_authz, emqx_management]),
|
||||
meck:unload(emqx_resource),
|
||||
ok.
|
||||
|
||||
set_special_configs(emqx_management) ->
|
||||
|
|
@ -111,12 +166,7 @@ t_api(_) ->
|
|||
?assertEqual([], get_rules(Result1)),
|
||||
|
||||
lists:foreach(fun(_) ->
|
||||
{ok, 204, _} = request(post, uri(["authorization"]),
|
||||
#{<<"action">> => <<"all">>,
|
||||
<<"permission">> => <<"deny">>,
|
||||
<<"principal">> => <<"all">>,
|
||||
<<"topics">> => [<<"#">>]}
|
||||
)
|
||||
{ok, 204, _} = request(post, uri(["authorization"]), ?RULE1)
|
||||
end, lists:seq(1, 20)),
|
||||
{ok, 200, Result2} = request(get, uri(["authorization"]), []),
|
||||
?assertEqual(20, length(get_rules(Result2))),
|
||||
|
|
@ -128,30 +178,23 @@ t_api(_) ->
|
|||
?assertEqual(10, length(get_rules(Result)))
|
||||
end, lists:seq(1, 2)),
|
||||
|
||||
{ok, 204, _} = request(put, uri(["authorization"]),
|
||||
[ #{<<"action">> => <<"all">>, <<"permission">> => <<"allow">>, <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}
|
||||
, #{<<"action">> => <<"all">>, <<"permission">> => <<"allow">>, <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}
|
||||
, #{<<"action">> => <<"all">>, <<"permission">> => <<"allow">>, <<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}
|
||||
]),
|
||||
{ok, 204, _} = request(put, uri(["authorization"]), [?RULE1, ?RULE2, ?RULE3, ?RULE4]),
|
||||
|
||||
{ok, 200, Result3} = request(get, uri(["authorization"]), []),
|
||||
Rules = get_rules(Result3),
|
||||
?assertEqual(3, length(Rules)),
|
||||
|
||||
lists:foreach(fun(#{<<"permission">> := Allow}) ->
|
||||
?assertEqual(<<"allow">>, Allow)
|
||||
end, Rules),
|
||||
?assertEqual(4, length(Rules)),
|
||||
?assertMatch([ #{<<"type">> := <<"http">>}
|
||||
, #{<<"type">> := <<"mongo">>}
|
||||
, #{<<"type">> := <<"mysql">>}
|
||||
, #{<<"type">> := <<"pgsql">>}
|
||||
], Rules),
|
||||
|
||||
#{<<"annotations">> := #{<<"id">> := Id}} = lists:nth(2, Rules),
|
||||
|
||||
{ok, 204, _} = request(put, uri(["authorization", binary_to_list(Id)]),
|
||||
#{<<"action">> => <<"all">>, <<"permission">> => <<"deny">>,
|
||||
<<"principal">> => <<"all">>, <<"topics">> => [<<"#">>]}),
|
||||
{ok, 204, _} = request(put, uri(["authorization", binary_to_list(Id)]), ?RULE5),
|
||||
|
||||
{ok, 200, Result4} = request(get, uri(["authorization", binary_to_list(Id)]), []),
|
||||
?assertMatch(#{<<"annotations">> := #{<<"id">> := Id},
|
||||
<<"permission">> := <<"deny">>
|
||||
}, jsx:decode(Result4)),
|
||||
?assertMatch(#{<<"type">> := <<"redis">>}, jsx:decode(Result4)),
|
||||
|
||||
lists:foreach(fun(#{<<"annotations">> := #{<<"id">> := Id0}}) ->
|
||||
{ok, 204, _} = request(delete, uri(["authorization", binary_to_list(Id0)]), [])
|
||||
|
|
@ -161,11 +204,12 @@ t_api(_) ->
|
|||
ok.
|
||||
|
||||
t_move_rule(_) ->
|
||||
{ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4]),
|
||||
{ok, _} = emqx_authz:update(replace, [?RULE1, ?RULE2, ?RULE3, ?RULE4, ?RULE5]),
|
||||
[#{annotations := #{id := Id1}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id4}}
|
||||
#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id5}}
|
||||
] = emqx_authz:lookup(),
|
||||
|
||||
{ok, 204, _} = request(post, uri(["authorization", Id4, "move"]),
|
||||
|
|
@ -173,7 +217,8 @@ t_move_rule(_) ->
|
|||
?assertMatch([#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id1}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id3}}
|
||||
#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id5}}
|
||||
], emqx_authz:lookup()),
|
||||
|
||||
{ok, 204, _} = request(post, uri(["authorization", Id1, "move"]),
|
||||
|
|
@ -181,6 +226,7 @@ t_move_rule(_) ->
|
|||
?assertMatch([#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id5}},
|
||||
#{annotations := #{id := Id1}}
|
||||
], emqx_authz:lookup()),
|
||||
|
||||
|
|
@ -189,6 +235,7 @@ t_move_rule(_) ->
|
|||
?assertMatch([#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id2}},
|
||||
#{annotations := #{id := Id5}},
|
||||
#{annotations := #{id := Id1}}
|
||||
], emqx_authz:lookup()),
|
||||
|
||||
|
|
@ -196,6 +243,7 @@ t_move_rule(_) ->
|
|||
#{<<"position">> => #{<<"after">> => Id1}}),
|
||||
?assertMatch([#{annotations := #{id := Id3}},
|
||||
#{annotations := #{id := Id4}},
|
||||
#{annotations := #{id := Id5}},
|
||||
#{annotations := #{id := Id1}},
|
||||
#{annotations := #{id := Id2}}
|
||||
], emqx_authz:lookup()),
|
||||
|
|
|
|||
|
|
@ -41,14 +41,13 @@ init_per_suite(Config) ->
|
|||
|
||||
{ok, _} = emqx:update_config([authorization, cache, enable], false),
|
||||
{ok, _} = emqx:update_config([authorization, no_match], deny),
|
||||
Rules = [#{ <<"config">> => #{
|
||||
Rules = [#{<<"type">> => <<"http">>,
|
||||
<<"config">> => #{
|
||||
<<"url">> => <<"https://fake.com:443/">>,
|
||||
<<"headers">> => #{},
|
||||
<<"method">> => <<"get">>,
|
||||
<<"request_timeout">> => 5000
|
||||
},
|
||||
<<"principal">> => <<"all">>,
|
||||
<<"type">> => <<"http">>}
|
||||
}}
|
||||
],
|
||||
{ok, _} = emqx_authz:update(replace, Rules),
|
||||
Config.
|
||||
|
|
|
|||
|
|
@ -39,17 +39,16 @@ init_per_suite(Config) ->
|
|||
ok = emqx_ct_helpers:start_apps([emqx_authz]),
|
||||
{ok, _} = emqx:update_config([authorization, cache, enable], false),
|
||||
{ok, _} = emqx:update_config([authorization, no_match], deny),
|
||||
Rules = [#{ <<"config">> => #{
|
||||
Rules = [#{<<"type">> => <<"mongo">>,
|
||||
<<"config">> => #{
|
||||
<<"mongo_type">> => <<"single">>,
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"principal">> => <<"all">>,
|
||||
<<"collection">> => <<"fake">>,
|
||||
<<"find">> => #{<<"a">> => <<"b">>},
|
||||
<<"type">> => <<"mongo">>}
|
||||
],
|
||||
<<"find">> => #{<<"a">> => <<"b">>}
|
||||
}],
|
||||
{ok, _} = emqx_authz:update(replace, Rules),
|
||||
Config.
|
||||
|
||||
|
|
|
|||
|
|
@ -40,18 +40,17 @@ init_per_suite(Config) ->
|
|||
|
||||
{ok, _} = emqx:update_config([authorization, cache, enable], false),
|
||||
{ok, _} = emqx:update_config([authorization, no_match], deny),
|
||||
Rules = [#{ <<"config">> => #{
|
||||
Rules = [#{<<"type">> => <<"mysql">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"username">> => <<"xx">>,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}
|
||||
},
|
||||
<<"principal">> => <<"all">>,
|
||||
<<"sql">> => <<"abcb">>,
|
||||
<<"type">> => <<"mysql">> }],
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"sql">> => <<"abcb">>
|
||||
}],
|
||||
{ok, _} = emqx_authz:update(replace, Rules),
|
||||
Config.
|
||||
|
||||
|
|
@ -60,17 +59,14 @@ end_per_suite(_Config) ->
|
|||
emqx_ct_helpers:stop_apps([emqx_authz, emqx_resource]),
|
||||
meck:unload(emqx_resource).
|
||||
|
||||
-define(COLUMNS, [ <<"ipaddress">>
|
||||
, <<"username">>
|
||||
, <<"clientid">>
|
||||
, <<"action">>
|
||||
-define(COLUMNS, [ <<"action">>
|
||||
, <<"permission">>
|
||||
, <<"topic">>
|
||||
]).
|
||||
-define(RULE1, [[<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"deny">>, <<"#">>]]).
|
||||
-define(RULE2, [[<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"allow">>, <<"eq #">>]]).
|
||||
-define(RULE3, [[<<>>, <<"^test">>, <<"^test">> ,<<"subscribe">>, <<"allow">>, <<"test/%c">>]]).
|
||||
-define(RULE4, [[<<>>, <<"^test">>, <<"^test">> ,<<"publish">>, <<"allow">>, <<"test/%u">>]]).
|
||||
-define(RULE1, [[<<"all">>, <<"deny">>, <<"#">>]]).
|
||||
-define(RULE2, [[<<"all">>, <<"allow">>, <<"eq #">>]]).
|
||||
-define(RULE3, [[<<"subscribe">>, <<"allow">>, <<"test/%c">>]]).
|
||||
-define(RULE4, [[<<"publish">>, <<"allow">>, <<"test/%u">>]]).
|
||||
|
||||
%%------------------------------------------------------------------------------
|
||||
%% Testcases
|
||||
|
|
|
|||
|
|
@ -40,17 +40,17 @@ init_per_suite(Config) ->
|
|||
|
||||
{ok, _} = emqx:update_config([authorization, cache, enable], false),
|
||||
{ok, _} = emqx:update_config([authorization, no_match], deny),
|
||||
Rules = [#{ <<"config">> => #{
|
||||
Rules = [#{<<"type">> => <<"pgsql">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => <<"mqtt">>,
|
||||
<<"username">> => <<"xx">>,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}
|
||||
},
|
||||
<<"sql">> => <<"abcb">>,
|
||||
<<"type">> => <<"pgsql">> }],
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"sql">> => <<"abcb">>
|
||||
}],
|
||||
{ok, _} = emqx_authz:update(replace, Rules),
|
||||
Config.
|
||||
|
||||
|
|
@ -59,17 +59,14 @@ end_per_suite(_Config) ->
|
|||
emqx_ct_helpers:stop_apps([emqx_authz, emqx_resource]),
|
||||
meck:unload(emqx_resource).
|
||||
|
||||
-define(COLUMNS, [ {column, <<"ipaddress">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
, {column, <<"username">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
, {column, <<"clientid">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
, {column, <<"action">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
-define(COLUMNS, [ {column, <<"action">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
, {column, <<"permission">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
, {column, <<"topic">>, meck, meck, meck, meck, meck, meck, meck}
|
||||
]).
|
||||
-define(RULE1, [{<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"deny">>, <<"#">>}]).
|
||||
-define(RULE2, [{<<"127.0.0.1">>, <<>>, <<>>, <<"all">>, <<"allow">>, <<"eq #">>}]).
|
||||
-define(RULE3, [{<<>>, <<"^test">>, <<"^test">> ,<<"subscribe">>, <<"allow">>, <<"test/%c">>}]).
|
||||
-define(RULE4, [{<<>>, <<"^test">>, <<"^test">> ,<<"publish">>, <<"allow">>, <<"test/%u">>}]).
|
||||
-define(RULE1, [{<<"all">>, <<"deny">>, <<"#">>}]).
|
||||
-define(RULE2, [{<<"all">>, <<"allow">>, <<"eq #">>}]).
|
||||
-define(RULE3, [{<<"subscribe">>, <<"allow">>, <<"test/%c">>}]).
|
||||
-define(RULE4, [{<<"publish">>, <<"allow">>, <<"test/%u">>}]).
|
||||
|
||||
%%------------------------------------------------------------------------------
|
||||
%% Testcases
|
||||
|
|
|
|||
|
|
@ -41,16 +41,16 @@ init_per_suite(Config) ->
|
|||
|
||||
{ok, _} = emqx:update_config([authorization, cache, enable], false),
|
||||
{ok, _} = emqx:update_config([authorization, no_match], deny),
|
||||
Rules = [#{ <<"config">> => #{
|
||||
Rules = [#{<<"type">> => <<"redis">>,
|
||||
<<"config">> => #{
|
||||
<<"server">> => <<"127.0.0.1:27017">>,
|
||||
<<"pool_size">> => 1,
|
||||
<<"database">> => 0,
|
||||
<<"password">> => <<"ee">>,
|
||||
<<"auto_reconnect">> => true,
|
||||
<<"ssl">> => #{<<"enable">> => false}
|
||||
},
|
||||
<<"cmd">> => <<"HGETALL mqtt_authz:%u">>,
|
||||
<<"type">> => <<"redis">> }],
|
||||
<<"ssl">> => #{<<"enable">> => false}},
|
||||
<<"cmd">> => <<"HGETALL mqtt_authz:%u">>
|
||||
}],
|
||||
{ok, _} = emqx_authz:update(replace, Rules),
|
||||
Config.
|
||||
|
||||
|
|
|
|||
|
|
@ -19,8 +19,9 @@
|
|||
-include_lib("typerefl/include/types.hrl").
|
||||
-include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
|
||||
|
||||
-type server() :: string().
|
||||
-type server() :: emqx_schema:ip_port().
|
||||
-reflect_type([server/0]).
|
||||
-typerefl_from_string({server/0, emqx_connector_schema_lib, to_ip_port}).
|
||||
|
||||
%% callbacks of behaviour emqx_resource
|
||||
-export([ on_start/2
|
||||
|
|
@ -95,7 +96,7 @@ on_start(InstId, Config = #{server := Server,
|
|||
mongo_type := single}) ->
|
||||
logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
|
||||
Opts = [{type, single},
|
||||
{hosts, [Server]}
|
||||
{hosts, [emqx_connector_schema_lib:ip_port_to_string(Server)]}
|
||||
],
|
||||
do_start(InstId, Opts, Config);
|
||||
|
||||
|
|
@ -104,14 +105,17 @@ on_start(InstId, Config = #{servers := Servers,
|
|||
replica_set_name := RsName}) ->
|
||||
logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
|
||||
Opts = [{type, {rs, RsName}},
|
||||
{hosts, Servers}],
|
||||
{hosts, [emqx_connector_schema_lib:ip_port_to_string(S)
|
||||
|| S <- Servers]}
|
||||
],
|
||||
do_start(InstId, Opts, Config);
|
||||
|
||||
on_start(InstId, Config = #{servers := Servers,
|
||||
mongo_type := sharded}) ->
|
||||
logger:info("starting mongodb connector: ~p, config: ~p", [InstId, Config]),
|
||||
Opts = [{type, sharded},
|
||||
{hosts, Servers}
|
||||
{hosts, [emqx_connector_schema_lib:ip_port_to_string(S)
|
||||
|| S <- Servers]}
|
||||
],
|
||||
do_start(InstId, Opts, Config).
|
||||
|
||||
|
|
|
|||
|
|
@ -19,10 +19,9 @@
|
|||
-include_lib("typerefl/include/types.hrl").
|
||||
-include_lib("emqx_resource/include/emqx_resource_behaviour.hrl").
|
||||
|
||||
-type server() :: tuple().
|
||||
-type server() :: emqx_schema:ip_port().
|
||||
-reflect_type([server/0]).
|
||||
-typerefl_from_string({server/0, ?MODULE, to_server}).
|
||||
-export([to_server/1]).
|
||||
-typerefl_from_string({server/0, emqx_connector_schema_lib, to_ip_port}).
|
||||
|
||||
-export([structs/0, fields/1]).
|
||||
|
||||
|
|
@ -170,9 +169,3 @@ redis_fields() ->
|
|||
default => 0}}
|
||||
, {auto_reconnect, fun emqx_connector_schema_lib:auto_reconnect/1}
|
||||
].
|
||||
|
||||
to_server(Server) ->
|
||||
case string:tokens(Server, ":") of
|
||||
[Host, Port] -> {ok, {Host, list_to_integer(Port)}};
|
||||
_ -> {error, Server}
|
||||
end.
|
||||
|
|
|
|||
Loading…
Reference in New Issue