Repository: rabbitmq/rabbitmq-server (language: javascript)

Issue:
Strange behaviour when stream PublisherID is too long
### Describe the bug

Two bugs in one:
First:
PublisherID length not checked by stream plugin front end. It happily serializes it and PublisherID ends up in a `osiris_tracking:add` call which checks theTrkId lenght. And then crashes with function_clause. 

Also I get no slap from python client. It happily keeps running connection. While when say rabbit is dead or gen_batch_server is busy from breakpoints I get connection timeout

Second (probably to be moved to osiris directly):

I also got this when sending a smaller PublisherId of 256 bytes:

```
2024-10-09 13:29:44.217235+02:00 [debug] <0.1181.0> rabbit_stream_coordinator: running action: 'stop' for __hello-python-stream-small3_1728473315576504036 on node rabbit@arch in epoch 1874
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>     supervisor: {local,osiris_server_sup}
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>     errorContext: child_terminated
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>     reason: {function_clause,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                 [{osiris_tracking,parse_trailer,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                      [<<87,68,109,107,122,90,97,82,75,85,73,113,122,69,79,120,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         117,120,109,101,65,121,83,102,85,108,102,120,107,71,71,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         89,108,97,101,70,106,115,98,109,69,103,100,107,118,115,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         86,72,99,67,122,99,98,71,114,100,70,102,71,75,65,70,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         122,116,75,83,90,97,120,121,109,79,105,121,102,72,69,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         107,68,77,68,76,103,88,78,119,110,66,102,100,75,120,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         116,111,114,83,113,108,83,106,78,106,114,80,89,74,79,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         71,74,80,70,76,81,78,87,109,76,71,82,112,77,119,69,98,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         120,66,106,78,88,76,76,110,101,67,90,102,122,114,107,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         88,87,116,71,80,98,105,97,115,103,104,114,111,121,107,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         67,73,101,106,110,99,101,74,108,83,122,90,76,119,102,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         77,113,103,86,112,118,114,119,109,101,104,89,115,86,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         110,79,112,71,116,109,76,77,68,118,106,68,84,73,79,122,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         119,108,90,82,114,111,69,98,110,106,69,110,85,104,100,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         109,85,118,97,103,97,113,118,85,66,86,114,80,111,98,68,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         109,111,120,76,72,70,101,102,88,111,117,66,70,90,105,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                         87,121,48,0,0,0,0,0,0,0,1>>,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                       7,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                       {osiris_tracking,
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           {cfg,255},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           #{sequences => #{},offsets => #{},timestamps => #{}},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           #{<<>> => {7,6147522654068625751},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe0">> => {0,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe1">> => {1,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe2">> => {2,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe3">> => {3,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe4">> => {4,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe5">> => {5,1},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                             <<"qwe6">> => {6,1}},
2024-10-09 13:29:44.216542+02:00 [error] <0.149.0>                           #{},#{}}],

```


### Reproduction steps

```
iimport asyncio
import random
import string
from rstream import Producer

STREAM_NAME = "hello-python-stream-small3"
# 5GB
STREAM_RETENTION = 500

message =''.join(random.choices(string.ascii_letters, k=4))
message_bytes = message.encode()


async def send():
    
    i = 0
    while True:
        async with Producer(
                host="localhost",
                username="guest",
                password="guest",
        ) as producer:
            await producer.create_stream(
                STREAM_NAME, exists_ok=True, arguments={"max-length-bytes": STREAM_RETENTION}
            )

            #await producer.send(stream=STREAM_NAME, publisher_name="qwe{}".format(i), message=message_bytes)

            #await producer.send(stream=STREAM_NAME, publisher_name=''.join(random.choices(string.ascii_letters, k=4096)), message=message_bytes)
            #await producer.send(stream=STREAM_NAME, publisher_name="tvyOZOJWJEZdOPQRGLgJXbUQIxHDlMdaLIdpRCYViiNYvpTkyvNeXHHeDBHjVLzdEJoJNRyRVvFoDPaZQFBVyEwGkMvLbtppYkGRSvPJBClcFxdZrHSmpDNiZYdbiITtxoalsEPwYVIKYBVHnqnoUxRBUEbBXKhLHaEKVvJPdeunrcHuWwrFbdBxNEszlbMMflqLfUknSpWFanftkgDHvffuEajEnqqIZbtQAZXtqmEysubzHPILuubQaupmTykT", message=message_bytes)
            
            print(" [x] Hello, World! message {} sent".format(i))
            i += 1
            
            # input(" [x] Press Enter to close the producer  ...")
            await producer.close()

with asyncio.Runner() as runner:
    runner.run(send())

```

