fix: ws fragment rest

src/nhttp_huffman.erl

@@ -63,7 +63,6 @@ decode(Data) ->
 %%%-----------------------------------------------------------------------------
 %% INTERNAL FUNCTIONS
 %%%-----------------------------------------------------------------------------
-
 -spec finalize_padding({ok, binary(), bitstring()} | {error, invalid_huffman}) ->
     {ok, binary()} | {error, invalid_huffman}.
 finalize_padding({ok, Decoded, <<>>}) ->

src/nhttp_lib.app.src

@@ -1,6 +1,6 @@
 {application, nhttp_lib, [
     {description, "HTTP protocol primitives for Erlang/OTP 27+ (HTTP/1.1, HTTP/2, HTTP/3, QPACK)"},
-    {vsn, "1.0.2"},
+    {vsn, "1.0.3"},
     {registered, []},
     {applications, [
         kernel,

src/nhttp_ws.erl

@@ -141,6 +141,7 @@ for backwards compatibility.
 
 -type stateful_decode_result() ::
     {ok, ws_message(), Rest :: binary(), ws_decoder()}
+    | {continue, Rest :: binary(), ws_decoder()}
     | {more, MinBytes :: pos_integer(), ws_decoder()}
     | {error, term()}.
 
@@ -198,7 +199,8 @@ for backwards compatibility.
     frag_opcode :: 0..15 | undefined,
     frag_acc = [] :: [binary()],
     frag_acc_size = 0 :: non_neg_integer(),
-    max_message_size = infinity :: pos_integer() | infinity
+    max_message_size = infinity :: pos_integer() | infinity,
+    utf8_carry = <<>> :: binary()
 }).
 
 -opaque ws_decoder() :: #ws_decoder{}.
@@ -406,6 +408,18 @@ Decode a frame with fragmentation support.
 Continuation frames are accumulated until FIN=1, then the complete
 message is delivered. Control frames (ping, pong, close) may appear
 between fragments and are delivered immediately.
+
+A successful return has three forms, and each one says what to keep:
+
+- `{ok, Message, Rest, Decoder}`: a frame was consumed and the message
+  is complete. Keep `Rest`.
+- `{continue, Rest, Decoder}`: a frame was consumed and buffered as a
+  non-final fragment. No message yet. Keep `Rest`.
+- `{more, MinBytes, Decoder}`: nothing was consumed. Keep the input
+  buffer and wait for `MinBytes` more bytes.
+
+If you keep the whole buffer after a consumed frame, the next call
+decodes that frame again and fails with `expected_continuation`.
 """.
 -spec decode_with_state(binary(), ws_decoder()) -> stateful_decode_result().
 decode_with_state(Data, #ws_decoder{role = Role} = Dec) ->
@@ -621,6 +635,10 @@ check_message_size(Size, #ws_decoder{max_message_size = Max}) when Size =< Max -
 check_message_size(_Size, _Dec) ->
     {error, message_too_large}.
 
+-spec fragmented_message(?OP_TEXT | ?OP_BINARY, binary()) -> ws_message().
+fragmented_message(?OP_TEXT, Payload) -> {text, Payload};
+fragmented_message(?OP_BINARY, Payload) -> {binary, Payload}.
+
 -spec get_websocket_key(nhttp_lib:headers()) -> {ok, binary()} | {error, missing_key}.
 get_websocket_key(Headers) ->
     case nhttp_headers:get(<<"sec-websocket-key">>, Headers) of
@@ -641,17 +659,19 @@ process_fragment(1, Opcode, Payload, Rest, #ws_decoder{frag_opcode = undefined}
         {error, _} = Err ->
             Err
     end;
-process_fragment(0, Opcode, Payload, _Rest, #ws_decoder{frag_opcode = undefined} = Dec) when
+process_fragment(0, Opcode, Payload, Rest, #ws_decoder{frag_opcode = undefined} = Dec) when
     Opcode =:= ?OP_TEXT; Opcode =:= ?OP_BINARY
 ->
     Size = byte_size(Payload),
-    case check_message_size(Size, Dec) of
-        ok ->
-            {more, 1, Dec#ws_decoder{
-                frag_opcode = Opcode, frag_acc = [Payload], frag_acc_size = Size
-            }};
-        {error, _} = Err ->
-            Err
+    maybe
+        ok ?= check_message_size(Size, Dec),
+        {ok, Carry} ?= scan_text(Opcode, Payload, <<>>),
+        {continue, Rest, Dec#ws_decoder{
+            frag_opcode = Opcode,
+            frag_acc = [Payload],
+            frag_acc_size = Size,
+            utf8_carry = Carry
+        }}
     end;
 process_fragment(
     _Fin, Opcode, Payload, Rest, #ws_decoder{frag_opcode = _FragOp} = Dec
@@ -664,15 +684,16 @@ process_fragment(
     0,
     ?OP_CONTINUATION,
     Payload,
-    _Rest,
+    Rest,
     #ws_decoder{frag_opcode = FragOp, frag_acc = Acc, frag_acc_size = AccSize} = Dec
 ) when FragOp =/= undefined ->
     NewSize = AccSize + byte_size(Payload),
-    case check_message_size(NewSize, Dec) of
-        ok ->
-            {more, 1, Dec#ws_decoder{frag_acc = [Payload | Acc], frag_acc_size = NewSize}};
-        {error, _} = Err ->
-            Err
+    maybe
+        ok ?= check_message_size(NewSize, Dec),
+        {ok, Carry} ?= scan_text(FragOp, Payload, Dec#ws_decoder.utf8_carry),
+        {continue, Rest, Dec#ws_decoder{
+            frag_acc = [Payload | Acc], frag_acc_size = NewSize, utf8_carry = Carry
+        }}
     end;
 process_fragment(
     1,
@@ -682,23 +703,26 @@ process_fragment(
     #ws_decoder{frag_opcode = FragOp, frag_acc = Acc, frag_acc_size = AccSize} = Dec
 ) when FragOp =/= undefined ->
     NewSize = AccSize + byte_size(Payload),
-    case check_message_size(NewSize, Dec) of
-        ok ->
-            FullPayload = iolist_to_binary(lists:reverse([Payload | Acc])),
-            case nhttp_ws_frame:opcode_to_complete_message(FragOp, FullPayload) of
-                {ok, Msg} ->
-                    {ok, Msg, Rest, Dec#ws_decoder{
-                        frag_opcode = undefined, frag_acc = [], frag_acc_size = 0
-                    }};
-                {error, _} = Err ->
-                    Err
-            end;
-        {error, _} = Err ->
-            Err
+    maybe
+        ok ?= check_message_size(NewSize, Dec),
+        {ok, <<>>} ?= scan_text(FragOp, Payload, Dec#ws_decoder.utf8_carry),
+        FullPayload = iolist_to_binary(lists:reverse([Payload | Acc])),
+        {ok, fragmented_message(FragOp, FullPayload), Rest, Dec#ws_decoder{
+            frag_opcode = undefined, frag_acc = [], frag_acc_size = 0, utf8_carry = <<>>
+        }}
+    else
+        {ok, _Truncated} -> {error, invalid_utf8};
+        {error, _} = Err -> Err
     end;
 process_fragment(_, _, _, _, _) ->
     {error, expected_continuation}.
 
+-spec scan_text(?OP_TEXT | ?OP_BINARY, binary(), binary()) ->
+    {ok, Carry :: binary()} | {error, invalid_utf8}.
+scan_text(?OP_TEXT, Payload, <<>>) -> nhttp_ws_frame:scan_utf8(Payload);
+scan_text(?OP_TEXT, Payload, Carry) -> nhttp_ws_frame:scan_utf8(Carry, Payload);
+scan_text(?OP_BINARY, _Payload, _Carry) -> {ok, <<>>}.
+
 -spec validate_connection_header(nhttp_lib:headers()) -> ok | {error, invalid_connection}.
 validate_connection_header(Headers) ->
     case nhttp_headers:get(<<"connection">>, Headers) of

src/nhttp_ws_frame.erl

@@ -23,6 +23,8 @@ masked (RFC 6455 §5.1).
     encode_masked/1,
     opcode_to_complete_message/2,
     opcode_to_message/3,
+    scan_utf8/1,
+    scan_utf8/2,
     validate_control_frame/3
 ]).
 
@@ -206,6 +208,49 @@ decode_unmasked(_) ->
 %%%-----------------------------------------------------------------------------
 %% SHARED HELPERS (USED BY STATEFUL MESSAGE-LEVEL DECODER)
 %%%-----------------------------------------------------------------------------
+-doc """
+Scan a run of text for UTF-8 validity (RFC 3629) and return the trailing
+bytes that do not yet form a character.
+
+A fragmented text message can split a character across two frames, so a
+fragment is not valid or invalid on its own: it ends in a carry, which the
+next fragment starts with. The run is refused as soon as no continuation
+can complete it, which puts the refusal on the fragment that breaks rather
+than on the whole reassembled message.
+
+A carry left at the end of a message is a truncated character, so the
+caller must require an empty one on the final fragment.
+""".
+-spec scan_utf8(binary()) -> {ok, Carry :: binary()} | {error, invalid_utf8}.
+scan_utf8(<<_/utf8, Rest/binary>>) ->
+    scan_utf8(Rest);
+scan_utf8(<<>>) ->
+    {ok, <<>>};
+scan_utf8(Rest) ->
+    case is_utf8_prefix(Rest) of
+        true -> {ok, Rest};
+        false -> {error, invalid_utf8}
+    end.
+
+-doc """
+Scan a run of text that continues an unfinished character.
+
+`Carry` comes from the previous fragment and is at most three bytes. Only
+the bytes needed to finish that character are copied; the rest of the
+payload is scanned where it lies.
+""".
+-spec scan_utf8(Carry :: binary(), binary()) -> {ok, Carry :: binary()} | {error, invalid_utf8}.
+scan_utf8(Carry, Payload) when byte_size(Payload) =< 3 ->
+    scan_utf8(<<Carry/binary, Payload/binary>>);
+scan_utf8(Carry, <<Head:3/binary, Rest/binary>>) ->
+    maybe
+        {ok, <<>>} ?= scan_utf8(<<Carry/binary, Head/binary>>),
+        scan_utf8(Rest)
+    else
+        {ok, NewCarry} -> scan_utf8(NewCarry, Rest);
+        {error, _} = Err -> Err
+    end.
+
 -doc """
 Map a complete (FIN=1) frame's opcode and payload to a `ws_message/0`.
 Text payloads are validated as UTF-8 (RFC 6455 §5.6 / §8.1).
@@ -443,6 +488,27 @@ is_valid_close_code(Code) when Code >= 1007, Code =< 1014 -> true;
 is_valid_close_code(Code) when Code >= 3000, Code =< 4999 -> true;
 is_valid_close_code(_) -> false.
 
+-doc """
+Whether the bytes are a proper prefix of a UTF-8 character, so a
+continuation can still complete them. The constraints on the second byte
+are the ones that make an overlong encoding (`E0 80`), a surrogate half
+(`ED A0`) and a scalar above U+10FFFF (`F4 90`) refusals here rather than
+one byte later.
+""".
+-spec is_utf8_prefix(binary()) -> boolean().
+is_utf8_prefix(<<B>>) when B >= 16#C2, B =< 16#F4 -> true;
+is_utf8_prefix(<<16#E0, B>>) when B >= 16#A0, B =< 16#BF -> true;
+is_utf8_prefix(<<16#ED, B>>) when B >= 16#80, B =< 16#9F -> true;
+is_utf8_prefix(<<A, B>>) when A >= 16#E1, A =< 16#EF, A =/= 16#ED, B >= 16#80, B =< 16#BF ->
+    true;
+is_utf8_prefix(<<16#F0, B>>) when B >= 16#90, B =< 16#BF -> true;
+is_utf8_prefix(<<16#F4, B>>) when B >= 16#80, B =< 16#8F -> true;
+is_utf8_prefix(<<A, B>>) when A >= 16#F1, A =< 16#F3, B >= 16#80, B =< 16#BF -> true;
+is_utf8_prefix(<<A, B, C>>) when A >= 16#F0, A =< 16#F4, C >= 16#80, C =< 16#BF ->
+    is_utf8_prefix(<<A, B>>);
+is_utf8_prefix(_) ->
+    false.
+
 -spec is_valid_utf8(binary()) -> boolean().
 is_valid_utf8(<<_/utf8, Rest/binary>>) -> is_valid_utf8(Rest);
 is_valid_utf8(<<>>) -> true;

test/nhttp_props_SUITE.erl

@@ -146,7 +146,8 @@ groups() ->
             ws_fragmentation_reassembly,
             ws_continuation_without_start_rejected,
             ws_new_message_mid_fragmentation_rejected,
-            ws_max_message_size_cumulative
+            ws_max_message_size_cumulative,
+            ws_incremental_utf8_matches_whole
         ]},
         {cookie_props, [parallel], [
             cookie_roundtrip,
@@ -488,6 +489,9 @@ ws_new_message_mid_fragmentation_rejected(Config) ->
 ws_max_message_size_cumulative(Config) ->
     run_property(nhttp_ws_props, prop_ws_max_message_size_cumulative, Config).
 
+ws_incremental_utf8_matches_whole(Config) ->
+    run_property(nhttp_ws_props, prop_ws_incremental_utf8_matches_whole, Config).
+
 %%%-----------------------------------------------------------------------------
 %%% COOKIE PROPERTY TESTS
 %%%-----------------------------------------------------------------------------

test/nhttp_ws_SUITE.erl

@@ -84,6 +84,12 @@ decoding, and stateful fragmentation for both server and client roles.
     frag_text/1,
     frag_binary/1,
     frag_with_control/1,
+    frag_coalesced_fragments/1,
+    frag_coalesced_with_control/1,
+    frag_text_invalid_utf8_fails_at_fragment/1,
+    frag_text_split_code_point/1,
+    frag_text_truncated_at_fin/1,
+    frag_binary_not_utf8_checked/1,
     frag_error_no_start/1,
     frag_server_decoder/1,
     frag_message_too_large_start/1,
@@ -193,6 +199,12 @@ groups() ->
             frag_text,
             frag_binary,
             frag_with_control,
+            frag_coalesced_fragments,
+            frag_coalesced_with_control,
+            frag_text_invalid_utf8_fails_at_fragment,
+            frag_text_split_code_point,
+            frag_text_truncated_at_fin,
+            frag_binary_not_utf8_checked,
             frag_error_no_start,
             frag_server_decoder,
             frag_message_too_large_start,
@@ -630,26 +642,63 @@ decode_unmasked_partial_ext64(_Config) ->
 frag_text(_Config) ->
     Dec0 = nhttp_ws:decoder_new(client),
     First = <<16#01, 3, "Hel">>,
-    {more, 1, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
     Cont = <<16#80, 2, "lo">>,
     {ok, {text, <<"Hello">>}, <<>>, _Dec2} = nhttp_ws:decode_with_state(Cont, Dec1).
 
 frag_binary(_Config) ->
     Dec0 = nhttp_ws:decoder_new(client),
     First = <<16#02, 2, "AB">>,
-    {more, 1, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
     Cont = <<16#80, 2, "CD">>,
     {ok, {binary, <<"ABCD">>}, <<>>, _Dec2} = nhttp_ws:decode_with_state(Cont, Dec1).
 
 frag_with_control(_Config) ->
     Dec0 = nhttp_ws:decoder_new(client),
     First = <<16#01, 3, "Hel">>,
-    {more, 1, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
     Ping = <<16#89, 0>>,
     {ok, ping, <<>>, Dec2} = nhttp_ws:decode_with_state(Ping, Dec1),
     Cont = <<16#80, 2, "lo">>,
     {ok, {text, <<"Hello">>}, <<>>, _Dec3} = nhttp_ws:decode_with_state(Cont, Dec2).
 
+frag_coalesced_fragments(_Config) ->
+    Dec0 = nhttp_ws:decoder_new(client),
+    Buffer = <<16#01, 3, "Hel", 16#80, 2, "lo">>,
+    {continue, Rest, Dec1} = nhttp_ws:decode_with_state(Buffer, Dec0),
+    ?assertEqual(<<16#80, 2, "lo">>, Rest),
+    {ok, {text, <<"Hello">>}, <<>>, _Dec2} = nhttp_ws:decode_with_state(Rest, Dec1).
+
+frag_coalesced_with_control(_Config) ->
+    Dec0 = nhttp_ws:decoder_new(client),
+    Buffer = <<16#01, 3, "Hel", 16#89, 0, 16#80, 2, "lo">>,
+    {continue, Rest1, Dec1} = nhttp_ws:decode_with_state(Buffer, Dec0),
+    {ok, ping, Rest2, Dec2} = nhttp_ws:decode_with_state(Rest1, Dec1),
+    {ok, {text, <<"Hello">>}, <<>>, _Dec3} = nhttp_ws:decode_with_state(Rest2, Dec2).
+
+frag_text_invalid_utf8_fails_at_fragment(_Config) ->
+    Dec0 = nhttp_ws:decoder_new(client),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(<<16#01, 3, "Hel">>, Dec0),
+    Broken = <<16#00, 2, 16#C0, 16#AF>>,
+    ?assertEqual({error, invalid_utf8}, nhttp_ws:decode_with_state(Broken, Dec1)).
+
+frag_text_split_code_point(_Config) ->
+    Dec0 = nhttp_ws:decoder_new(client),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(<<16#01, 1, 16#C3>>, Dec0),
+    Last = <<16#80, 1, 16#A9>>,
+    {ok, {text, <<16#C3, 16#A9>>}, <<>>, _Dec2} = nhttp_ws:decode_with_state(Last, Dec1).
+
+frag_text_truncated_at_fin(_Config) ->
+    Dec0 = nhttp_ws:decoder_new(client),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(<<16#01, 1, 16#C3>>, Dec0),
+    ?assertEqual({error, invalid_utf8}, nhttp_ws:decode_with_state(<<16#80, 0>>, Dec1)).
+
+frag_binary_not_utf8_checked(_Config) ->
+    Dec0 = nhttp_ws:decoder_new(client),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(<<16#02, 2, 16#C0, 16#AF>>, Dec0),
+    Last = <<16#80, 1, 16#FF>>,
+    {ok, {binary, <<16#C0, 16#AF, 16#FF>>}, <<>>, _Dec2} = nhttp_ws:decode_with_state(Last, Dec1).
+
 frag_error_no_start(_Config) ->
     Dec0 = nhttp_ws:decoder_new(client),
     NonFrag = <<16#81, 3, "ABC">>,
@@ -672,7 +721,7 @@ frag_message_too_large_start(_Config) ->
 frag_message_too_large_continuation(_Config) ->
     Dec0 = nhttp_ws:decoder_new(client, #{max_message_size => 5}),
     First = <<16#01, 3, "Hel">>,
-    {more, 1, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
+    {continue, <<>>, Dec1} = nhttp_ws:decode_with_state(First, Dec0),
     Cont = <<16#80, 5, "lothe">>,
     ?assertEqual({error, message_too_large}, nhttp_ws:decode_with_state(Cont, Dec1)).