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

feat(config): make quic listener start with the new config

This commit is contained in:
Shawn 2021-07-08 15:05:07 +08:00
parent 477097c062
commit 27d5c5b2d9
4 changed files with 25 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -841,6 +841,7 @@ zones.default {
stats.enable: true
## Maximum number of concurrent connections in this zone.
##
## This value must be larger than the sum of `max_connections` set
## in the listeners under this zone.
##
@ -1258,7 +1259,6 @@ zones.default {
min_alarm_sustain_duration: 1m
}
listeners.mqtt_tcp:
#${example_common_tcp_options} # common options can be written in a separate config entry and reference it from here.
{
@ -1531,7 +1531,6 @@ zones.default {
}
listeners.mqtt_quic:
#${example_common_ssl_options} # common options can be written in a separate config entry and reference it from here.
{
## The type of the listener.
@ -1566,13 +1565,19 @@ zones.default {
## Default: infinity
max_connections: 1024000
## SSL options
## See ${example_common_ssl_options} for more information
ssl.enable: false
#ssl.versions: ["tlsv1.3", "tlsv1.2", "tlsv1.1", "tlsv1"]
#ssl.keyfile: "{{ platform_etc_dir }}/certs/key.pem"
#ssl.certfile: "{{ platform_etc_dir }}/certs/cert.pem"
#ssl.cacertfile: "{{ platform_etc_dir }}/certs/cacert.pem"
## Path to the file containing the user's private PEM-encoded key.
##
## @doc zones.<name>.listeners.<name>.keyfile
## ValueType: String
## Default: "{{ platform_etc_dir }}/certs/key.pem"
keyfile: "{{ platform_etc_dir }}/certs/key.pem"
## Path to a file containing the user certificate.
##
## @doc zones.<name>.listeners.<name>.certfile
## ValueType: String
## Default: "{{ platform_etc_dir }}/certs/cert.pem"
certfile: "{{ platform_etc_dir }}/certs/cert.pem"
}
listeners.mqtt_ws:

10
apps/emqx/src/emqx_listeners.erl
View File

@ -86,10 +86,9 @@ do_start_listener(ZoneName, ListenerName, #{type := quic, bind := ListenOn} = Op
%% @fixme unsure why we need reopen lib and reopen config.
quicer_nif:open_lib(),
quicer_nif:reg_open(),
SSLOpts = ssl_opts(Opts),
DefAcceptors = erlang:system_info(schedulers_online) * 8,
ListenOpts = [ {cert, maps:get(certfile, SSLOpts, undefined)}
, {key, maps:get(keyfile, SSLOpts, undefined)}
ListenOpts = [ {cert, maps:get(certfile, Opts)}
, {key, maps:get(keyfile, Opts)}
, {alpn, ["mqtt"]}
, {conn_acceptors, maps:get(acceptors, Opts, DefAcceptors)}
, {idle_timeout_ms, emqx_config:get_listener_conf(ZoneName, ListenerName,
@ -100,7 +99,7 @@ do_start_listener(ZoneName, ListenerName, #{type := quic, bind := ListenOn} = Op
, peer_bidi_stream_count => 10
},
StreamOpts = [],
quicer:start_listener('mqtt:quic', ListenOn, {ListenOpts, ConnectionOpts, StreamOpts}).
quicer:start_listener('mqtt:quic', port(ListenOn), {ListenOpts, ConnectionOpts, StreamOpts}).
esockd_opts(Opts0) ->
Opts1 = maps:with([acceptors, max_connections, proxy_protocol, proxy_protocol_timeout], Opts0),
@ -140,6 +139,9 @@ ip_port(Port) when is_integer(Port) ->
ip_port({Addr, Port}) ->
[{ip, Addr}, {port, Port}].
port(Port) when is_integer(Port) -> Port;
port({_Addr, Port}) when is_integer(Port) -> Port.
esockd_access_rules(StrRules) ->
Access = fun(S) ->
[A, CIDR] = string:tokens(S, " "),

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

@ -372,6 +372,11 @@ fields("mqtt_ws_listener") ->
fields("mqtt_quic_listener") ->
[ {"type", t(quic)}
, {"certfile", t(string(), undefined, undefined)}
, {"keyfile", t(string(), undefined, undefined)}
, {"ciphers", t(comma_separated_list(), undefined, "TLS_AES_256_GCM_SHA384,"
"TLS_AES_128_GCM_SHA256,TLS_CHACHA20_POLY1305_SHA256")}
, {"idle_timeout", t(duration(), undefined, 60000)}
] ++ base_listener();
fields("ws_opts") ->

121
apps/emqx_modules/src/emqx_mod_acl_internal.erl
View File

@ -1,121 +0,0 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mod_acl_internal).
-behaviour(emqx_gen_mod).
-include_lib("emqx/include/emqx.hrl").
-include_lib("emqx/include/logger.hrl").
-logger_header("[ACL_INTERNAL]").
%% APIs
-export([ check_acl/5
, rules_from_file/1
]).
%% emqx_gen_mod callbacks
-export([ load/1
, unload/1
, reload/1
, description/0
]).
-type(acl_rules() :: #{publish => [emqx_access_rule:rule()],
subscribe => [emqx_access_rule:rule()]}).
%%--------------------------------------------------------------------
%% API
%%--------------------------------------------------------------------
load(Env) ->
Rules = rules_from_file(proplists:get_value(acl_file, Env)),
emqx_hooks:add('client.check_acl', {?MODULE, check_acl, [Rules]}, -1).
unload(_Env) ->
emqx_hooks:del('client.check_acl', {?MODULE, check_acl}).
reload(Env) ->
lists:foreach(
fun(Pid) -> erlang:send(Pid, clean_acl_cache) end,
emqx_cm:all_channels()),
unload(Env), load(Env).
description() ->
"EMQ X Internal ACL Module".
%%--------------------------------------------------------------------
%% ACL callbacks
%%--------------------------------------------------------------------
%% @doc Check ACL
-spec(check_acl(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_topic:topic(),
emqx_access_rule:acl_result(), acl_rules())
-> {ok, allow} | {ok, deny} | ok).
check_acl(Client, PubSub, Topic, _AclResult, Rules) ->
case match(Client, Topic, lookup(PubSub, Rules)) of
{matched, allow} -> {ok, allow};
{matched, deny} -> {ok, deny};
nomatch -> ok
end.
%%--------------------------------------------------------------------
%% Internal Functions
%%--------------------------------------------------------------------
lookup(PubSub, Rules) ->
maps:get(PubSub, Rules, []).
match(_Client, _Topic, []) ->
nomatch;
match(Client, Topic, [Rule|Rules]) ->
case emqx_access_rule:match(Client, Topic, Rule) of
nomatch ->
match(Client, Topic, Rules);
{matched, AllowDeny} ->
{matched, AllowDeny}
end.
-spec(rules_from_file(file:filename()) -> map()).
rules_from_file(AclFile) ->
case file:consult(AclFile) of
{ok, Terms} ->
Rules = [emqx_access_rule:compile(Term) || Term <- Terms],
#{publish => [Rule || Rule <- Rules, filter(publish, Rule)],
subscribe => [Rule || Rule <- Rules, filter(subscribe, Rule)]};
{error, eacces} ->
?LOG(alert, "Insufficient permissions to read the ~s file", [AclFile]),
#{};
{error, enoent} ->
?LOG(alert, "The ~s file does not exist", [AclFile]),
#{};
{error, Reason} ->
?LOG(alert, "Failed to read ~s: ~p", [AclFile, Reason]),
#{}
end.
filter(_PubSub, {allow, all}) ->
true;
filter(_PubSub, {deny, all}) ->
true;
filter(publish, {_AllowDeny, _Who, publish, _Topics}) ->
true;
filter(_PubSub, {_AllowDeny, _Who, pubsub, _Topics}) ->
true;
filter(subscribe, {_AllowDeny, _Who, subscribe, _Topics}) ->
true;
filter(_PubSub, {_AllowDeny, _Who, _, _Topics}) ->
false.