uncomment the third send (length of 256 and constant)  and then second to get writer restarted. restarting Rabbit also gives same result

### Expected behavior

Stream writer doesn't crash, Python client yells at me probably with proper errorcode

### Additional context

```
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>   crasher:
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     initial call: osiris_writer:init/1
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     pid: <0.902.0>
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     registered_name: []
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>     exception exit: {{case_clause,
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>                       {'EXIT',
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>                        {function_clause,
2024-10-09 13:06:02.600679+02:00 [error] <0.902.0>                         [{osiris_tracking,add,
````


Discussion / hints:

Fix patch:
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
index ffada5519745..357283cc8066 100644
--- a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
+++ b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -80,6 +80,10 @@
          peer_cert_validity]).
 -define(UNKNOWN_FIELD, unknown_field).
 -define(SILENT_CLOSE_DELAY, 3_000).
+-define(MAX_REFERENCE_SIZE, 255).
+
+-import(rabbit_stream_utils, [check_write_permitted/2,
+                              check_read_permitted/3]).
 
 %% client API
 -export([start_link/4,
@@ -1655,6 +1659,26 @@ handle_frame_post_auth(Transport,
               {C1#stream_connection{connection_step = failure}, S1}
       end,
     {Connection1, State1};
+handle_frame_post_auth(Transport,
+                       #stream_connection{user = User,
+                                          resource_alarm = false} = C,
+                       State,
+                       {request, CorrelationId,
+                        {declare_publisher, _PublisherId, WriterRef, S}})
+                      when is_binary(WriterRef), byte_size(WriterRef) > ?MAX_REFERENCE_SIZE ->
+  {Code, Counter} = case check_write_permitted(stream_r(S, C), User) of
+                      ok ->
+                        {?RESPONSE_CODE_PRECONDITION_FAILED, ?PRECONDITION_FAILED};
+                      error ->
+                        {?RESPONSE_CODE_ACCESS_REFUSED, ?ACCESS_REFUSED}
+                    end,
+  response(Transport,
+           C,
+           declare_publisher,
+           CorrelationId,
+           Code),
+  rabbit_global_counters:increase_protocol_counter(stream, Counter, 1),
+  {C, State};
 handle_frame_post_auth(Transport,
                        #stream_connection{user = User,
                                           publishers = Publishers0,
@@ -1664,7 +1688,7 @@ handle_frame_post_auth(Transport,
                        State,
                        {request, CorrelationId,
                         {declare_publisher, PublisherId, WriterRef, Stream}}) ->
-    case rabbit_stream_utils:check_write_permitted(stream_r(Stream,
+    case check_write_permitted(stream_r(Stream,
                                                             Connection0),
                                                    User)
     of
@@ -1895,6 +1919,19 @@ handle_frame_post_auth(Transport, #stream_connection{} = Connection, State,
                         {subscribe,
                          _, _, _, _, _}} = Request) ->
     handle_frame_post_auth(Transport, {ok, Connection}, State, Request);
+handle_frame_post_auth(Transport, {ok, #stream_connection{user = User} = C}, State,
+                       {request, CorrelationId,
+                        {subscribe, _, S, _, _, #{ <<"name">> := N}}})
+                      when is_binary(N), byte_size(N) > ?MAX_REFERENCE_SIZE ->
+  {Code, Counter} = case check_read_permitted(stream_r(S, C), User,#{}) of
+                      ok ->
+                        {?RESPONSE_CODE_PRECONDITION_FAILED, ?PRECONDITION_FAILED};
+                      error ->
+                        {?RESPONSE_CODE_ACCESS_REFUSED, ?ACCESS_REFUSED}
+                    end,
+  response(Transport, C, subscribe, CorrelationId, Code),
+  rabbit_global_counters:increase_protocol_counter(stream, Counter, 1),
+  {C, State};
 handle_frame_post_auth(Transport,
                        {ok, #stream_connection{
                                name = ConnName,
@@ -3102,7 +3139,7 @@ evaluate_state_after_secret_update(Transport,
     {_, Conn1} = ensure_token_expiry_timer(User, Conn0),
     PublisherStreams =
     lists:foldl(fun(#publisher{stream = Str}, Acc) ->
-                        case rabbit_stream_utils:check_write_permitted(stream_r(Str, Conn0), User) of
+                        case check_write_permitted(stream_r(Str, Conn0), User) of
                             ok ->
                                 Acc;
                             _ ->
@@ -3426,6 +3463,9 @@ clean_state_after_stream_deletion_or_failure(MemberPid, Stream,
             {not_cleaned, C2#stream_connection{stream_leaders = Leaders1}, S2}
     end.
 
+store_offset(Reference, _, _, C) when is_binary(Reference), byte_size(Reference) > ?MAX_REFERENCE_SIZE ->
+  rabbit_log:warning("Reference is too long to store offset: ~p", [byte_size(Reference)]),
+  C;
 store_offset(Reference, Stream, Offset, Connection0) ->
     case lookup_leader(Stream, Connection0) of
         {error, Error} ->


Test patch:
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
index 06792b4e739d..91644f1364f6 100644
--- a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
+++ b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
@@ -64,7 +64,9 @@ groups() ->
        test_super_stream_duplicate_partitions,
        authentication_error_should_close_with_delay,
        unauthorized_vhost_access_should_close_with_delay,
-       sasl_anonymous
+       sasl_anonymous,
+       test_publisher_with_too_long_reference_errors,
+       test_consumer_with_too_long_reference_errors
       ]},
      %% Run `test_global_counters` on its own so the global metrics are
      %% initialised to 0 for each testcase
@@ -945,6 +947,70 @@ unauthorized_vhost_access_should_close_with_delay(Config) ->
     closed = wait_for_socket_close(T, S, 10),
     ok.
 
+test_publisher_with_too_long_reference_errors(Config) ->
+  FunctionName = atom_to_binary(?FUNCTION_NAME, utf8),
+  T = gen_tcp,
+  Port = get_port(T, Config),
+  Opts = get_opts(T),
+  {ok, S} = T:connect("localhost", Port, Opts),
+  C = rabbit_stream_core:init(0),
+  ConnectionName = FunctionName,
+  test_peer_properties(T, S, #{<<"connection_name">> => ConnectionName}, C),
+  test_authenticate(T, S, C),
+
+  Stream = FunctionName,
+  test_create_stream(T, S, Stream, C),
+
+  MaxSize = 255,
+  ReferenceOK = iolist_to_binary(lists:duplicate(MaxSize, <<"a">>)),
+  ReferenceKO = iolist_to_binary(lists:duplicate(MaxSize + 1, <<"a">>)),
+
+  Tests = [{1, ReferenceOK, ?RESPONSE_CODE_OK},
+           {2, ReferenceKO, ?RESPONSE_CODE_PRECONDITION_FAILED}],
+
+  [begin
+     F = request({declare_publisher, PubId, Ref, Stream}),
+     ok = T:send(S, F),
+     {Cmd, C} = receive_commands(T, S, C),
+     ?assertMatch({response, 1, {declare_publisher, ExpectedResponseCode}}, Cmd)
+   end || {PubId, Ref, ExpectedResponseCode} <- Tests],
+
+  test_delete_stream(T, S, Stream, C),
+  test_close(T, S, C),
+  ok.
+
+test_consumer_with_too_long_reference_errors(Config) ->
+  FunctionName = atom_to_binary(?FUNCTION_NAME, utf8),
+  T = gen_tcp,
+  Port = get_port(T, Config),
+  Opts = get_opts(T),
+  {ok, S} = T:connect("localhost", Port, Opts),
+  C = rabbit_stream_core:init(0),
+  ConnectionName = FunctionName,
+  test_peer_properties(T, S, #{<<"connection_name">> => ConnectionName}, C),
+  test_authenticate(T, S, C),
+
+  Stream = FunctionName,
+  test_create_stream(T, S, Stream, C),
+
+  MaxSize = 255,
+  ReferenceOK = iolist_to_binary(lists:duplicate(MaxSize, <<"a">>)),
+  ReferenceKO = iolist_to_binary(lists:duplicate(MaxSize + 1, <<"a">>)),
+
+  Tests = [{1, ReferenceOK, ?RESPONSE_CODE_OK},
+           {2, ReferenceKO, ?RESPONSE_CODE_PRECONDITION_FAILED}],
+
+  [begin
+     F = request({subscribe, SubId, Stream, first, 1, #{<<"name">> => Ref}}),
+     ok = T:send(S, F),
+     {Cmd, C} = receive_commands(T, S, C),
+     ?assertMatch({response, 1, {subscribe, ExpectedResponseCode}}, Cmd)
+   end || {SubId, Ref, ExpectedResponseCode} <- Tests],
+
+  test_delete_stream(T, S, Stream, C),
+  test_close(T, S, C),
+  ok.
+
 consumer_offset_info(Config, ConnectionName) ->
     [[{offset, Offset},
       {offset_lag, Lag}]] = rpc(Config, 0, ?MODULE,