Implement SM resumption into SASL2

big_tests/tests/sasl2_SUITE.erl

@@ -22,7 +22,8 @@ groups() ->
      {all_tests, [parallel],
       [
        {group, basic},
-       {group, scram}
+       {group, scram},
+       {group, stream_management}
       ]},
      {basic, [parallel],
       [
@@ -40,6 +41,14 @@ groups() ->
        authenticate_with_scram_bad_abort,
        authenticate_with_scram_bad_response,
        authenticate_with_scram
+      ]},
+     {stream_management, [parallel],
+      [
+       sm_failure_missing_previd_does_not_stop_sasl2,
+       sm_failure_invalid_h_does_not_stop_sasl2,
+       sm_failure_exceeding_h_does_not_stop_sasl2,
+       sm_failure_unknown_smid_does_not_stop_sasl2,
+       sm_is_bound_at_sasl2_success
       ]}
     ].
 
@@ -78,7 +87,9 @@ end_per_testcase(Name, Config) ->
 load_sasl_extensible(Config) ->
     HostType = domain_helper:host_type(),
     Config1 = dynamic_modules:save_modules(HostType, Config),
-    dynamic_modules:ensure_modules(HostType, [{mod_sasl2, #{}}]),
+    Modules = [{mod_sasl2, config_parser_helper:default_mod_config(mod_sasl2)},
+               {mod_stream_management, config_parser_helper:mod_config(mod_stream_management, #{ack_freq => never})}],
+    dynamic_modules:ensure_modules(HostType, Modules),
     Config1.
 
 %%--------------------------------------------------------------------
@@ -100,7 +111,7 @@ server_announces_sasl2_with_some_mechanism(Config) ->
     ?assertNotEqual([], Mechs).
 
 authenticate_stanza_has_invalid_mechanism(Config) ->
-    Steps = [connect_tls_user, start_stream_get_features, send_invalid_authenticate_stanza],
+    Steps = [connect_tls_user, start_stream_get_features, send_invalid_mech_auth_stanza],
     #{answer := Response} = sasl2_helper:apply_steps(Steps, Config),
     ?assertMatch(#xmlel{name = <<"failure">>, attrs = [{<<"xmlns">>, ?NS_SASL_2}]}, Response).
 
@@ -159,3 +170,43 @@ authenticate_again_results_in_stream_error(Config) ->
              plain_authentication, receive_features, plain_authentication],
     #{answer := Response} = sasl2_helper:apply_steps(Steps, Config),
     escalus:assert(is_stream_error, [<<"policy-violation">>, <<>>], Response).
+
+sm_failure_missing_previd_does_not_stop_sasl2(Config) ->
+    Steps = [buffer_messages_and_die, connect_tls_user, start_stream_get_features,
+             auth_with_resumption_missing_previd, receive_features],
+    #{answer := Success} = sasl2_helper:apply_steps(Steps, Config),
+    ?assertMatch(#xmlel{name = <<"success">>, attrs = [{<<"xmlns">>, ?NS_SASL_2}]}, Success),
+    Resumed = exml_query:path(Success, [{element_with_ns, <<"failed">>, ?NS_STREAM_MGNT_3}]),
+    escalus:assert(is_sm_failed, [<<"bad-request">>], Resumed).
+
+sm_failure_invalid_h_does_not_stop_sasl2(Config) ->
+    Steps = [buffer_messages_and_die, connect_tls_user, start_stream_get_features,
+             auth_with_resumption_invalid_h, receive_features],
+    #{answer := Success} = sasl2_helper:apply_steps(Steps, Config),
+    ?assertMatch(#xmlel{name = <<"success">>, attrs = [{<<"xmlns">>, ?NS_SASL_2}]}, Success),
+    Resumed = exml_query:path(Success, [{element_with_ns, <<"failed">>, ?NS_STREAM_MGNT_3}]),
+    escalus:assert(is_sm_failed, [<<"bad-request">>], Resumed).
+
+sm_failure_exceeding_h_does_not_stop_sasl2(Config) ->
+    Steps = [buffer_messages_and_die, connect_tls_user, start_stream_get_features,
+             auth_with_resumption_exceeding_h, receive_features],
+    #{answer := Success} = sasl2_helper:apply_steps(Steps, Config),
+    ?assertMatch(#xmlel{name = <<"success">>, attrs = [{<<"xmlns">>, ?NS_SASL_2}]}, Success),
+    Resumed = exml_query:path(Success, [{element_with_ns, <<"failed">>, ?NS_STREAM_MGNT_3}]),
+    escalus:assert(is_sm_failed, [<<"bad-request">>], Resumed).
+
+sm_failure_unknown_smid_does_not_stop_sasl2(Config) ->
+    Steps = [buffer_messages_and_die, connect_tls_user, start_stream_get_features,
+             auth_with_resumption_unknown_smid, receive_features],
+    #{answer := Success} = sasl2_helper:apply_steps(Steps, Config),
+    ?assertMatch(#xmlel{name = <<"success">>, attrs = [{<<"xmlns">>, ?NS_SASL_2}]}, Success),
+    Resumed = exml_query:path(Success, [{element_with_ns, <<"failed">>, ?NS_STREAM_MGNT_3}]),
+    escalus:assert(is_sm_failed, [<<"item-not-found">>], Resumed).
+
+sm_is_bound_at_sasl2_success(Config) ->
+    Steps = [buffer_messages_and_die, connect_tls_user, start_stream_get_features,
+             auth_with_resumption, has_no_more_stanzas],
+    #{answer := Success, smid := SMID} = sasl2_helper:apply_steps(Steps, Config),
+    ?assertMatch(#xmlel{name = <<"success">>, attrs = [{<<"xmlns">>, ?NS_SASL_2}]}, Success),
+    Resumed = exml_query:path(Success, [{element_with_ns, <<"resumed">>, ?NS_STREAM_MGNT_3}]),
+    ?assert(escalus_pred:is_sm_resumed(SMID, Resumed)).

big_tests/tests/sasl2_helper.erl

@@ -3,6 +3,7 @@
 
 -include_lib("exml/include/exml.hrl").
 -include_lib("escalus/include/escalus.hrl").
+-include_lib("escalus/include/escalus_xmlns.hrl").
 
 -define(NS_SASL_2, <<"urn:xmpp:sasl:2">>).
 
@@ -42,14 +43,22 @@ start_stream_get_features(_Config, Client, Data) ->
     Features = escalus_connection:get_stanza(Client1, wait_for_features),
     {Client1, Data#{features => Features}}.
 
-send_invalid_authenticate_stanza(_Config, Client, Data) ->
+send_invalid_mech_auth_stanza(_Config, Client, Data) ->
     Authenticate = #xmlel{name = <<"authenticate">>,
                           attrs = [{<<"xmlns">>, ?NS_SASL_2},
                                    {<<"mechanism">>, <<"invalid-non-existent-mechanism">>}]},
     escalus:send(Client, Authenticate),
     Answer = escalus_client:wait_for_stanza(Client),
     {Client, Data#{answer => Answer}}.
 
+send_invalid_ns_auth_stanza(_Config, Client, Data) ->
+    Authenticate = #xmlel{name = <<"authenticate">>,
+                          attrs = [{<<"xmlns">>, <<"bad-namespace">>},
+                                   {<<"mechanism">>, <<"PLAIN">>}]},
+    escalus:send(Client, Authenticate),
+    Answer = escalus_client:wait_for_stanza(Client),
+    {Client, Data#{answer => Answer}}.
+
 send_bad_user_agent(_Config, Client, Data) ->
     InitialResponse = initial_response_elem(<<"some-random-payload">>),
     Agent = bad_user_agent_elem(),
@@ -58,6 +67,37 @@ send_bad_user_agent(_Config, Client, Data) ->
     Answer = escalus_client:wait_for_stanza(Client),
     {Client, Data#{answer => Answer}}.
 
+auth_with_resumption(Config, Client, #{smid := SMID, texts := Texts} = Data) ->
+    Resume = escalus_stanza:resume(SMID, 1),
+    {Client1, Data1} = plain_auth(Config, Client, Data, [Resume]),
+    Msgs = sm_helper:wait_for_messages(Client, Texts),
+    {Client1, Data1#{sm_storage => Msgs}}.
+
+auth_with_resumption_invalid_h(Config, Client, #{smid := SMID} = Data) ->
+    Resume = #xmlel{name = <<"resume">>,
+                    attrs = [{<<"xmlns">>, ?NS_STREAM_MGNT_3},
+                             {<<"previd">>, SMID},
+                             {<<"h">>, <<"aaa">>}]},
+    plain_auth(Config, Client, Data, [Resume]).
+
+auth_with_resumption_missing_previd(Config, Client, Data) ->
+    Resume = #xmlel{name = <<"resume">>,
+                    attrs = [{<<"xmlns">>, ?NS_STREAM_MGNT_3},
+                             {<<"h">>, <<"aaa">>}]},
+    plain_auth(Config, Client, Data, [Resume]).
+
+auth_with_resumption_exceeding_h(Config, Client, #{smid := SMID} = Data) ->
+    Resume = escalus_stanza:resume(SMID, 999),
+    plain_auth(Config, Client, Data, [Resume]).
+
+auth_with_resumption_unknown_smid(Config, Client, Data) ->
+    Resume = escalus_stanza:resume(<<"123456">>, 1),
+    plain_auth(Config, Client, Data, [Resume]).
+
+has_no_more_stanzas(_Config, Client, Data) ->
+    escalus_assert:has_no_stanzas(Client),
+    {Client, Data}.
+
 plain_auth_user_agent_without_id(Config, Client, Data) ->
     plain_auth(Config, Client, Data, [user_agent_elem_without_id()]).
 
@@ -124,6 +164,22 @@ scram_step_2(_Config, Client,
         {error, _, _} -> throw({auth_failed, SuccessStanza})
     end.
 
+buffer_messages_and_die(Config, _Client, Data) ->
+    Spec = escalus_fresh:create_fresh_user(Config, alice),
+    Client = sm_helper:connect_spec(Spec, sr_presence, manual),
+    C2SPid = mongoose_helper:get_session_pid(Client),
+    BobSpec = escalus_fresh:create_fresh_user(Config, bob),
+    {ok, Bob, _} = escalus_connection:start(BobSpec),
+    Texts = [ integer_to_binary(N) || N <- [1, 2, 3]],
+    sm_helper:send_messages(Bob, Client, Texts),
+    %% Client receives them, but doesn't ack.
+    sm_helper:wait_for_messages(Client, Texts),
+    %% Client's connection is violently terminated.
+    escalus_client:kill_connection(Config, Client),
+    sm_SUITE:wait_until_resume_session(C2SPid),
+    SMID = sm_helper:client_to_smid(Client),
+    {C2SPid, Data#{spec => Spec, smid => SMID, smh => 3, texts => Texts}}.
+
 receive_features(_Config, Client, Data) ->
     Features = escalus_client:wait_for_stanza(Client),
     {Client, Data#{features => Features}}.

src/c2s/mongoose_c2s_acc.erl

@@ -13,6 +13,7 @@
 %% - `route': mongoose_acc elements to trigger the whole `handle_route' pipeline.
 %% - `flush': mongoose_acc elements to trigger the `handle_flush` pipeline.
 %% - `socket_send': xml elements to send on the socket to the user.
+%% - `socket_send_first': xml elements to send on the socket to the user, but appended first.
 -module(mongoose_c2s_acc).
 
 -export([new/0, new/1,
@@ -21,7 +22,16 @@
          to_acc/3, to_acc_many/2
         ]).
 
--type key() :: state_mod | actions | c2s_state | c2s_data | stop | hard_stop | route | flush | socket_send.
+-type key() :: state_mod
+             | actions
+             | c2s_state
+             | c2s_data
+             | stop
+             | hard_stop
+             | route
+             | flush
+             | socket_send
+             | socket_send_first.
 -type pairs() :: [pair()].
 -type pair() :: {state_mod, {module(), term()}}
               | {actions, gen_statem:action()}
@@ -31,7 +41,8 @@
               | {hard_stop, term() | {shutdown, atom()}}
               | {route, mongoose_acc:t()}
               | {flush, mongoose_acc:t()}
-              | {socket_send, exml:element()}.
+              | {socket_send, exml:element() | [exml:element()]}
+              | {socket_send_first, exml:element() | [exml:element()]}.
 
 -type t() :: #{
         state_mod := #{module() => term()},
@@ -54,7 +65,7 @@
         socket_send => [exml:element()]
        }.
 
--export_type([t/0]).
+-export_type([t/0, pairs/0]).
 
 %% --------------------------------------------------------
 %% API
@@ -122,7 +133,8 @@ from_mongoose_acc(Acc, Key) ->
             (mongoose_acc:t(), stop, atom() | {shutdown, atom()}) -> mongoose_acc:t();
             (mongoose_acc:t(), route, mongoose_acc:t()) -> mongoose_acc:t();
             (mongoose_acc:t(), flush, mongoose_acc:t()) -> mongoose_acc:t();
-            (mongoose_acc:t(), socket_send, exml:element()) -> mongoose_acc:t().
+            (mongoose_acc:t(), socket_send, exml:element() | [exml:element()]) -> mongoose_acc:t();
+            (mongoose_acc:t(), socket_send_first, exml:element() | [exml:element()]) -> mongoose_acc:t().
 to_acc(Acc, Key, NewValue) ->
     C2SAcc = mongoose_acc:get_statem_acc(Acc),
     C2SAcc1 = to_c2s_acc(C2SAcc, {Key, NewValue}),
@@ -161,6 +173,10 @@ to_c2s_acc(C2SAcc = #{socket_send := Stanzas}, {socket_send, NewStanzas}) when i
     C2SAcc#{socket_send := lists:reverse(NewStanzas) ++ Stanzas};
 to_c2s_acc(C2SAcc = #{socket_send := Stanzas}, {socket_send, Stanza}) ->
     C2SAcc#{socket_send := [Stanza | Stanzas]};
+to_c2s_acc(C2SAcc = #{socket_send := Stanzas}, {socket_send_first, NewStanzas}) when is_list(NewStanzas) ->
+    C2SAcc#{socket_send := Stanzas ++ NewStanzas};
+to_c2s_acc(C2SAcc = #{socket_send := Stanzas}, {socket_send_first, Stanza}) ->
+    C2SAcc#{socket_send := Stanzas ++ [Stanza]};
 to_c2s_acc(C2SAcc = #{actions := Actions}, {stop, Reason}) ->
     C2SAcc#{actions := [{next_event, cast, {stop, Reason}} | Actions]};
 to_c2s_acc(C2SAcc, {Key, NewValue}) ->