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

fix(postgres): fix usage by authn/authz modules when prepared statements are disabled 

Fixes https://emqx.atlassian.net/browse/EMQX-12550
This commit is contained in:
Thales Macedo Garitezi 2024-06-14 12:19:46 -03:00
parent fd5e844cc1
commit 6e702cd16c
3 changed files with 108 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
apps/emqx_auth_postgresql/test/emqx_authn_postgresql_SUITE.erl
View File

@ -30,15 +30,28 @@
-define(PATH, [authentication]).
-import(emqx_common_test_helpers, [on_exit/1]).
all() ->
AllTCs = emqx_common_test_helpers:all(?MODULE),
TCs = AllTCs -- require_seeds_tests(),
[
{group, require_seeds},
t_update_with_invalid_config,
t_update_with_bad_config_value
{group, require_seeds}
| TCs
].
groups() ->
[{require_seeds, [], [t_create, t_authenticate, t_update, t_destroy, t_is_superuser]}].
[{require_seeds, [], require_seeds_tests()}].
require_seeds_tests() ->
[
t_create,
t_authenticate,
t_authenticate_disabled_prepared_statements,
t_update,
t_destroy,
t_is_superuser
].
init_per_testcase(_, Config) ->
emqx_authn_test_lib:delete_authenticators(
@ -47,6 +60,10 @@ init_per_testcase(_, Config) ->
),
Config.
end_per_testcase(_TestCase, _Config) ->
emqx_common_test_helpers:call_janitor(),
ok.
init_per_group(require_seeds, Config) ->
ok = init_seeds(),
Config.
@ -70,7 +87,12 @@ init_per_suite(Config) ->
),
[{apps, Apps} | Config];
false ->
{skip, no_pgsql}
case os:getenv("IS_CI") of
"yes" ->
throw(no_postgres);
_ ->
{skip, no_postgres}
end
end.
end_per_suite(Config) ->
@ -174,6 +196,25 @@ test_user_auth(#{
?GLOBAL
).
t_authenticate_disabled_prepared_statements(Config) ->
ResConfig = maps:merge(pgsql_config(), #{disable_prepared_statements => true}),
{ok, _} = emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, ResConfig),
on_exit(fun() ->
emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, pgsql_config())
end),
ok = lists:foreach(
fun(Sample0) ->
Sample = maps:update_with(
config_params,
fun(Cfg) -> Cfg#{<<"disable_prepared_statements">> => true} end,
Sample0
),
ct:pal("test_user_auth sample: ~p", [Sample]),
test_user_auth(Sample)
end,
user_seeds()
).
t_destroy(_Config) ->
AuthConfig = raw_pgsql_auth_config(),

4
apps/emqx_bridge_pgsql/test/emqx_bridge_pgsql_SUITE.erl
View File

@ -601,7 +601,7 @@ t_simple_sql_query(Config) ->
{ok, _},
create_bridge(Config)
),
Request = {sql, <<"SELECT count(1) AS T">>},
Request = {query, <<"SELECT count(1) AS T">>},
Result =
case QueryMode of
sync ->
@ -651,7 +651,7 @@ t_bad_sql_parameter(Config) ->
{ok, _},
create_bridge(Config)
),
Request = {sql, <<"">>, [bad_parameter]},
Request = {query, <<"">>, [bad_parameter]},
Result =
case QueryMode of
sync ->

107
apps/emqx_postgresql/src/emqx_postgresql.erl
View File

