diff --git a/etc/emqx.conf b/etc/emqx.conf index 09ce1b526..7d4d8c966 100644 --- a/etc/emqx.conf +++ b/etc/emqx.conf @@ -779,7 +779,7 @@ listener.tcp.internal.acceptors = 4 ## Maximum number of concurrent MQTT/TCP connections. ## ## Value: Number -listener.tcp.internal.max_connections = 10240000 +listener.tcp.internal.max_connections = 1024000 ## Maximum internal connections per second. ## diff --git a/src/emqx_access_rule.erl b/src/emqx_access_rule.erl index 3fb6dd7ef..47b20676b 100644 --- a/src/emqx_access_rule.erl +++ b/src/emqx_access_rule.erl @@ -34,16 +34,21 @@ -export([match/3]). -define(ALLOW_DENY(A), ((A =:= allow) orelse (A =:= deny))). +-define(PUBSUB(A), ((A =:= subscribe) orelse (A =:= publish) orelse (A =:= pubsub))). %% @doc Compile Access Rule. compile({A, all}) when ?ALLOW_DENY(A) -> {A, all}; -compile({A, Who, Access, Topic}) when ?ALLOW_DENY(A), is_binary(Topic) -> +compile({A, Who, Access, Topic}) when ?ALLOW_DENY(A), ?PUBSUB(Access), is_binary(Topic) -> {A, compile(who, Who), Access, [compile(topic, Topic)]}; -compile({A, Who, Access, TopicFilters}) when ?ALLOW_DENY(A) -> - {A, compile(who, Who), Access, [compile(topic, Topic) || Topic <- TopicFilters]}. +compile({A, Who, Access, TopicFilters}) when ?ALLOW_DENY(A), ?PUBSUB(Access) -> + {A, compile(who, Who), Access, [compile(topic, Topic) || Topic <- TopicFilters]}; + +compile(Rule) -> + emqx_logger:error("[ACCESS_RULE] Malformed rule: ~p", [Rule]), + {error, bad_rule}. compile(who, all) -> all; diff --git a/src/emqx_acl_internal.erl b/src/emqx_acl_internal.erl index f8b995096..328993a78 100644 --- a/src/emqx_acl_internal.erl +++ b/src/emqx_acl_internal.erl @@ -45,19 +45,27 @@ all_rules() -> -spec(init([File :: string()]) -> {ok, #{}}). init([File]) -> - ok = emqx_tables:new(?ACL_RULE_TAB, [set, public, {read_concurrency, true}]), - true = load_rules_from_file(File), + _ = emqx_tables:new(?ACL_RULE_TAB, [set, public, {read_concurrency, true}]), + ok = load_rules_from_file(File), {ok, #{acl_file => File}}. load_rules_from_file(AclFile) -> - {ok, Terms} = file:consult(AclFile), - Rules = [emqx_access_rule:compile(Term) || Term <- Terms], - lists:foreach(fun(PubSub) -> - ets:insert(?ACL_RULE_TAB, {PubSub, - lists:filter(fun(Rule) -> filter(PubSub, Rule) end, Rules)}) - end, [publish, subscribe]), - ets:insert(?ACL_RULE_TAB, {all_rules, Terms}). + case file:consult(AclFile) of + {ok, Terms} -> + Rules = [emqx_access_rule:compile(Term) || Term <- Terms], + lists:foreach(fun(PubSub) -> + ets:insert(?ACL_RULE_TAB, {PubSub, + lists:filter(fun(Rule) -> filter(PubSub, Rule) end, Rules)}) + end, [publish, subscribe]), + ets:insert(?ACL_RULE_TAB, {all_rules, Terms}), + ok; + {error, Reason} -> + emqx_logger:error("[ACL_INTERNAL] Failed to read ~s: ~p", [AclFile, Reason]), + {error, Reason} + end. +filter(_PubSub, {error, _}) -> + false; filter(_PubSub, {allow, all}) -> true; filter(_PubSub, {deny, all}) -> @@ -100,9 +108,11 @@ match(Credentials, Topic, [Rule|Rules]) -> -spec(reload_acl(state()) -> ok | {error, term()}). reload_acl(#{acl_file := AclFile}) -> case catch load_rules_from_file(AclFile) of - true -> + ok -> emqx_logger:info("Reload acl_file ~s successfully", [AclFile]), ok; + {error, Error} -> + {error, Error}; {'EXIT', Error} -> {error, Error} end. diff --git a/src/emqx_banned.erl b/src/emqx_banned.erl index af580b48e..0c9ffdd10 100644 --- a/src/emqx_banned.erl +++ b/src/emqx_banned.erl @@ -104,7 +104,7 @@ code_change(_OldVsn, State, _Extra) -> -ifdef(TEST). ensure_expiry_timer(State) -> - State#{expiry_timer := emqx_misc:start_timer(timer:seconds(2), expire)}. + State#{expiry_timer := emqx_misc:start_timer(timer:seconds(1), expire)}. -else. ensure_expiry_timer(State) -> State#{expiry_timer := emqx_misc:start_timer(timer:minutes(1), expire)}. diff --git a/src/emqx_protocol.erl b/src/emqx_protocol.erl index dad40e190..8f845a94e 100644 --- a/src/emqx_protocol.erl +++ b/src/emqx_protocol.erl @@ -620,14 +620,12 @@ maybe_use_username_as_clientid(ClientId, undefined, _PState) -> ClientId; maybe_use_username_as_clientid(ClientId, Username, #pstate{zone = Zone}) -> case emqx_zone:get_env(Zone, use_username_as_clientid, false) of - true -> - Username; - false -> - ClientId + true -> Username; + false -> ClientId end. %%------------------------------------------------------------------------------ -%% Assign a clientid +%% Assign a clientId maybe_assign_client_id(PState = #pstate{client_id = <<>>, ack_props = AckProps}) -> ClientId = emqx_guid:to_base62(emqx_guid:gen()), @@ -651,41 +649,30 @@ try_open_session(PState = #pstate{zone = Zone, clean_start => CleanStart, will_msg => WillMsg }, - SessAttrs1 = lists:foldl(fun set_session_attrs/2, SessAttrs, [{max_inflight, PState}, {expiry_interval, PState}, {topic_alias_maximum, PState}]), + SessAttrs1 = lists:foldl(fun set_session_attrs/2, SessAttrs, [{max_inflight, PState}, {expiry_interval, PState}]), case emqx_sm:open_session(SessAttrs1) of {ok, SPid} -> {ok, SPid, false}; Other -> Other end. -set_session_attrs({max_inflight, #pstate{zone = Zone, proto_ver = ProtoVer, conn_props = ConnProps}}, SessAttrs) -> - maps:put(max_inflight, if - ProtoVer =:= ?MQTT_PROTO_V5 -> - get_property('Receive-Maximum', ConnProps, 65535); - true -> - emqx_zone:get_env(Zone, max_inflight, 65535) - end, SessAttrs); -set_session_attrs({expiry_interval, #pstate{zone = Zone, proto_ver = ProtoVer, conn_props = ConnProps, clean_start = CleanStart}}, SessAttrs) -> - maps:put(expiry_interval, if - ProtoVer =:= ?MQTT_PROTO_V5 -> - get_property('Session-Expiry-Interval', ConnProps, 0); - true -> - case CleanStart of - true -> 0; - false -> - emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff) - end - end, SessAttrs); -set_session_attrs({topic_alias_maximum, #pstate{zone = Zone, proto_ver = ProtoVer, conn_props = ConnProps}}, SessAttrs) -> - maps:put(topic_alias_maximum, if - ProtoVer =:= ?MQTT_PROTO_V5 -> - get_property('Topic-Alias-Maximum', ConnProps, 0); - true -> - emqx_zone:get_env(Zone, max_topic_alias, 0) - end, SessAttrs); -set_session_attrs({_, #pstate{}}, SessAttrs) -> - SessAttrs. +set_session_attrs({max_inflight, #pstate{proto_ver = ?MQTT_PROTO_V5, conn_props = ConnProps}}, SessAttrs) -> + maps:put(max_inflight, get_property('Receive-Maximum', ConnProps, 65535), SessAttrs); +set_session_attrs({max_inflight, #pstate{zone = Zone}}, SessAttrs) -> + maps:put(max_inflight, emqx_zone:get_env(Zone, max_inflight, 65535), SessAttrs); + +set_session_attrs({expiry_interval, #pstate{proto_ver = ?MQTT_PROTO_V5, conn_props = ConnProps}}, SessAttrs) -> + maps:put(expiry_interval, get_property('Session-Expiry-Interval', ConnProps, 0), SessAttrs); + +set_session_attrs({expiry_interval, #pstate{zone = Zone, clean_start = CleanStart}}, SessAttrs) -> + maps:put(expiry_interval, case CleanStart of + true -> 0; + false -> emqx_zone:get_env(Zone, session_expiry_interval, 16#ffffffff) + end, SessAttrs); + +set_session_attrs(_, SessAttrs) -> + SessAttrs. authenticate(Credentials, Password) -> case emqx_access_control:authenticate(Credentials, Password) of diff --git a/src/emqx_session.erl b/src/emqx_session.erl index f41a25fc5..1e186b0ab 100644 --- a/src/emqx_session.erl +++ b/src/emqx_session.erl @@ -147,8 +147,6 @@ %% Created at created_at :: erlang:timestamp(), - topic_alias_maximum :: pos_integer(), - will_msg :: emqx:message(), will_delay_timer :: reference() | undefined @@ -341,7 +339,6 @@ init([Parent, #{zone := Zone, clean_start := CleanStart, expiry_interval := ExpiryInterval, max_inflight := MaxInflight, - topic_alias_maximum := TopicAliasMaximum, will_msg := WillMsg}]) -> emqx_logger:set_metadata_client_id(ClientId), process_flag(trap_exit, true), @@ -367,7 +364,6 @@ init([Parent, #{zone := Zone, deliver_stats = 0, enqueue_stats = 0, created_at = os:timestamp(), - topic_alias_maximum = TopicAliasMaximum, will_msg = WillMsg }, ok = emqx_sm:register_session(ClientId, self()), @@ -519,22 +515,23 @@ handle_cast({pubcomp, PacketId, _ReasonCode}, State = #state{inflight = Inflight end; %% RESUME: -handle_cast({resume, #{conn_pid := ConnPid, - will_msg := WillMsg, - expiry_interval := SessionExpiryInterval, - max_inflight := MaxInflight, - topic_alias_maximum := TopicAliasMaximum}}, State = #state{client_id = ClientId, - conn_pid = OldConnPid, - clean_start = CleanStart, - retry_timer = RetryTimer, - await_rel_timer = AwaitTimer, - expiry_timer = ExpireTimer, - will_delay_timer = WillDelayTimer}) -> +handle_cast({resume, #{conn_pid := ConnPid, + will_msg := WillMsg, + expiry_interval := ExpiryInterval, + max_inflight := MaxInflight}}, + State = #state{client_id = ClientId, + conn_pid = OldConnPid, + clean_start = CleanStart, + retry_timer = RetryTimer, + await_rel_timer = AwaitTimer, + expiry_timer = ExpireTimer, + will_delay_timer = WillDelayTimer}) -> ?LOG(info, "Resumed by connection ~p ", [ConnPid], State), %% Cancel Timers - lists:foreach(fun emqx_misc:cancel_timer/1, [RetryTimer, AwaitTimer, ExpireTimer, WillDelayTimer]), + lists:foreach(fun emqx_misc:cancel_timer/1, + [RetryTimer, AwaitTimer, ExpireTimer, WillDelayTimer]), case kick(ClientId, OldConnPid, ConnPid) of ok -> ?LOG(warning, "Connection ~p kickout ~p", [ConnPid, OldConnPid], State); @@ -543,19 +540,18 @@ handle_cast({resume, #{conn_pid := ConnPid, true = link(ConnPid), - State1 = State#state{conn_pid = ConnPid, - binding = binding(ConnPid), - old_conn_pid = OldConnPid, - clean_start = false, - retry_timer = undefined, - awaiting_rel = #{}, - await_rel_timer = undefined, - expiry_timer = undefined, - expiry_interval = SessionExpiryInterval, - inflight = emqx_inflight:update_size(MaxInflight, State#state.inflight), - topic_alias_maximum = TopicAliasMaximum, - will_delay_timer = undefined, - will_msg = WillMsg}, + State1 = State#state{conn_pid = ConnPid, + binding = binding(ConnPid), + old_conn_pid = OldConnPid, + clean_start = false, + retry_timer = undefined, + awaiting_rel = #{}, + await_rel_timer = undefined, + expiry_timer = undefined, + expiry_interval = ExpiryInterval, + inflight = emqx_inflight:update_size(MaxInflight, State#state.inflight), + will_delay_timer = undefined, + will_msg = WillMsg}, %% Clean Session: true -> false??? CleanStart andalso emqx_sm:set_session_attrs(ClientId, attrs(State1)), @@ -574,9 +570,10 @@ handle_cast(Msg, State) -> %% Batch dispatch handle_info({dispatch, Topic, Msgs}, State) when is_list(Msgs) -> - {noreply, lists:foldl(fun(Msg, NewState) -> - element(2, handle_info({dispatch, Topic, Msg}, NewState)) - end, State, Msgs)}; + noreply(lists:foldl( + fun(Msg, St) -> + element(2, handle_info({dispatch, Topic, Msg}, St)) + end, State, Msgs)); %% Dispatch message handle_info({dispatch, Topic, Msg = #message{}}, State) -> @@ -585,12 +582,11 @@ handle_info({dispatch, Topic, Msg = #message{}}, State) -> %% Require ack, but we do not have connection %% negative ack the message so it can try the next subscriber in the group ok = emqx_shared_sub:nack_no_connection(Msg), - noreply(State); + {noreply, State}; false -> - handle_dispatch(Topic, Msg, State) + noreply(handle_dispatch(Topic, Msg, State)) end; - %% Do nothing if the client has been disconnected. handle_info({timeout, Timer, retry_delivery}, State = #state{conn_pid = undefined, retry_timer = Timer}) -> noreply(State#state{retry_timer = undefined}); @@ -678,23 +674,14 @@ maybe_shutdown(Pid, Reason) -> has_connection(#state{conn_pid = Pid}) -> is_pid(Pid) andalso is_process_alive(Pid). -handle_dispatch(Topic, Msg = #message{headers = Headers}, - State = #state{subscriptions = SubMap, - topic_alias_maximum = TopicAliasMaximum - }) -> - TopicAlias = maps:get('Topic-Alias', Headers, undefined), - if - TopicAlias =:= undefined orelse TopicAlias =< TopicAliasMaximum -> - noreply(case maps:find(Topic, SubMap) of - {ok, #{nl := Nl, qos := QoS, rap := Rap, subid := SubId}} -> - run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}, {subid, SubId}], Msg, State); - {ok, #{nl := Nl, qos := QoS, rap := Rap}} -> - run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}], Msg, State); - error -> - dispatch(emqx_message:unset_flag(dup, Msg), State) - end); - true -> - noreply(State) +handle_dispatch(Topic, Msg, State = #state{subscriptions = SubMap}) -> + case maps:find(Topic, SubMap) of + {ok, #{nl := Nl, qos := QoS, rap := Rap, subid := SubId}} -> + run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}, {subid, SubId}], Msg, State); + {ok, #{nl := Nl, qos := QoS, rap := Rap}} -> + run_dispatch_steps([{nl, Nl}, {qos, QoS}, {rap, Rap}], Msg, State); + error -> + dispatch(emqx_message:unset_flag(dup, Msg), State) end. suback(_From, undefined, _ReasonCodes) -> diff --git a/test/emqx_access_SUITE.erl b/test/emqx_access_SUITE.erl index b49c90d80..5d0bcf049 100644 --- a/test/emqx_access_SUITE.erl +++ b/test/emqx_access_SUITE.erl @@ -355,7 +355,8 @@ compile_rule(_) -> {deny, all, subscribe, [ [<<"$SYS">>, '#'], ['#'] ]} = compile({deny, all, subscribe, ["$SYS/#", "#"]}), {allow, all} = compile({allow, all}), - {deny, all} = compile({deny, all}). + {deny, all} = compile({deny, all}), + {error, bad_rule} = compile({test, malformed}). match_rule(_) -> User = #{client_id => <<"testClient">>, username => <<"TestUser">>, peername => {{127,0,0,1}, 2948}},