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

fix(authz): write acl and cert files after nodes config synced

This commit is contained in:
JimMoen 2022-03-22 03:14:47 +08:00
parent c2e1c38fdf
commit 41c808a262
4 changed files with 146 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

175
apps/emqx_authz/src/emqx_authz.erl
View File

@ -128,41 +128,34 @@ update(Cmd, Sources) ->
pre_config_update(_, Cmd, Sources) ->
{ok, do_pre_config_update(Cmd, Sources)}.
do_pre_config_update({?CMD_MOVE, Type, ?CMD_MOVE_FRONT}, Sources) ->
{Source, Front, Rear} = take(Type, Sources),
[Source | Front] ++ Rear;
do_pre_config_update({?CMD_MOVE, Type, ?CMD_MOVE_REAR}, Sources) ->
{Source, Front, Rear} = take(Type, Sources),
Front ++ Rear ++ [Source];
do_pre_config_update({?CMD_MOVE, Type, ?CMD_MOVE_BEFORE(Before)}, Sources) ->
{S1, Front1, Rear1} = take(Type, Sources),
{S2, Front2, Rear2} = take(Before, Front1 ++ Rear1),
Front2 ++ [S1, S2] ++ Rear2;
do_pre_config_update({?CMD_MOVE, Type, ?CMD_MOVE_AFTER(After)}, Sources) ->
{S1, Front1, Rear1} = take(Type, Sources),
{S2, Front2, Rear2} = take(After, Front1 ++ Rear1),
Front2 ++ [S2, S1] ++ Rear2;
do_pre_config_update({?CMD_PREPEND, NewSource}, Sources) ->
NSources = [NewSource] ++ Sources,
do_pre_config_update({?CMD_MOVE, _, _} = Cmd, Sources) ->
do_move(Cmd, Sources);
do_pre_config_update({?CMD_PREPEND, Source}, Sources) ->
NSource = maybe_write_files(Source),
NSources = [NSource] ++ Sources,
ok = check_dup_types(NSources),
NSources;
do_pre_config_update({?CMD_APPEND, NewSource}, Sources) ->
NSources = Sources ++ [NewSource],
do_pre_config_update({?CMD_APPEND, Source}, Sources) ->
NSource = maybe_write_files(Source),
NSources = Sources ++ [NSource],
ok = check_dup_types(NSources),
NSources;
do_pre_config_update({{?CMD_REPLACE, Type}, #{<<"enable">> := Enable} = Source}, Sources)
when ?IS_ENABLED(Enable) ->
case create_dry_run(Type, Source) of
ok ->
NSource = maybe_write_files(Source),
{_Old, Front, Rear} = take(Type, Sources),
NSources = Front ++ [Source | Rear],
case create_dry_run(Type, NSource) of
ok ->
NSources = Front ++ [NSource | Rear],
ok = check_dup_types(NSources),
NSources;
{error, _} = Error -> Error
{error, _} = Error ->
throw(Error)
end;
do_pre_config_update({{?CMD_REPLACE, Type}, Source}, Sources) ->
NSource = maybe_write_files(Source),
{_Old, Front, Rear} = take(Type, Sources),
NSources = Front ++ [Source | Rear],
NSources = Front ++ [NSource | Rear],
ok = check_dup_types(NSources),
NSources;
do_pre_config_update({{?CMD_DELETE, Type}, _Source}, Sources) ->
@ -171,50 +164,64 @@ do_pre_config_update({{?CMD_DELETE, Type}, _Source}, Sources) ->
NSources;
do_pre_config_update({?CMD_REPLACE, Sources}, _OldSources) ->
%% overwrite the entire config!
Sources;
NSources = lists:map(fun maybe_write_files/1, Sources),
ok = check_dup_types(NSources),
NSources;
do_pre_config_update({Op, Source}, Sources) ->
error({bad_request, #{op => Op, source => Source, sources => Sources}}).
throw({bad_request, #{op => Op, source => Source, sources => Sources}}).
post_config_update(_, _, undefined, _OldSource, _AppEnvs) ->
ok;
post_config_update(_, Cmd, NewSources, _OldSource, _AppEnvs) ->
ok = do_post_config_update(Cmd, NewSources),
Actions = do_post_config_update(Cmd, NewSources),
ok = update_authz_chain(Actions),
ok = emqx_authz_cache:drain_cache().
do_post_config_update({?CMD_MOVE, _Type, _Where} = Cmd, _NewSources) ->
do_post_config_update({?CMD_MOVE, _Type, _Where} = Cmd, _Sources) ->
InitedSources = lookup(),
MovedSources = do_pre_config_update(Cmd, InitedSources),
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [MovedSources]}, -1),
ok = emqx_authz_cache:drain_cache();
do_post_config_update({?CMD_PREPEND, Source}, _NewSources) ->
InitedSources = init_sources(check_sources([Source])),
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [InitedSources ++ lookup()]}, -1),
ok = emqx_authz_cache:drain_cache();
do_post_config_update({?CMD_APPEND, Source}, _NewSources) ->
InitedSources = init_sources(check_sources([Source])),
emqx_hooks:put('client.authorize', {?MODULE, authorize, [lookup() ++ InitedSources]}, -1),
ok = emqx_authz_cache:drain_cache();
do_post_config_update({{?CMD_REPLACE, Type}, Source}, _NewSources) when is_map(Source) ->
do_move(Cmd, InitedSources);
do_post_config_update({?CMD_PREPEND, RawNewSource}, Sources) ->
InitedNewSource = init_source(get_source_by_type(type(RawNewSource), Sources)),
[InitedNewSource] ++ lookup();
do_post_config_update({?CMD_APPEND, RawNewSource}, Sources) ->
InitedNewSource = init_source(get_source_by_type(type(RawNewSource), Sources)),
lookup() ++ [InitedNewSource];
do_post_config_update({{?CMD_REPLACE, Type}, RawNewSource}, Sources) ->
OldSources = lookup(),
{OldSource, Front, Rear} = take(Type, OldSources),
NewSource = get_source_by_type(type(RawNewSource), Sources),
ok = ensure_resource_deleted(OldSource),
clear_certs(OldSource),
InitedSources = init_source(NewSource),
Front ++ [InitedSources] ++ Rear;
do_post_config_update({{?CMD_DELETE, Type}, _RawNewSource}, _Sources) ->
OldInitedSources = lookup(),
{OldSource, Front, Rear} = take(Type, OldInitedSources),
ok = ensure_resource_deleted(OldSource),
InitedSources = init_sources(check_sources([Source])),
ok = emqx_hooks:put( 'client.authorize'
, {?MODULE, authorize, [Front ++ InitedSources ++ Rear]}, -1),
ok = emqx_authz_cache:drain_cache();
do_post_config_update({{?CMD_DELETE, Type}, _Source}, _NewSources) ->
OldInitedSources = lookup(),
{OldSource, Front, Rear} = take(Type, OldInitedSources),
ok = ensure_resource_deleted(OldSource),
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [Front ++ Rear]}, -1),
ok = emqx_authz_cache:drain_cache();
do_post_config_update({?CMD_REPLACE, Sources}, _NewSources) ->
clear_certs(OldSource),
Front ++ Rear;
do_post_config_update({?CMD_REPLACE, _RawNewSources}, Sources) ->
%% overwrite the entire config!
OldInitedSources = lookup(),
InitedSources = init_sources(check_sources(Sources)),
ok = emqx_hooks:put('client.authorize', {?MODULE, authorize, [InitedSources]}, -1),
lists:foreach(fun ensure_resource_deleted/1, OldInitedSources),
ok = emqx_authz_cache:drain_cache().
lists:foreach(fun clear_certs/1, OldInitedSources),
init_sources(Sources).
%% @doc do source move
do_move({?CMD_MOVE, Type, ?CMD_MOVE_FRONT}, Sources) ->
{Source, Front, Rear} = take(Type, Sources),
[Source | Front] ++ Rear;
do_move({?CMD_MOVE, Type, ?CMD_MOVE_REAR}, Sources) ->
{Source, Front, Rear} = take(Type, Sources),
Front ++ Rear ++ [Source];
do_move({?CMD_MOVE, Type, ?CMD_MOVE_BEFORE(Before)}, Sources) ->
{S1, Front1, Rear1} = take(Type, Sources),
{S2, Front2, Rear2} = take(Before, Front1 ++ Rear1),
Front2 ++ [S1, S2] ++ Rear2;
do_move({?CMD_MOVE, Type, ?CMD_MOVE_AFTER(After)}, Sources) ->
{S1, Front1, Rear1} = take(Type, Sources),
{S2, Front2, Rear2} = take(After, Front1 ++ Rear1),
Front2 ++ [S2, S1] ++ Rear2.
ensure_resource_deleted(#{enable := false}) -> ok;
ensure_resource_deleted(#{type := Type} = Source) ->
@ -231,14 +238,14 @@ check_dup_types([Source | Sources], Checked) ->
Type = case maps:get(<<"type">>, Source, maps:get(type, Source, undefined)) of
undefined ->
%% this should never happen if the value is type checked by honcon schema
error({bad_source_input, Source});
throw({bad_source_input, Source});
Type0 ->
type(Type0)
end,
case lists:member(Type, Checked) of
true ->
%% we have made it clear not to support more than one authz instance for each type
error({duplicated_authz_source_type, Type});
throw({duplicated_authz_source_type, Type});
false ->
check_dup_types(Sources, [Type | Checked])
end.
@ -330,7 +337,7 @@ take(Type, Sources) ->
{Front, Rear} = lists:splitwith(fun(T) -> type(T) =/= type(Type) end, Sources),
case Rear =:= [] of
true ->
error({not_found_source, Type});
throw({not_found_source, Type});
_ ->
{hd(Rear), Front, tl(Rear)}
end.
@ -362,8 +369,62 @@ type(<<"postgresql">>) -> postgresql;
type('built_in_database') -> 'built_in_database';
type(<<"built_in_database">>) -> 'built_in_database';
%% should never happen if the input is type-checked by hocon schema
type(Unknown) -> error({unknown_authz_source_type, Unknown}).
type(Unknown) -> throw({unknown_authz_source_type, Unknown}).
maybe_write_files(#{<<"type">> := <<"file">>} = Source) ->
write_acl_file(Source);
maybe_write_files(NewSource) ->
maybe_write_certs(NewSource).
write_acl_file(#{<<"rules">> := Rules} = Source) ->
NRules = check_acl_file_rules(Rules),
Path = acl_conf_file(),
{ok, _Filename} = write_file(Path, NRules),
maps:without([<<"rules">>], Source#{<<"path">> => Path}).
%% @doc where the acl.conf file is stored.
acl_conf_file() ->
filename:join([emqx:data_dir(), "authz", "acl.conf"]).
maybe_write_certs(#{<<"type">> := Type} = Source) ->
case emqx_tls_lib:ensure_ssl_files(
ssl_file_path(Type), maps:get(<<"ssl">>, Source, undefined)) of
{ok, SSL} ->
new_ssl_source(Source, SSL);
{error, Reason} ->
?SLOG(error, Reason#{msg => "bad_ssl_config"}),
throw({bad_ssl_config, Reason})
end.
clear_certs(OldSource) ->
OldSSL = maps:get(ssl, OldSource, undefined),
ok = emqx_tls_lib:delete_ssl_files(ssl_file_path(type(OldSource)), undefined, OldSSL).
write_file(Filename, Bytes) ->
ok = filelib:ensure_dir(Filename),
case file:write_file(Filename, Bytes) of
ok -> {ok, iolist_to_binary(Filename)};
{error, Reason} ->
?SLOG(error, #{filename => Filename, msg => "write_file_error", reason => Reason}),
throw(Reason)
end.
ssl_file_path(Type) ->
filename:join(["authz", Type]).
new_ssl_source(Source, undefined) ->
Source;
new_ssl_source(Source, SSL) ->
Source#{<<"ssl">> => SSL}.
get_source_by_type(Type, Sources) ->
{Source, _Front, _Rear} = take(Type, Sources),
Source.
%% @doc put hook with (maybe) initialized new source and old sources
update_authz_chain(Actions) ->
emqx_hooks:put('client.authorize', {?MODULE, authorize, [Actions]}, -1).
check_acl_file_rules(RawRules) ->
%% TODO: make sure the bin rules checked
RawRules.

5
apps/emqx_authz/src/emqx_authz_api_schema.erl
View File

@ -66,12 +66,11 @@ fields(redis_cluster) ->
fields(file) ->
authz_common_fields(file)
++ [ { rules, #{ type => binary()
, required => true
, example =>
<<"{allow,{username,\"^dashboard?\"},","subscribe,[\"$SYS/#\"]}.\n",
"{allow,{ipaddr,\"127.0.0.1\"},all,[\"$SYS/#\",\"#\"]}.">>}}
%% The path will be deprecated, `acl.conf` will be fixed in subdir of `data`
, {path, #{ type => binary()
, example => <<"acl.conf">>}}];
];
fields(position) ->
[ { position
, mk( string()

83
apps/emqx_authz/src/emqx_authz_api_sources.erl
View File

@ -27,8 +27,6 @@
-define(BAD_REQUEST, 'BAD_REQUEST').
-define(NOT_FOUND, 'NOT_FOUND').
-define(IS_TRUE(Val), ((Val =:= true) or (Val =:= <<"true">>))).
-define(API_SCHEMA_MODULE, emqx_authz_api_schema).
-export([ get_raw_sources/0
@ -168,18 +166,10 @@ sources(get, _) ->
lists:append(AccIn, [read_certs(Source)])
end, [], get_raw_sources()),
{200, #{sources => Sources}};
sources(post, #{body := #{<<"type">> := <<"file">>, <<"rules">> := Rules}}) ->
{ok, Filename} = write_file(acl_conf_file(), Rules),
update_config(?CMD_PREPEND, #{<<"type">> => <<"file">>,
<<"enable">> => true, <<"path">> => Filename});
sources(post, #{body := Body}) when is_map(Body) ->
case maybe_write_certs(Body) of
Config when is_map(Config) ->
update_config(?CMD_PREPEND, Config);
{error, Reason} ->
{400, #{code => <<"BAD_REQUEST">>,
message => bin(Reason)}}
end.
sources(post, #{body := #{<<"type">> := <<"file">>} = Body}) ->
create_authz_file(Body);
sources(post, #{body := Body}) ->
update_config(?CMD_PREPEND, Body).
source(Method, #{bindings := #{type := Type} = Bindings } = Req)
when is_atom(Type) ->
@ -201,13 +191,10 @@ source(get, #{bindings := #{type := Type}}) ->
end;
[Source] -> {200, read_certs(Source)}
end;
source(put, #{bindings := #{type := <<"file">>}, body := #{<<"type">> := <<"file">>,
<<"rules">> := Rules,
<<"enable">> := Enable}}) ->
{ok, Filename} = write_file(maps:get(path, emqx_authz:lookup(file), ""), Rules),
case emqx_authz:update({?CMD_REPLACE, <<"file">>}, #{<<"type">> => <<"file">>,
<<"enable">> => Enable,
<<"path">> => Filename}) of
source(put, #{bindings := #{type := <<"file">>}, body := #{<<"type">> := <<"file">>} = Body}) ->
update_authz_file(Body);
source(put, #{bindings := #{type := Type}, body := Body}) ->
case emqx_authz:update({?CMD_REPLACE, Type}, Body) of
{ok, _} -> {204};
{error, {emqx_conf_schema, _}} ->
{400, #{code => <<"BAD_REQUEST">>,
@ -216,14 +203,6 @@ source(put, #{bindings := #{type := <<"file">>}, body := #{<<"type">> := <<"file
{400, #{code => <<"BAD_REQUEST">>,
message => bin(Reason)}}
end;
source(put, #{bindings := #{type := Type}, body := Body}) when is_map(Body) ->
case maybe_write_certs(Body#{<<"type">> => Type}) of
Config when is_map(Config) ->
update_config({?CMD_REPLACE, Type}, Config);
{error, Reason} ->
{400, #{code => <<"BAD_REQUEST">>,
message => bin(Reason)}}
end;
source(delete, #{bindings := #{type := Type}}) ->
update_config({?CMD_DELETE, Type}, #{}).
@ -423,45 +402,13 @@ update_config(Cmd, Sources) ->
read_certs(#{<<"ssl">> := SSL} = Source) ->
case emqx_tls_lib:file_content_as_options(SSL) of
{error, Reason} ->
?SLOG(error, Reason#{msg => failed_to_readd_ssl_file}),
throw(failed_to_readd_ssl_file);
?SLOG(error, Reason#{msg => failed_to_read_ssl_file}),
throw(failed_to_read_ssl_file);
{ok, NewSSL} ->
Source#{<<"ssl">> => NewSSL}
end;
read_certs(Source) -> Source.
maybe_write_certs(#{<<"ssl">> := #{<<"enable">> := True} = SSL} = Source) when ?IS_TRUE(True) ->
Type = maps:get(<<"type">>, Source),
case emqx_tls_lib:ensure_ssl_files(filename:join(["authz", Type]), SSL) of
{ok, Return} ->
maps:put(<<"ssl">>, Return, Source);
{error, _} ->
{error, ensuer_ssl_files_failed}
end;
maybe_write_certs(Source) -> Source.
write_file(Filename, Bytes0) ->
ok = filelib:ensure_dir(Filename),
case file:read_file(Filename) of
{ok, Bytes1} ->
case crypto:hash(md5, Bytes1) =:= crypto:hash(md5, Bytes0) of
true -> {ok, iolist_to_binary(Filename)};
false -> do_write_file(Filename, Bytes0)
end;
_ -> do_write_file(Filename, Bytes0)
end.
do_write_file(Filename, Bytes) ->
case file:write_file(Filename, Bytes) of
ok -> {ok, iolist_to_binary(Filename)};
{error, Reason} ->
?SLOG(error, #{filename => Filename, msg => "write_file_error", reason => Reason}),
error(Reason)
end.
acl_conf_file() ->
emqx_authz:acl_conf_file().
parameters_field() ->
[ {type, mk( enum(?API_SCHEMA_MODULE:authz_sources_types(simple))
, #{in => path, desc => <<"Authorization type">>})
@ -528,3 +475,13 @@ status_metrics_example() ->
}
]
}.
create_authz_file(Body) ->
do_update_authz_file(?CMD_PREPEND, Body).
update_authz_file(Body) ->
do_update_authz_file({?CMD_REPLACE, <<"file">>}, Body).
do_update_authz_file(Cmd, Body) ->
%% API update will placed in `authz` subdirectory inside EMQX's `data_dir`
update_config(Cmd, Body).

1
apps/emqx_authz/src/emqx_authz_schema.erl
View File

@ -93,6 +93,7 @@ fields(file) ->
, {enable, #{type => boolean(),
default => true}}
, {path, #{type => string(),
required => true,
desc => """
Path to the file which contains the ACL rules.<br>
If the file provisioned before starting EMQX node,