@ -59,6 +59,9 @@
default_port => ?PGSQL_DEFAULT_PORT
}).
-type connector_resource_id() :: binary().
-type action_resource_id() :: binary().
-type template() :: {unicode:chardata(), emqx_template_sql:row_template()}.
-type state() ::
#{
@ -319,38 +322,40 @@ do_check_channel_sql(
on_get_channels(ResId) ->
emqx_bridge_v2:get_channels_for_connector(ResId).
on_query(InstId, {TypeOrKey, NameOrSQL}, State) ->
on_query(InstId, {TypeOrKey, NameOrSQL, []}, State);
-spec on_query
%% Called from authn and authz modules
(connector_resource_id(), {prepared_query, binary(), [term()]}, state()) ->
{ok, _} | {error, term()};
%% Called from bridges
(connector_resource_id(), {action_resource_id(), map()}, state()) ->
{ok, _} | {error, term()}.
on_query(InstId, {TypeOrKey, NameOrMap}, State) ->
on_query(InstId, {TypeOrKey, NameOrMap, []}, State);
on_query(
InstId,
{TypeOrKey, NameOrSQL, Params},
{TypeOrKey, NameOrMap, Params},
#{pool_name := PoolName} = State
) ->
?SLOG(debug, #{
msg => "postgresql_connector_received_sql_query",
connector => InstId,
type => TypeOrKey,
sql => NameOrSQL,
sql => NameOrMap,
state => State
}),
Type = pgsql_query_type(TypeOrKey, State),
{NameOrSQL2, Data} = proc_sql_params(TypeOrKey, NameOrSQL, Params, State),
Res = on_sql_query(TypeOrKey, InstId, PoolName, Type, NameOrSQL2, Data),
{QueryType, NameOrSQL2, Data} = proc_sql_params(TypeOrKey, NameOrMap, Params, State),
emqx_trace:rendered_action_template(
TypeOrKey,
#{
statement_type => QueryType,
statement_or_name => NameOrSQL2,
data => Data
}
),
Res = on_sql_query(InstId, PoolName, QueryType, NameOrSQL2, Data),
?tp(postgres_bridge_connector_on_query_return, #{instance_id => InstId, result => Res}),
handle_result(Res).
pgsql_query_type(_TypeOrTag, #{prepares := disabled}) ->
query;
pgsql_query_type(sql, _ConnectorState) ->
query;
pgsql_query_type(query, _ConnectorState) ->
query;
pgsql_query_type(prepared_query, _ConnectorState) ->
prepared_query;
%% for bridge
pgsql_query_type(_, ConnectorState) ->
pgsql_query_type(prepared_query, ConnectorState).
on_batch_query(
InstId,
[{Key, _} = Request | _] = BatchReq,
@ -370,7 +375,15 @@ on_batch_query(
{_Statement, RowTemplate} ->
StatementTemplate = get_templated_statement(BinKey, State),
Rows = [render_prepare_sql_row(RowTemplate, Data) || {_Key, Data} <- BatchReq],
case on_sql_query(Key, InstId, PoolName, execute_batch, StatementTemplate, Rows) of
emqx_trace:rendered_action_template(
Key,
#{
statement_type => execute_batch,
statement_or_name => StatementTemplate,
data => Rows
}
),
case on_sql_query(InstId, PoolName, execute_batch, StatementTemplate, Rows) of
{error, _Error} = Result ->
handle_result(Result);
{_Column, Results} ->
@ -386,25 +399,33 @@ on_batch_query(InstId, BatchReq, State) ->
}),
{error, {unrecoverable_error, invalid_request}}.
proc_sql_params(query, SQLOrKey, Params, _State) ->
{SQLOrKey, Params};
proc_sql_params(prepared_query, SQLOrKey, Params, _State) ->
{SQLOrKey, Params};
proc_sql_params(TypeOrKey, SQLOrData, Params, State) ->
proc_sql_params(ActionResId, #{} = Map, [], State) when is_binary(ActionResId) ->
%% When this connector is called from actions/bridges.
DisablePreparedStatements = maps:get(prepares, State, #{}) =:= disabled,
BinKey = to_bin(TypeOrKey),
case get_template(BinKey, State) of
undefined ->
{SQLOrData, Params};
{Statement, RowTemplate} ->
Rendered = render_prepare_sql_row(RowTemplate, SQLOrData),
case DisablePreparedStatements of
true ->
{Statement, Rendered};
false ->
{BinKey, Rendered}
end
end.
{ExprTemplate, RowTemplate} = get_template(ActionResId, State),
Rendered = render_prepare_sql_row(RowTemplate, Map),
case DisablePreparedStatements of
true ->
{query, ExprTemplate, Rendered};
false ->
{prepared_query, ActionResId, Rendered}
end;
proc_sql_params(prepared_query, ConnResId, Params, State) ->
%% When this connector is called from authn/authz modules
DisablePreparedStatements = maps:get(prepares, State, #{}) =:= disabled,
case DisablePreparedStatements of
true ->
#{query_templates := #{ConnResId := {ExprTemplate, _VarsTemplate}}} = State,
{query, ExprTemplate, Params};
false ->
%% Connector resource id itself is the prepared statement name
{prepared_query, ConnResId, Params}
end;
proc_sql_params(QueryType, SQL, Params, _State) when
is_atom(QueryType), is_binary(SQL) orelse is_list(SQL), is_list(Params)
->
%% When called to do ad-hoc commands/queries.
{QueryType, SQL, Params}.
get_template(Key, #{installed_channels := Channels} = _State) when is_map_key(Key, Channels) ->
BinKey = to_bin(Key),
@ -430,15 +451,7 @@ get_templated_statement(Key, #{prepares := PrepStatements}) ->
BinKey = to_bin(Key),
maps:get(BinKey, PrepStatements).
on_sql_query(Key, InstId, PoolName, Type, NameOrSQL, Data) ->
emqx_trace:rendered_action_template(
Key,
#{
statement_type => Type,
statement_or_name => NameOrSQL,
data => Data
}
),
on_sql_query(InstId, PoolName, Type, NameOrSQL, Data) ->
try ecpool:pick_and_do(PoolName, {?MODULE, Type, [NameOrSQL, Data]}, no_handover) of
{error, Reason} ->
?tp(