fix(ft): restrict max filename length in transfers

Reject transfers with too long filenames right away, during `init`
handling, to avoid having to deal with filesystem errors later.

apps/emqx_ft/src/emqx_ft.erl

@@ -110,8 +110,8 @@ decode_filemeta(Map) when is_map(Map) ->
         Meta = hocon_tconf:check_plain(Schema, Map, #{atom_key => true, required => false}),
         {ok, Meta}
     catch
-        throw:Error ->
-            {error, {invalid_filemeta, Error}}
+        throw:{_Schema, Errors} ->
+            {error, {invalid_filemeta, Errors}}
     end.
 
 encode_filemeta(Meta = #{}) ->
@@ -381,7 +381,7 @@ do_validate([{filemeta, Payload} | Rest], Parsed) ->
         {ok, Meta} ->
             do_validate(Rest, [Meta | Parsed]);
         {error, Reason} ->
-            {error, {invalid_filemeta, Reason}}
+            {error, Reason}
     end;
 do_validate([{offset, Offset} | Rest], Parsed) ->
     case string:to_integer(Offset) of

apps/emqx_ft/src/emqx_ft_schema.erl

@@ -35,6 +35,13 @@
 
 -reflect_type([json_value/0]).
 
+%% NOTE
+%% This is rather conservative limit, mostly dictated by the filename limitations
+%% on most filesystems. Even though, say, S3 does not have such limitations, it's
+%% still useful to have a limit on the filename length, to avoid having to deal with
+%% limits in the storage backends.
+-define(MAX_FILENAME_BYTELEN, 255).
+
 -import(hoconsc, [ref/2, mk/2]).
 
 namespace() -> file_transfer.
@@ -234,7 +241,13 @@ schema(filemeta) ->
     }.
 
 validator(filename) ->
-    fun emqx_ft_fs_util:is_filename_safe/1.
+    [
+        fun(Value) ->
+            Bin = unicode:characters_to_binary(Value),
+            byte_size(Bin) =< ?MAX_FILENAME_BYTELEN orelse {error, max_length_exceeded}
+        end,
+        fun emqx_ft_fs_util:is_filename_safe/1
+    ].
 
 converter(checksum) ->
     fun

apps/emqx_ft/test/emqx_ft_SUITE.erl

@@ -183,9 +183,18 @@ t_invalid_filename(Config) ->
         unspecified_error,
         emqtt:publish(C, mk_init_topic(<<"f3">>), encode_meta(meta("/etc/passwd", <<>>)), 1)
     ),
+    ?assertRCName(
+        unspecified_error,
+        emqtt:publish(
+            C,
+            mk_init_topic(<<"f4">>),
+            encode_meta(meta(lists:duplicate(1000, $A), <<>>)),
+            1
+        )
+    ),
     ?assertRCName(
         success,
-        emqtt:publish(C, mk_init_topic(<<"f4">>), encode_meta(meta("146%", <<>>)), 1)
+        emqtt:publish(C, mk_init_topic(<<"f5">>), encode_meta(meta("146%", <<>>)), 1)
     ).
 
 t_simple_transfer(Config) ->