fix(authz): authz http resource url query string

apps/emqx_authz/src/emqx_authz_http.erl

@@ -40,43 +40,28 @@
 description() ->
     "AuthZ with http".
 
-init(#{url := Url} = Source) ->
+init(Config) ->
-    NSource = maps:put(base_url, maps:remove(query, Url), Source),
+    NConfig = parse_config(Config),
-    case emqx_authz_utils:create_resource(emqx_connector_http, NSource) of
+    case emqx_authz_utils:create_resource(emqx_connector_http, NConfig) of
         {error, Reason} -> error({load_config_error, Reason});
-        {ok, Id} -> Source#{annotations => #{id => Id}}
+        {ok, Id} -> NConfig#{annotations => #{id => Id}}
     end.
 
 destroy(#{annotations := #{id := Id}}) ->
     ok = emqx_resource:remove_local(Id).
 
-dry_run(Source) ->
+dry_run(Config) ->
-    URIMap = maps:get(url, Source),
+    emqx_resource:create_dry_run_local(emqx_connector_http, parse_config(Config)).
-    NSource = maps:put(base_url, maps:remove(query, URIMap), Source),
-    emqx_resource:create_dry_run_local(emqx_connector_http, NSource).
 
-authorize(Client, PubSub, Topic,
+authorize( Client
-            #{type := http,
+         , PubSub
-              url := #{path := Path} = URL,
+         , Topic
-              headers := Headers,
+         , #{ type            := http
-              method := Method,
+            , annotations     := #{id := ResourceID}
-              request_timeout := RequestTimeout,
+            , method := Method
-              annotations := #{id := ResourceID}
+            , request_timeout := RequestTimeout
-             } = Source) ->
+            } = Config) ->
-    Request = case Method of
+    Request = generate_request(PubSub, Topic, Client, Config),
-                  get  ->
-                      Query = maps:get(query, URL, ""),
-                      Path1 = replvar(Path ++ "?" ++ Query, PubSub, Topic, Client),
-                      {Path1, maps:to_list(Headers)};
-                  _ ->
-                      Body0 = serialize_body(
-                                maps:get('Accept', Headers, <<"application/json">>),
-                                maps:get(body, Source, #{})
-                              ),
-                      Body1 = replvar(Body0, PubSub, Topic, Client),
-                      Path1 = replvar(Path, PubSub, Topic, Client),
-                      {Path1, maps:to_list(Headers), Body1}
-              end,
     case emqx_resource:query(ResourceID, {Method, Request, RequestTimeout}) of
         {ok, 200, _Headers} ->
             {matched, allow};
@@ -84,6 +69,8 @@ authorize(Client, PubSub, Topic,
             {matched, allow};
         {ok, 200, _Headers, _Body} ->
             {matched, allow};
+        {ok, _Status, _Headers} ->
+            nomatch;
         {ok, _Status, _Headers, _Body} ->
             nomatch;
         {error, Reason} ->
@@ -93,6 +80,24 @@ authorize(Client, PubSub, Topic,
             ignore
     end.
 
+parse_config(#{ url := URL
+              , method := Method
+              , headers := Headers
+              , request_timeout := ReqTimeout
+              } = Conf) ->
+    {BaseURLWithPath, Query} = parse_fullpath(URL),
+    BaseURLMap = parse_url(BaseURLWithPath),
+    Conf#{ method          => Method
+         , base_url        => maps:remove(query, BaseURLMap)
+         , base_query      => cow_qs:parse_qs(bin(Query))
+         , body            => maps:get(body, Conf, #{})
+         , headers         => Headers
+         , request_timeout => ReqTimeout
+         }.
+
+parse_fullpath(RawURL) ->
+    cow_http:parse_fullpath(bin(RawURL)).
+
 parse_url(URL)
   when URL =:= undefined ->
     #{};
@@ -105,12 +110,45 @@ parse_url(URL) ->
             URIMap
     end.
 
+generate_request( PubSub
+                , Topic
+                , Client
+                , #{ method := Method
+                   , base_url := #{path := Path}
+                   , base_query := BaseQuery
+                   , headers := Headers
+                   , body := Body0
+                   }) ->
+    Body = replace_placeholders(maps:to_list(Body0), PubSub, Topic, Client),
+    NBaseQuery = replace_placeholders(BaseQuery, PubSub, Topic, Client),
+    case Method of
+        get  ->
+            NPath = append_query(Path, NBaseQuery ++ Body),
+            {NPath, maps:to_list(Headers)};
+        _ ->
+            NPath = append_query(Path, NBaseQuery),
+            NBody = serialize_body(
+                      maps:get(<<"Accept">>, Headers, <<"application/json">>),
+                      Body
+                     ),
+            {NPath, maps:to_list(Headers), NBody}
+    end.
+
+append_query(Path, []) ->
+    Path;
+append_query(Path, Query) ->
+    Path ++ "?" ++ binary_to_list(query_string(Query)).
+
 query_string(Body) ->
-    query_string(maps:to_list(Body), []).
+    query_string(Body, []).
 
 query_string([], Acc) ->
-    <<$&, Str/binary>> = iolist_to_binary(lists:reverse(Acc)),
+    case iolist_to_binary(lists:reverse(Acc)) of
-    Str;
+        <<$&, Str/binary>> ->
+            Str;
+        <<>> ->
+            <<>>
+    end;
 query_string([{K, V} | More], Acc) ->
     query_string( More
                 , [ ["&", emqx_http_lib:uri_encode(K), "=", emqx_http_lib:uri_encode(V)]
@@ -121,30 +159,34 @@ serialize_body(<<"application/json">>, Body) ->
 serialize_body(<<"application/x-www-form-urlencoded">>, Body) ->
     query_string(Body).
 
-replvar(Str0, PubSub, Topic,
+replace_placeholders(KVs, PubSub, Topic, Client) ->
-        #{username := Username,
+    replace_placeholders(KVs, PubSub, Topic, Client, []).
-          clientid := Clientid,
+
-          peerhost := IpAddress,
+replace_placeholders([], _PubSub, _Topic, _Client, Acc) ->
-          protocol := Protocol,
+    lists:reverse(Acc);
-          mountpoint := Mountpoint
+replace_placeholders([{K, V0} | More], PubSub, Topic, Client, Acc) ->
-         }) when is_list(Str0);
+    case replace_placeholder(V0, PubSub, Topic, Client) of
-                 is_binary(Str0) ->
+        undefined ->
-    NTopic = emqx_http_lib:uri_encode(Topic),
+            error({cannot_get_variable, V0});
-    Str1 = re:replace( Str0, emqx_authz:ph_to_re(?PH_S_CLIENTID)
+        V ->
-                     , bin(Clientid), [global, {return, binary}]),
+            replace_placeholders(More, PubSub, Topic, Client, [{bin(K), bin(V)} | Acc])
-    Str2 = re:replace( Str1, emqx_authz:ph_to_re(?PH_S_USERNAME)
+    end.
-                     , bin(Username), [global, {return, binary}]),
+
-    Str3 = re:replace( Str2, emqx_authz:ph_to_re(?PH_S_HOST)
+replace_placeholder(?PH_USERNAME, _PubSub, _Topic, Client) ->
-                     , inet_parse:ntoa(IpAddress), [global, {return, binary}]),
+    bin(maps:get(username, Client, undefined));
-    Str4 = re:replace( Str3, emqx_authz:ph_to_re(?PH_S_PROTONAME)
+replace_placeholder(?PH_CLIENTID, _PubSub, _Topic, Client) ->
-                     , bin(Protocol), [global, {return, binary}]),
+    bin(maps:get(clientid, Client, undefined));
-    Str5 = re:replace( Str4, emqx_authz:ph_to_re(?PH_S_MOUNTPOINT)
+replace_placeholder(?PH_HOST,     _PubSub, _Topic, Client) ->
-                     , bin(Mountpoint), [global, {return, binary}]),
+    inet_parse:ntoa(maps:get(peerhost, Client, undefined));
-    Str6 = re:replace( Str5, emqx_authz:ph_to_re(?PH_S_TOPIC)
+replace_placeholder(?PH_PROTONAME, _PubSub, _Topic, Client) ->
-                     , bin(NTopic), [global, {return, binary}]),
+    bin(maps:get(protocol, Client, undefined));
-    Str7 = re:replace( Str6, emqx_authz:ph_to_re(?PH_S_ACTION)
+replace_placeholder(?PH_TOPIC, _PubSub, Topic, _Client) ->
-                     , bin(PubSub), [global, {return, binary}]),
+    bin(emqx_http_lib:uri_encode(Topic));
-    Str7.
+replace_placeholder(?PH_ACTION, PubSub, _Topic, _Client) ->
+    bin(PubSub);
+
+replace_placeholder(Constant, _, _, _) ->
+    Constant.
 
 bin(A) when is_atom(A) -> atom_to_binary(A, utf8);
 bin(B) when is_binary(B) -> B;

apps/emqx_authz/src/emqx_authz_schema.erl

@@ -17,17 +17,14 @@
 -module(emqx_authz_schema).
 
 -include_lib("typerefl/include/types.hrl").
+-include_lib("emqx_connector/include/emqx_connector.hrl").
 
 -reflect_type([ permission/0
               , action/0
-              , url/0
               ]).
 
--typerefl_from_string({url/0, emqx_http_lib, uri_parse}).
-
 -type action() :: publish | subscribe | all.
 -type permission() :: allow | deny.
--type url() :: emqx_http_lib:uri_map().
 
 -export([ namespace/0
         , roots/0
@@ -143,7 +140,7 @@ fields(redis_cluster) ->
 http_common_fields() ->
     [ {type,            #{type => http}}
     , {enable,          #{type => boolean(), default => true}}
-    , {url,             #{type => url()}}
+    , {url,             fun url/1}
     , {request_timeout, mk_duration("request timeout", #{default => "30s"})}
     , {body,            #{type => map(), nullable => true}}
     ] ++ proplists:delete(base_url, emqx_connector_http:fields(config)).
@@ -177,6 +174,11 @@ headers_no_content_type(converter) ->
 headers_no_content_type(default) -> default_headers_no_content_type();
 headers_no_content_type(_) -> undefined.
 
+url(type) -> binary();
+url(validator) -> [?NOT_EMPTY("the value of the field 'url' cannot be empty")];
+url(nullable) -> false;
+url(_) -> undefined.
+
 %%--------------------------------------------------------------------
 %% Internal functions
 %%--------------------------------------------------------------------