Initial version of new XMPP stream behaviour (for review)

src/cyrsasl.erl

@@ -31,7 +31,7 @@
 
 -export([start/0, register_mechanism/3, listmech/1,
 	 server_new/7, server_start/3, server_step/2,
-	 opt_type/1]).
+	 get_mech/1, opt_type/1]).
 
 -include("ejabberd.hrl").
 -include("logger.hrl").
@@ -53,6 +53,7 @@
 -type(password_type() :: plain | digest | scram).
 -type(props() :: [{username, binary()} |
                   {authzid, binary()} |
+		  {mechanism, binary()} |
                   {auth_module, atom()}]).
 
 -type(sasl_mechanism() :: #sasl_mechanism{}).
@@ -65,9 +66,11 @@
     get_password,
     check_password,
     check_password_digest,
+    mech_name = <<"">>,
     mech_mod,
     mech_state
 }).
+-type sasl_state() :: #sasl_state{}.
 
 -callback mech_new(binary(), fun(), fun(), fun()) -> any().
 -callback mech_step(any(), binary()) -> {ok, props()} |
@@ -150,6 +153,7 @@ server_start(State, Mech, ClientIn) ->
 				    State#sasl_state.check_password,
 				    State#sasl_state.check_password_digest),
 		server_step(State#sasl_state{mech_mod = Module,
+					     mech_name = Mech,
 					     mech_state = MechState},
 			    ClientIn);
 	    _ -> {error, 'no-mechanism'}
@@ -181,6 +185,10 @@ server_step(State, ClientIn) ->
             {error, Error}
     end.
 
+-spec get_mech(sasl_state()) -> binary().
+get_mech(#sasl_state{mech_name = Mech}) ->
+    Mech.
+
 %% Remove the anonymous mechanism from the list if not enabled for the given
 %% host
 %%

src/ejabberd_sm.erl

@@ -34,6 +34,7 @@
 %% API
 -export([start/0,
 	 start_link/0,
+	 route/2,
 	 route/3,
 	 process_iq/3,
 	 open_session/5,
@@ -69,6 +70,7 @@
 	 get_all_pids/0,
 	 is_existing_resource/3,
 	 get_commands_spec/0,
+	 c2s_handle_info/2,
 	 make_sid/0
 	]).
 
@@ -98,15 +100,6 @@
 %% default value for the maximum number of user connections
 -define(MAX_USER_SESSIONS, infinity).
 
--type broadcast() :: {broadcast, broadcast_data()}.
-
--type broadcast_data() ::
-        {rebind, pid(), binary()} | %% ejabberd_c2s
-        {item, ljid(), mod_roster:subscription()} | %% mod_roster/mod_shared_roster
-        {exit, binary()} | %% mod_roster/mod_shared_roster
-        {privacy_list, mod_privacy:userlist(), binary()} | %% mod_privacy
-        {blocking, unblock_all | {block | unblock, [ljid()]}}. %% mod_blocking
-
 %%====================================================================
 %% API
 %%====================================================================
@@ -120,7 +113,18 @@ start() ->
 start_link() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
--spec route(jid(), jid(), stanza() | broadcast()) -> ok.
+-spec route(jid(), term()) -> ok.
+%% @doc route arbitrary term to c2s process(es)
+route(To, Term) ->
+    case catch do_route(To, Term) of
+	{'EXIT', Reason} ->
+	    ?ERROR_MSG("route ~p to ~p failed: ~p",
+		       [Term, To, Reason]);
+	_ ->
+	    ok
+    end.
+
+-spec route(jid(), jid(), stanza()) -> ok.
 
 route(From, To, Packet) ->
     case catch do_route(From, To, Packet) of
@@ -180,9 +184,7 @@ bounce_offline_message(From, To, Packet) ->
 -spec disconnect_removed_user(binary(), binary()) -> ok.
 
 disconnect_removed_user(User, Server) ->
-    ejabberd_sm:route(jid:make(<<"">>, <<"">>, <<"">>),
+    route(jid:make(User, Server, <<"">>), {exit, <<"User removed">>}).
-		      jid:make(User, Server, <<"">>),
-                      {broadcast, {exit, <<"User removed">>}}).
 
 get_user_resources(User, Server) ->
     LUser = jid:nodeprep(User),
@@ -356,6 +358,21 @@ register_iq_handler(Host, XMLNS, Module, Fun, Opts) ->
 unregister_iq_handler(Host, XMLNS) ->
     ejabberd_sm ! {unregister_iq_handler, Host, XMLNS}.
 
+%% Why the hell do we have so many similar kicks?
+c2s_handle_info({noreply, #{lang := Lang} = State}, replaced) ->
+    State1 = State#{replaced => true},
+    Err = xmpp:serr_conflict(<<"Replaced by new connection">>, Lang),
+    ejabberd_c2s:send(State1, Err);
+c2s_handle_info({noreply, #{lang := Lang} = State}, kick) ->
+    Err = xmpp:serr_policy_violation(<<"has been kicked">>, Lang),
+    c2s_handle_info({noreply, State}, {kick, kicked_by_admin, Err});
+c2s_handle_info({noreply, State}, {kick, _Reason, Err}) ->
+    ejabberd_c2s:send(State, Err);
+c2s_handle_info({noreply, #{lang := Lang} = State}, {exit, Reason}) ->
+    Err = xmpp:serr_conflict(Reason, Lang),
+    ejabberd_c2s:send(State, Err);
+c2s_handle_info(Acc, _) ->
+    Acc.
 
 %%====================================================================
 %% gen_server callbacks
@@ -366,6 +383,8 @@ init([]) ->
     ets:new(sm_iqtable, [named_table]),
     lists:foreach(
       fun(Host) ->
+	      ejabberd_hooks:add(c2s_handle_info, Host,
+				 ejabberd_sm, c2s_handle_info, 50),
 	      ejabberd_hooks:add(roster_in_subscription, Host,
 				 ejabberd_sm, check_in_subscription, 20),
 	      ejabberd_hooks:add(offline_message_hook, Host,
@@ -411,6 +430,17 @@ handle_info({unregister_iq_handler, Host, XMLNS},
 handle_info(_Info, State) -> {noreply, State}.
 
 terminate(_Reason, _State) ->
+    lists:foreach(
+      fun(Host) ->
+	      ejabberd_hooks:delete(c2s_handle_info, Host,
+				    ejabberd_sm, c2s_handle_info, 50),
+	      ejabberd_hooks:delete(roster_in_subscription, Host,
+				    ejabberd_sm, check_in_subscription, 20),
+	      ejabberd_hooks:delete(offline_message_hook, Host,
+				    ejabberd_sm, bounce_offline_message, 100),
+	      ejabberd_hooks:delete(remove_user, Host,
+				    ejabberd_sm, disconnect_removed_user, 100)
+      end, ?MYHOSTS),
     ejabberd_commands:unregister_commands(get_commands_spec()),
     ok.
 
@@ -444,26 +474,27 @@ is_online(#session{info = Info}) ->
     not proplists:get_bool(offline, Info).
 
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
--spec do_route(jid(), jid(), stanza() | broadcast()) -> any().
+-spec do_route(jid(), term()) -> any().
-do_route(From, #jid{lresource = <<"">>} = To, {broadcast, _} = Packet) ->
+do_route(#jid{lresource = <<"">>} = To, Term) ->
-    ?DEBUG("processing broadcast to bare JID: ~p", [Packet]),
     lists:foreach(
       fun(R) ->
-	      do_route(From, jid:replace_resource(To, R), Packet)
+	      do_route(jid:replace_resource(To, R), Term)
       end, get_user_resources(To#jid.user, To#jid.server));
-do_route(From, To, {broadcast, _} = Packet) ->
+do_route(To, Term) ->
-    ?DEBUG("processing broadcast to full JID: ~p", [Packet]),
+    ?DEBUG("broadcasting ~p to ~s", [Term, jid:to_string(To)]),
     {U, S, R} = jid:tolower(To),
     Mod = get_sm_backend(S),
     case online(Mod:get_sessions(U, S, R)) of
 	[] ->
-	    ?DEBUG("dropping broadcast to unavailable resourse: ~p", [Packet]);
+	    ?DEBUG("dropping broadcast to unavailable resourse: ~p", [Term]);
 	Ss ->
 	    Session = lists:max(Ss),
 	    Pid = element(2, Session#session.sid),
-	    ?DEBUG("sending to process ~p: ~p", [Pid, Packet]),
+	    ?DEBUG("sending to process ~p: ~p", [Pid, Term]),
-	    Pid ! {route, From, To, Packet}
+	    Pid ! Term
-    end;
+    end.
+
+-spec do_route(jid(), jid(), stanza()) -> any().
 do_route(From, To, #presence{type = T, status = Status} = Packet)
   when T == subscribe; T == subscribed; T == unsubscribe; T == unsubscribed ->
     ?DEBUG("processing subscription:~n~s", [xmpp:pp(Packet)]),

src/ejabberd_socket.erl

@@ -49,6 +49,7 @@
 	 sockname/1, peername/1]).
 
 -include("ejabberd.hrl").
+-include("xmpp.hrl").
 -include("logger.hrl").
 
 -type sockmod() :: ejabberd_http_bind |
@@ -150,15 +151,25 @@ connect(Addr, Port, Opts, Timeout, Owner) ->
     end.
 
 starttls(SocketData, TLSOpts) ->
-    {ok, TLSSocket} = fast_tls:tcp_to_tls(SocketData#socket_state.socket, TLSOpts),
+    case fast_tls:tcp_to_tls(SocketData#socket_state.socket, TLSOpts) of
-    ejabberd_receiver:starttls(SocketData#socket_state.receiver, TLSSocket),
+	{ok, TLSSocket} ->
-    SocketData#socket_state{socket = TLSSocket, sockmod = fast_tls}.
+	    ejabberd_receiver:starttls(SocketData#socket_state.receiver, TLSSocket),
+	    {ok, SocketData#socket_state{socket = TLSSocket, sockmod = fast_tls}};
+	Err ->
+	    ?ERROR_MSG("starttls failed: ~p", [Err]),
+	    Err
+    end.
 
 starttls(SocketData, TLSOpts, Data) ->
-    {ok, TLSSocket} = fast_tls:tcp_to_tls(SocketData#socket_state.socket, TLSOpts),
+    case fast_tls:tcp_to_tls(SocketData#socket_state.socket, TLSOpts) of
-    ejabberd_receiver:starttls(SocketData#socket_state.receiver, TLSSocket),
+	{ok, TLSSocket} ->
-    send(SocketData, Data),
+	    ejabberd_receiver:starttls(SocketData#socket_state.receiver, TLSSocket),
-    SocketData#socket_state{socket = TLSSocket, sockmod = fast_tls}.
+	    send(SocketData, Data),
+	    {ok, SocketData#socket_state{socket = TLSSocket, sockmod = fast_tls}};
+	Err ->
+	    ?ERROR_MSG("starttls failed: ~p", [Err]),
+	    Err
+    end.
 
 compress(SocketData) -> compress(SocketData, undefined).
 
@@ -184,10 +195,10 @@ send(SocketData, Data) ->
         ok -> ok;
 	{error, timeout} ->
 	    ?INFO_MSG("Timeout on ~p:send",[SocketData#socket_state.sockmod]),
-	    exit(normal);
+	    {error, timeout};
         Error ->
 	    ?DEBUG("Error in ~p:send: ~p",[SocketData#socket_state.sockmod, Error]),
-	    exit(normal)
+	    Error
     end.
 
 %% Can only be called when in c2s StateData#state.xml_socket is true

src/mod_admin_extra.erl

@@ -918,9 +918,8 @@ kick_session(User, Server, Resource, ReasonText) ->
     ok.
 
 kick_this_session(User, Server, Resource, Reason) ->
-    ejabberd_sm:route(jid:make(<<"">>, <<"">>, <<"">>),
+    ejabberd_sm:route(jid:make(User, Server, Resource),
-                      jid:make(User, Server, Resource),
+                      {exit, Reason}).
-                      {broadcast, {exit, Reason}}).
 
 status_num(Host, Status) ->
     length(get_status_list(Host, Status)).
@@ -948,7 +947,7 @@ get_status_list(Host, Status_required) ->
 	    end,
     Sessions3 = [ {Pid, Server, Priority} || {{_User, Server, _Resource}, {_, Pid}, Priority} <- Sessions2, apply(Fhost, [Server, Host])],
     %% For each Pid, get its presence
-    Sessions4 = [ {catch ejabberd_c2s:get_presence(Pid), Server, Priority} || {Pid, Server, Priority} <- Sessions3],
+    Sessions4 = [ {catch get_presence(Pid), Server, Priority} || {Pid, Server, Priority} <- Sessions3],
     %% Filter by status
     Fstatus = case Status_required of
 		  <<"all">> ->
@@ -1001,6 +1000,16 @@ stringize(String) ->
     %% Replace newline characters with other code
     ejabberd_regexp:greplace(String, <<"\n">>, <<"\\n">>).
 
+get_presence(Pid) ->
+    Pres = #presence{from = From} = ejabberd_c2s:get_presence(Pid),
+    Show = case Pres of
+	       #presence{type = unavailable} -> <<"unavailable">>;
+	       #presence{show = undefined} -> <<"available">>;
+	       #presence{show = S} -> atom_to_binary(S, utf8)
+	   end,
+    Status = xmpp:get_text(Pres#presence.status),
+    {From#jid.user, From#jid.resource, Show, Status}.
+
 get_presence(U, S) ->
     Pids = [ejabberd_sm:get_session_pid(U, S, R)
 	    || R <- ejabberd_sm:get_user_resources(U, S)],
@@ -1009,8 +1018,7 @@ get_presence(U, S) ->
 	[] ->
 	    {jid:to_string({U, S, <<>>}), <<"unavailable">>, <<"">>};
 	[SessionPid|_] ->
-	    {_User, Resource, Show, Status} =
+	    {_User, Resource, Show, Status} = get_presence(SessionPid),
-		     ejabberd_c2s:get_presence(SessionPid),
 	    FullJID = jid:to_string({U, S, Resource}),
 	    {FullJID, Show, Status}
     end.
@@ -1053,7 +1061,7 @@ user_sessions_info(User, Host) ->
       fun(Session) ->
 	      {_U, _S, Resource} = Session#session.usr,
 	      {Now, Pid} = Session#session.sid,
-	      {_U, _Resource, Status, StatusText} = ejabberd_c2s:get_presence(Pid),
+	      {_U, _Resource, Status, StatusText} = get_presence(Pid),
 	      Info = Session#session.info,
 	      Priority = Session#session.priority,
 	      Conn = proplists:get_value(conn, Info),
@@ -1306,7 +1314,7 @@ push_roster_item(LU, LS, U, S, Action) ->
 push_roster_item(LU, LS, R, U, S, Action) ->
     LJID = jid:make(LU, LS, R),
     BroadcastEl = build_broadcast(U, S, Action),
-    ejabberd_sm:route(LJID, LJID, BroadcastEl),
+    ejabberd_sm:route(LJID, BroadcastEl),
     Item = build_roster_item(U, S, Action),
     ResIQ = build_iq_roster_push(Item),
     ejabberd_router:route(jid:remove_resource(LJID), LJID, ResIQ).
@@ -1331,7 +1339,7 @@ build_broadcast(U, S, remove) ->
 %% @spec (U::binary(), S::binary(), Subs::atom()) -> any()
 %% Subs = both | from | to | none
 build_broadcast(U, S, SubsAtom) when is_atom(SubsAtom) ->
-    {broadcast, {item, {U, S, <<>>}, SubsAtom}}.
+    {item, {U, S, <<>>}, SubsAtom}.
 
 %%%
 %%% Last Activity

src/mod_blocking.erl

@@ -29,7 +29,7 @@
 
 -protocol({xep, 191, '1.2'}).
 
--export([start/2, stop/1, process_iq/1,
+-export([start/2, stop/1, process_iq/1, c2s_handle_info/2,
 	 process_iq_set/3, process_iq_get/3, mod_opt_type/1, depends/2]).
 
 -include("ejabberd.hrl").
@@ -52,6 +52,10 @@ start(Host, Opts) ->
 		       process_iq_get, 40),
     ejabberd_hooks:add(privacy_iq_set, Host, ?MODULE,
 		       process_iq_set, 40),
+    ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE,
+		       c2s_handle_info, 40),
+    ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE,
+		       c2s_handle_info, 40),
     mod_disco:register_feature(Host, ?NS_BLOCKING),
     gen_iq_handler:add_iq_handler(ejabberd_sm, Host,
 				  ?NS_BLOCKING, ?MODULE, process_iq, IQDisc).
@@ -229,14 +233,12 @@ make_userlist(Name, List) ->
 -spec broadcast_list_update(binary(), binary(), binary(), userlist()) -> ok.
 broadcast_list_update(LUser, LServer, Name, UserList) ->
     ejabberd_sm:route(jid:make(LUser, LServer, <<"">>),
-                      jid:make(LUser, LServer, <<"">>),
+                      {privacy_list, UserList, Name}).
-                      {broadcast, {privacy_list, UserList, Name}}).
 
 -spec broadcast_blocklist_event(binary(), binary(), block_event()) -> ok.
 broadcast_blocklist_event(LUser, LServer, Event) ->
     JID = jid:make(LUser, LServer, <<"">>),
-    ejabberd_sm:route(JID, JID,
+    ejabberd_sm:route(JID, {blocking, Event}).
-                      {broadcast, {blocking, Event}}).
 
 -spec process_blocklist_get(binary(), binary(), binary()) ->
 				   {error, stanza_error()} | {result, block_list()}.
@@ -251,6 +253,27 @@ process_blocklist_get(LUser, LServer, Lang) ->
 	  {result, #block_list{items = Items}}
     end.
 
+-spec c2s_handle_info(ejabberd_c2s:next_state(), term()) -> ejabberd_c2s:next_state().
+c2s_handle_info({noreply, #{user := U, server := S, resource := R} = State},
+		{blocking, Action}) ->
+    SubEl = case Action of
+		{block, JIDs} ->
+		    #block{items = JIDs};
+		{unblock, JIDs} ->
+		    #unblock{items = JIDs};
+		unblock_all ->
+		    #unblock{}
+	    end,
+    PushIQ = #iq{type = set,
+		 from = jid:make(U, S),
+		 to = jid:make(U, S, R),
+		 id = <<"push", (randoms:get_string())/binary>>,
+		 sub_els = [SubEl]},
+    %% No need to replace active privacy list here,
+    %% blocking pushes are always accompanied by
+    %% Privacy List pushes
+    ejabberd_c2s:send(State, PushIQ).
+
 -spec db_mod(binary()) -> module().
 db_mod(LServer) ->
     DBType = gen_mod:db_type(LServer, mod_privacy),

src/mod_caps.erl

@@ -35,10 +35,10 @@
 
 -behaviour(gen_mod).
 
--export([read_caps/1, caps_stream_features/2,
+-export([read_caps/1, list_features/1, caps_stream_features/2,
 	 disco_features/5, disco_identity/5, disco_info/5,
 	 get_features/2, export/1, import_info/0, import/5,
-         import_start/2, import_stop/2]).
+         get_user_caps/2, import_start/2, import_stop/2]).
 
 %% gen_mod callbacks
 -export([start/2, start_link/2, stop/1, depends/2]).
@@ -48,8 +48,7 @@
 	 handle_cast/2, terminate/2, code_change/3]).
 
 -export([user_send_packet/4, user_receive_packet/5,
-	 c2s_presence_in/2, c2s_filter_packet/6,
+	 c2s_presence_in/2, mod_opt_type/1]).
-	 c2s_broadcast_recipients/6, mod_opt_type/1]).
 
 -include("ejabberd.hrl").
 -include("logger.hrl").
@@ -104,6 +103,22 @@ get_features(Host, #caps{node = Node, version = Version,
 		end,
 		[], SubNodes).
 
+-spec list_features(ejabberd_c2s:state()) -> [{ljid(), caps()}].
+list_features(C2SState) ->
+    Rs = maps:get(caps_features, C2SState, gb_trees:empty()),
+    gb_trees:to_list(Rs).
+
+-spec get_user_caps(jid(), ejabberd_c2s:state()) -> {ok, caps()} | error.
+get_user_caps(JID, C2SState) ->
+    Rs = maps:get(caps_features, C2SState, gb_trees:empty()),
+    LJID = jid:tolower(JID),
+    case gb_trees:lookup(LJID, Rs) of
+	{value, Caps} ->
+	    {ok, Caps};
+	none ->
+	    error
+    end.
+
 -spec read_caps(#presence{}) -> nothing | caps().
 read_caps(Presence) ->
     case xmpp:get_subtag(Presence, #caps{}) of
@@ -194,87 +209,40 @@ disco_info(Acc, Host, Module, Node, Lang) when is_atom(Module) ->
 disco_info(Acc, _, _, _Node, _Lang) ->
     Acc.
 
--spec c2s_presence_in(ejabberd_c2s:state(), {jid(), jid(), presence()}) ->
+-spec c2s_presence_in(ejabberd_c2s:state(), presence()) -> ejabberd_c2s:state().
-			     ejabberd_c2s:state().
 c2s_presence_in(C2SState,
-		{From, To, #presence{type = Type} = Presence}) ->
+		#presence{from = From, to = To, type = Type} = Presence) ->
-    Subscription = ejabberd_c2s:get_subscription(From,
+    Subscription = ejabberd_c2s:get_subscription(From, C2SState),
-						 C2SState),
     Insert = (Type == available)
 	       and ((Subscription == both) or (Subscription == to)),
     Delete = (Type == unavailable) or (Type == error),
     if Insert or Delete ->
-	   LFrom = jid:tolower(From),
+	    LFrom = jid:tolower(From),
-	   Rs = case ejabberd_c2s:get_aux_field(caps_resources,
+	    Rs = maps:get(caps_resources, C2SState, gb_trees:empty()),
-						C2SState)
+	    Caps = read_caps(Presence),
-		    of
+	    NewRs = case Caps of
-		  {ok, Rs1} -> Rs1;
+			nothing when Insert == true -> Rs;
-		  error -> gb_trees:empty()
+			_ when Insert == true ->
-		end,
+			    case gb_trees:lookup(LFrom, Rs) of
-	   Caps = read_caps(Presence),
+				{value, Caps} -> Rs;
-	   NewRs = case Caps of
+				none ->
-		     nothing when Insert == true -> Rs;
+				    ejabberd_hooks:run(caps_add, To#jid.lserver,
-		     _ when Insert == true ->
+						       [From, To,
-			 case gb_trees:lookup(LFrom, Rs) of
+							get_features(To#jid.lserver, Caps)]),
-			   {value, Caps} -> Rs;
+				    gb_trees:insert(LFrom, Caps, Rs);
-			   none ->
+				_ ->
-				ejabberd_hooks:run(caps_add, To#jid.lserver,
+				    ejabberd_hooks:run(caps_update, To#jid.lserver,
-						   [From, To,
+						       [From, To,
-						    get_features(To#jid.lserver, Caps)]),
+							get_features(To#jid.lserver, Caps)]),
-				gb_trees:insert(LFrom, Caps, Rs);
+				    gb_trees:update(LFrom, Caps, Rs)
-			   _ ->
+			    end;
-				ejabberd_hooks:run(caps_update, To#jid.lserver,
+			_ -> gb_trees:delete_any(LFrom, Rs)
-						   [From, To,
+		    end,
-						    get_features(To#jid.lserver, Caps)]),
+	    C2SState#{caps_resources := NewRs};
-				gb_trees:update(LFrom, Caps, Rs)
+       true ->
-			 end;
+	    C2SState
-		     _ -> gb_trees:delete_any(LFrom, Rs)
-		   end,
-	   ejabberd_c2s:set_aux_field(caps_resources, NewRs,
-				      C2SState);
-       true -> C2SState
     end.
 
--spec c2s_filter_packet(boolean(), binary(), ejabberd_c2s:state(),
-			{pep_message, binary()}, jid(), stanza()) ->
-			       boolean().
-c2s_filter_packet(InAcc, Host, C2SState, {pep_message, Feature}, To, _Packet) ->
-    case ejabberd_c2s:get_aux_field(caps_resources, C2SState) of
-      {ok, Rs} ->
-	  LTo = jid:tolower(To),
-	  case gb_trees:lookup(LTo, Rs) of
-	    {value, Caps} ->
-		Drop = not lists:member(Feature, get_features(Host, Caps)),
-		{stop, Drop};
-	    none ->
-		{stop, true}
-	  end;
-      _ -> InAcc
-    end;
-c2s_filter_packet(Acc, _, _, _, _, _) -> Acc.
-
--spec c2s_broadcast_recipients([ljid()], binary(), ejabberd_c2s:state(),
-			       {pep_message, binary()}, jid(), stanza()) ->
-				      [ljid()].
-c2s_broadcast_recipients(InAcc, Host, C2SState,
-			 {pep_message, Feature}, _From, _Packet) ->
-    case ejabberd_c2s:get_aux_field(caps_resources,
-				    C2SState)
-	of
-      {ok, Rs} ->
-	  gb_trees_fold(fun (USR, Caps, Acc) ->
-				case lists:member(Feature,
-                                                  get_features(Host, Caps))
-				    of
-				  true -> [USR | Acc];
-				  false -> Acc
-				end
-			end,
-			InAcc, Rs);
-      _ -> InAcc
-    end;
-c2s_broadcast_recipients(Acc, _, _, _, _, _) -> Acc.
-
 -spec depends(binary(), gen_mod:opts()) -> [{module(), hard | soft}].
 depends(_Host, _Opts) ->
     [].
@@ -292,15 +260,11 @@ init([Host, Opts]) ->
 		  [{max_size, MaxSize}, {life_time, LifeTime}]),
     ejabberd_hooks:add(c2s_presence_in, Host, ?MODULE,
 		       c2s_presence_in, 75),
-    ejabberd_hooks:add(c2s_filter_packet, Host, ?MODULE,
-		       c2s_filter_packet, 75),
-    ejabberd_hooks:add(c2s_broadcast_recipients, Host,
-		       ?MODULE, c2s_broadcast_recipients, 75),
     ejabberd_hooks:add(user_send_packet, Host, ?MODULE,
 		       user_send_packet, 75),
     ejabberd_hooks:add(user_receive_packet, Host, ?MODULE,
 		       user_receive_packet, 75),
-    ejabberd_hooks:add(c2s_stream_features, Host, ?MODULE,
+    ejabberd_hooks:add(c2s_post_auth_features, Host, ?MODULE,
 		       caps_stream_features, 75),
     ejabberd_hooks:add(s2s_stream_features, Host, ?MODULE,
 		       caps_stream_features, 75),
@@ -325,15 +289,11 @@ terminate(_Reason, State) ->
     Host = State#state.host,
     ejabberd_hooks:delete(c2s_presence_in, Host, ?MODULE,
 			  c2s_presence_in, 75),
-    ejabberd_hooks:delete(c2s_filter_packet, Host, ?MODULE,
-			  c2s_filter_packet, 75),
-    ejabberd_hooks:delete(c2s_broadcast_recipients, Host,
-			  ?MODULE, c2s_broadcast_recipients, 75),
     ejabberd_hooks:delete(user_send_packet, Host, ?MODULE,
 			  user_send_packet, 75),
     ejabberd_hooks:delete(user_receive_packet, Host,
 			  ?MODULE, user_receive_packet, 75),
-    ejabberd_hooks:delete(c2s_stream_features, Host,
+    ejabberd_hooks:delete(c2s_post_auth_features, Host,
 			  ?MODULE, caps_stream_features, 75),
     ejabberd_hooks:delete(s2s_stream_features, Host,
 			  ?MODULE, caps_stream_features, 75),
@@ -494,20 +454,6 @@ concat_xdata_fields(#xdata{fields = Fields} = X) ->
 	      is_binary(Var), Var /= <<"FORM_TYPE">>],
     [Form, $<, lists:sort(Res)].
 
--spec gb_trees_fold(fun((_, _, T) -> T), T, gb_trees:tree()) -> T.
-gb_trees_fold(F, Acc, Tree) ->
-    Iter = gb_trees:iterator(Tree),
-    gb_trees_fold_iter(F, Acc, Iter).
-
--spec gb_trees_fold_iter(fun((_, _, T) -> T), T, gb_trees:iter()) -> T.
-gb_trees_fold_iter(F, Acc, Iter) ->
-    case gb_trees:next(Iter) of
-      {Key, Val, NewIter} ->
-	  NewAcc = F(Key, Val, Acc),
-	  gb_trees_fold_iter(F, NewAcc, NewIter);
-      _ -> Acc
-    end.
-
 -spec now_ts() -> integer().
 now_ts() ->
     p1_time_compat:system_time(seconds).

src/mod_client_state.erl

@@ -260,20 +260,15 @@ queue_take(Stanza, Host, C2SState) ->
     NewState = set_queue(Rest, C2SState),
     {NewState, get_stanzas(Selected, Host) ++ [Stanza]}.
 
--spec set_queue(csi_queue(), term()) -> term().
+-spec set_queue(csi_queue(), ejabberd_c2s:state()) -> ejabberd_c2s:state().
 
 set_queue(Queue, C2SState) ->
-    ejabberd_c2s:set_aux_field(csi_queue, Queue, C2SState).
+    C2SState#{csi_queue => Queue}.
 
--spec get_queue(term()) -> csi_queue().
+-spec get_queue(ejabberd_c2s:state()) -> csi_queue().
 
 get_queue(C2SState) ->
-    case ejabberd_c2s:get_aux_field(csi_queue, C2SState) of
+    maps:get(csi_queue, C2SState, []).
-      {ok, Queue} ->
-	      Queue;
-      error ->
-	      []
-    end.
 
 -spec get_stanzas(csi_queue(), binary()) -> [stanza()].
 

src/mod_last.erl

@@ -37,7 +37,7 @@
 	 process_sm_iq/1, on_presence_update/4, import_info/0,
 	 import/5, import_start/2, store_last_info/4, get_last_info/2,
 	 remove_user/2, transform_options/1, mod_opt_type/1,
-	 opt_type/1, register_user/2, depends/2]).
+	 opt_type/1, register_user/2, depends/2, privacy_check_packet/4]).
 
 -include("ejabberd.hrl").
 -include("logger.hrl").
@@ -64,6 +64,8 @@ start(Host, Opts) ->
 				  ?NS_LAST, ?MODULE, process_local_iq, IQDisc),
     gen_iq_handler:add_iq_handler(ejabberd_sm, Host,
 				  ?NS_LAST, ?MODULE, process_sm_iq, IQDisc),
+    ejabberd_hooks:add(privacy_check_packet, Host, ?MODULE,
+		       privacy_check_packet, 30),
     ejabberd_hooks:add(register_user, Host, ?MODULE,
 		       register_user, 50),
     ejabberd_hooks:add(remove_user, Host, ?MODULE,
@@ -143,6 +145,31 @@ process_sm_iq(#iq{from = From, to = To, lang = Lang} = IQ) ->
 	    xmpp:make_error(IQ, xmpp:err_subscription_required(Txt, Lang))
     end.
 
+privacy_check_packet(allow, C2SState,
+		     #iq{from = From, to = To, type = T} = IQ, in)
+  when T == get; T == set ->
+    case xmpp:has_subtag(IQ, #last{}) of
+	true ->
+	    Sub = ejabberd_c2s:get_subscription(From, C2SState),
+	    if Sub == from; Sub == both ->
+		    Pres = #presence{from = To, to = From},
+		    case ejabberd_hooks:run_fold(
+			   privacy_check_packet, allow,
+			   [C2SState, Pres, out]) of
+			allow ->
+			    allow;
+			deny ->
+			    {stop, deny}
+		    end;
+	       true ->
+		    {stop, deny}
+	    end;
+	false ->
+	    allow
+    end;
+privacy_check_packet(Acc, _, _, _) ->
+    Acc.
+
 %% @spec (LUser::string(), LServer::string()) ->
 %%      {ok, TimeStamp::integer(), Status::string()} | not_found | {error, Reason}
 -spec get_last(binary(), binary()) -> {ok, non_neg_integer(), binary()} |

src/mod_legacy_auth.erl

@@ -0,0 +1,159 @@
+%%%-------------------------------------------------------------------
+%%% Created : 11 Dec 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
+%%%
+%%%
+%%% ejabberd, Copyright (C) 2002-2016   ProcessOne
+%%%
+%%% This program is free software; you can redistribute it and/or
+%%% modify it under the terms of the GNU General Public License as
+%%% published by the Free Software Foundation; either version 2 of the
+%%% License, or (at your option) any later version.
+%%%
+%%% This program is distributed in the hope that it will be useful,
+%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
+%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+%%% General Public License for more details.
+%%%
+%%% You should have received a copy of the GNU General Public License along
+%%% with this program; if not, write to the Free Software Foundation, Inc.,
+%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+%%%
+%%%-------------------------------------------------------------------
+-module(mod_legacy_auth).
+-behaviour(gen_mod).
+
+-protocol({xep, 78, '2.5'}).
+
+%% gen_mod API
+-export([start/2, stop/1, depends/2, mod_opt_type/1]).
+%% hooks
+-export([c2s_unauthenticated_packet/2, c2s_stream_features/2]).
+
+-include("xmpp.hrl").
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+start(Host, _Opts) ->
+    ejabberd_hooks:add(c2s_unauthenticated_packet, Host, ?MODULE,
+		       c2s_unauthenticated_packet, 50),
+    ejabberd_hooks:add(c2s_pre_auth_features, Host, ?MODULE,
+		       c2s_stream_features, 50).
+
+stop(Host) ->
+    ejabberd_hooks:delete(c2s_unauthenticated_packet, Host, ?MODULE,
+			  c2s_unauthenticated_packet, 50),
+    ejabberd_hooks:delete(c2s_pre_auth_features, Host, ?MODULE,
+			  c2s_stream_features, 50).
+
+depends(_Host, _Opts) ->
+    [].
+
+mod_opt_type(_) ->
+    [].
+
+c2s_unauthenticated_packet({noreply, State}, #iq{type = T, sub_els = [_]} = IQ)
+  when T == get; T == set ->
+    case xmpp:get_subtag(IQ, #legacy_auth{}) of
+	#legacy_auth{} = Auth ->
+	    {stop, authenticate(State, xmpp:set_els(IQ, [Auth]))};
+	false ->
+	    {noreply, State}
+    end;
+c2s_unauthenticated_packet(Acc, _) ->
+    Acc.
+
+c2s_stream_features(Acc, LServer) ->
+    case gen_mod:is_loaded(LServer, ?MODULE) of
+	true ->
+	    [#legacy_auth_feature{}|Acc];
+	false ->
+	    Acc
+    end.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+authenticate(#{server := Server} = State,
+	     #iq{type = get, sub_els = [#legacy_auth{}]} = IQ) ->
+    LServer = jid:nameprep(Server),
+    Auth = #legacy_auth{username = <<>>, password = <<>>, resource = <<>>},
+    Res = case ejabberd_auth:plain_password_required(LServer) of
+	      false ->
+		  xmpp:make_iq_result(IQ, Auth#legacy_auth{digest = <<>>});
+	      true ->
+		  xmpp:make_iq_result(IQ, Auth)
+	  end,
+    ejabberd_c2s:send(State, Res);
+authenticate(State,
+	     #iq{type = set, lang = Lang,
+		 sub_els = [#legacy_auth{username = U,
+					 resource = R}]} = IQ)
+  when U == undefined; R == undefined; U == <<"">>; R == <<"">> ->
+    Txt = <<"Both the username and the resource are required">>,
+    Err = xmpp:make_error(IQ, xmpp:err_not_acceptable(Txt, Lang)),
+    ejabberd_c2s:send(State, Err);
+authenticate(#{stream_id := StreamID, server := Server,
+	       access := Access, ip := IP} = State,
+	     #iq{type = set, lang = Lang,
+		 sub_els = [#legacy_auth{username = U,
+					 password = P0,
+					 digest = D0,
+					 resource = R}]} = IQ) ->
+    P = if is_binary(P0) -> P0; true -> <<>> end,
+    D = if is_binary(D0) -> D0; true -> <<>> end,
+    DGen = fun (PW) -> p1_sha:sha(<<StreamID/binary, PW/binary>>) end,
+    JID = jid:make(U, Server, R),
+    case JID /= error andalso
+	acl:access_matches(Access,
+			   #{usr => jid:split(JID), ip => IP},
+			   JID#jid.lserver) == allow of
+	true ->
+	    case ejabberd_auth:check_password_with_authmodule(
+		   U, U, JID#jid.lserver, P, D, DGen) of
+		{true, AuthModule} ->
+		    case ejabberd_c2s:handle_auth_success(
+			   U, <<"legacy">>, AuthModule, State) of
+			{noreply, State1} ->
+			    State2 = State1#{user := U},
+			    open_session(State2, IQ, R);
+			Err ->
+			    Err
+		    end;
+		_ ->
+		    Err = xmpp:make_error(IQ, xmpp:err_not_authorized()),
+		    process_auth_failure(State, U, Err, 'not-authorized')
+	    end;
+	false when JID == error ->
+	    Err = xmpp:make_error(IQ, xmpp:err_jid_malformed()),
+	    process_auth_failure(State, U, Err, 'jid-malformed');
+	false ->
+	    Txt = <<"Denied by ACL">>,
+	    Err = xmpp:make_error(IQ, xmpp:err_forbidden(Txt, Lang)),
+	    process_auth_failure(State, U, Err, 'forbidden')
+    end.
+
+open_session(State, IQ, R) ->
+    case ejabberd_c2s:bind(R, State) of
+	{ok, State1} ->
+	    Res = xmpp:make_iq_result(IQ),
+	    case ejabberd_c2s:send(State1, Res) of
+		{noreply, State2} ->
+		    {noreply, State2#{stream_authenticated := true,
+				      stream_state := session_established}};
+		Err ->
+		    Err
+	    end;
+	{error, Err, State1} ->
+	    Res = xmpp:make_error(IQ, Err),
+	    ejabberd_c2s:send(State1, Res)
+    end.
+
+process_auth_failure(State, User, StanzaErr, Reason) ->
+    case ejabberd_c2s:send(State, StanzaErr) of
+	{noreply, State1} ->
+	    ejabberd_c2s:handle_auth_failure(
+	      User, <<"legacy">>, Reason, State1);
+	Err ->
+	    Err
+    end.

src/mod_offline.erl

@@ -61,6 +61,7 @@
 	 count_offline_messages/2,
 	 get_offline_els/2,
 	 find_x_expire/2,
+	 c2s_handle_info/2,
 	 webadmin_page/3,
 	 webadmin_user/4,
 	 webadmin_user_parse_query/5]).
@@ -156,6 +157,7 @@ init([Host, Opts]) ->
     ejabberd_hooks:add(disco_sm_items, Host,
 		       ?MODULE, get_sm_items, 50),
     ejabberd_hooks:add(disco_info, Host, ?MODULE, get_info, 50),
+    ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE, c2s_handle_info, 50),
     ejabberd_hooks:add(webadmin_page_host, Host,
 		       ?MODULE, webadmin_page, 50),
     ejabberd_hooks:add(webadmin_user, Host,
@@ -211,6 +213,7 @@ terminate(_Reason, State) ->
     ejabberd_hooks:delete(disco_sm_identity, Host, ?MODULE, get_sm_identity, 50),
     ejabberd_hooks:delete(disco_sm_items, Host, ?MODULE, get_sm_items, 50),
     ejabberd_hooks:delete(disco_info, Host, ?MODULE, get_info, 50),
+    ejabberd_hooks:delete(c2s_handle_info, Host, ?MODULE, c2s_handle_info, 50),
     ejabberd_hooks:delete(webadmin_page_host, Host,
 			  ?MODULE, webadmin_page, 50),
     ejabberd_hooks:delete(webadmin_user, Host,
@@ -277,38 +280,28 @@ get_sm_identity(Acc, #jid{luser = U, lserver = S}, #jid{luser = U, lserver = S},
 get_sm_identity(Acc, _From, _To, _Node, _Lang) ->
     Acc.
 
-get_sm_items(_Acc, #jid{luser = U, lserver = S, lresource = R} = JID,
+get_sm_items(_Acc, #jid{luser = U, lserver = S} = JID,
 	     #jid{luser = U, lserver = S},
 	     ?NS_FLEX_OFFLINE, _Lang) ->
-    case ejabberd_sm:get_session_pid(U, S, R) of
+    ejabberd_sm:route(JID, {resend_offline, false}),
-	Pid when is_pid(Pid) ->
+    Mod = gen_mod:db_mod(S, ?MODULE),
-	    Mod = gen_mod:db_mod(S, ?MODULE),
+    Hdrs = Mod:read_message_headers(U, S),
-	    Hdrs = Mod:read_message_headers(U, S),
+    BareJID = jid:remove_resource(JID),
-	    BareJID = jid:remove_resource(JID),
+    {result, lists:map(
-	    Pid ! dont_ask_offline,
+	       fun({Seq, From, _To, _TS, _El}) ->
-	    {result, lists:map(
+		       Node = integer_to_binary(Seq),
-		       fun({Seq, From, _To, _TS, _El}) ->
+		       #disco_item{jid = BareJID,
-			       Node = integer_to_binary(Seq),
+				   node = Node,
-			       #disco_item{jid = BareJID,
+				   name = jid:to_string(From)}
-					   node = Node,
+	       end, Hdrs)};
-					   name = jid:to_string(From)}
-		       end, Hdrs)};
-	none ->
-	    {result, []}
-    end;
 get_sm_items(Acc, _From, _To, _Node, _Lang) ->
     Acc.
 
 -spec get_info([xdata()], binary(), module(), binary(), binary()) -> [xdata()];
 	      ([xdata()], jid(), jid(), binary(), binary()) -> [xdata()].
-get_info(_Acc, #jid{luser = U, lserver = S, lresource = R},
+get_info(_Acc, #jid{luser = U, lserver = S} = JID,
 	 #jid{luser = U, lserver = S}, ?NS_FLEX_OFFLINE, Lang) ->
-    case ejabberd_sm:get_session_pid(U, S, R) of
+    ejabberd_sm:route(JID, {resend_offline, false}),
-	Pid when is_pid(Pid) ->
-	    Pid ! dont_ask_offline;
-	none ->
-	    ok
-    end,
     [#xdata{type = result,
 	    fields = flex_offline:encode(
 		       [{number_of_messages, count_offline_messages(U, S)}],
@@ -316,6 +309,12 @@ get_info(_Acc, #jid{luser = U, lserver = S, lresource = R},
 get_info(Acc, _From, _To, _Node, _Lang) ->
     Acc.
 
+-spec c2s_handle_info(ejabberd_c2s:next_state(), term()) -> ejabberd_c2s:next_state().
+c2s_handle_info({noreply, State}, {resend_offline, Flag}) ->
+    {noreply, State#{resend_offline => Flag}};
+c2s_handle_info(Acc, _) ->
+    Acc.
+
 -spec handle_offline_query(iq()) -> iq().
 handle_offline_query(#iq{from = #jid{luser = U1, lserver = S1},
 			 to = #jid{luser = U2, lserver = S2},
@@ -395,18 +394,15 @@ set_offline_tag(Msg, Node) ->
     xmpp:set_subtag(Msg, #offline{items = [#offline_item{node = Node}]}).
 
 -spec handle_offline_fetch(jid()) -> ok.
-handle_offline_fetch(#jid{luser = U, lserver = S, lresource = R}) ->
+handle_offline_fetch(#jid{luser = U, lserver = S} = JID) ->
-    case ejabberd_sm:get_session_pid(U, S, R) of
+    ejabberd_sm:route(JID, {resend_offline, false}),
-	none ->
+    lists:foreach(
-	    ok;
+      fun({Node, El}) ->
-	Pid when is_pid(Pid) ->
+	      NewEl = set_offline_tag(El, Node),
-	    Pid ! dont_ask_offline,
+	      From = xmpp:get_from(El),
-	    lists:foreach(
+	      To = xmpp:get_to(El),
-	      fun({Node, El}) ->
+	      ejabberd_router:route(From, To, NewEl)
-		      NewEl = set_offline_tag(El, Node),
+      end, read_messages(U, S)).
-		      Pid ! {route, xmpp:get_from(El), xmpp:get_to(El), NewEl}
-	      end, read_messages(U, S))
-    end.
 
 -spec fetch_msg_by_node(jid(), binary()) -> error | {ok, #offline_msg{}}.
 fetch_msg_by_node(To, Seq) ->

src/mod_privacy.erl

@@ -35,7 +35,7 @@
 	 process_iq_set/3, process_iq_get/3, get_user_list/3,
 	 check_packet/6, remove_user/2, encode_list_item/1,
 	 is_list_needdb/1, updated_list/3,
-	 import_start/2, import_stop/2,
+	 import_start/2, import_stop/2, c2s_handle_info/2,
          item_to_xml/1, get_user_lists/2, import/5,
 	 set_privacy_list/1, mod_opt_type/1, depends/2]).
 
@@ -77,6 +77,8 @@ start(Host, Opts) ->
 		       updated_list, 50),
     ejabberd_hooks:add(remove_user, Host, ?MODULE,
 		       remove_user, 50),
+    ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE,
+		       c2s_handle_info, 50),
     gen_iq_handler:add_iq_handler(ejabberd_sm, Host,
 				  ?NS_PRIVACY, ?MODULE, process_iq, IQDisc).
 
@@ -94,6 +96,8 @@ stop(Host) ->
 			  ?MODULE, updated_list, 50),
     ejabberd_hooks:delete(remove_user, Host, ?MODULE,
 			  remove_user, 50),
+    ejabberd_hooks:delete(c2s_handle_info, Host, ?MODULE,
+			  c2s_handle_info, 50),
     gen_iq_handler:remove_iq_handler(ejabberd_sm, Host,
 				     ?NS_PRIVACY).
 
@@ -310,13 +314,8 @@ process_lists_set(LUser, LServer, Name, [], _UserList, Lang) ->
 	    Txt = <<"No privacy list with this name found">>,
 	    {error, xmpp:err_item_not_found(Txt, Lang)};
 	{atomic, ok} ->
-	    ejabberd_sm:route(jid:make(LUser, LServer,
+	    ejabberd_sm:route(jid:make(LUser, LServer, <<"">>),
-				       <<"">>),
+			      {privacy_list, #userlist{name = Name}, Name}),
-			      jid:make(LUser, LServer, <<"">>),
-			      {broadcast, {privacy_list,
-					   #userlist{name = Name,
-						     list = []},
-					   Name}}),
 	    {result, undefined};
 	Err ->
 	    ?ERROR_MSG("failed to remove privacy list '~s' for user ~s@~s: ~p",
@@ -334,14 +333,12 @@ process_lists_set(LUser, LServer, Name, Items, _UserList, Lang) ->
 	    case Mod:set_privacy_list(LUser, LServer, Name, List) of
 		{atomic, ok} ->
 		    NeedDb = is_list_needdb(List),
-		    ejabberd_sm:route(jid:make(LUser, LServer,
+		    ejabberd_sm:route(jid:make(LUser, LServer, <<"">>),
-					       <<"">>),
+				      {privacy_list,
-				      jid:make(LUser, LServer, <<"">>),
+				       #userlist{name = Name,
-				      {broadcast, {privacy_list,
+						 list = List,
-						   #userlist{name = Name,
+						 needdb = NeedDb},
-							     list = List,
+				       Name}),
-							     needdb = NeedDb},
-						   Name}}),
 		    {result, undefined};
 		Err ->
 		    ?ERROR_MSG("failed to set privacy list '~s' "
@@ -538,6 +535,23 @@ remove_user(User, Server) ->
     Mod = gen_mod:db_mod(LServer, ?MODULE),
     Mod:remove_user(LUser, LServer).
 
+c2s_handle_info({noreply, #{privacy_list := Old,
+			    user := U, server := S, resource := R} = State},
+		{privacy_list, New, Name}) ->
+    List = if Old#userlist.name == New#userlist.name -> New;
+	      true -> Old
+	   end,
+    From = jid:make(U, S),
+    To = jid:make(U, S, R),
+    PushIQ = #iq{type = set, from = From, to = To,
+		 id = <<"push", (randoms:get_string())/binary>>,
+		 sub_els = [#privacy_query{
+			       lists = [#privacy_list{name = Name}]}]},
+    State1 = State#{privacy_list => List},
+    ejabberd_c2s:send(State1, PushIQ);
+c2s_handle_info(Acc, _) ->
+    Acc.
+
 -spec updated_list(userlist(), userlist(), userlist()) -> userlist().
 updated_list(_, #userlist{name = OldName} = Old,
 	     #userlist{name = NewName} = New) ->

src/mod_pubsub.erl

@@ -54,7 +54,8 @@
     on_user_offline/3, remove_user/2,
     disco_local_identity/5, disco_local_features/5,
     disco_local_items/5, disco_sm_identity/5,
-    disco_sm_features/5, disco_sm_items/5]).
+    disco_sm_features/5, disco_sm_items/5,
+    c2s_handle_info/2]).
 
 %% exported iq handlers
 -export([iq_sm/1, process_disco_info/1, process_disco_items/1,
@@ -305,6 +306,8 @@ init([ServerHost, Opts]) ->
 	?MODULE, remove_user, 50),
     ejabberd_hooks:add(anonymous_purge_hook, ServerHost,
 	?MODULE, remove_user, 50),
+    ejabberd_hooks:add(c2s_handle_info, ServerHost,
+	?MODULE, c2s_handle_info, 50),
     gen_iq_handler:add_iq_handler(ejabberd_local, Host, ?NS_DISCO_INFO,
 				  ?MODULE, process_disco_info, IQDisc),
     gen_iq_handler:add_iq_handler(ejabberd_local, Host, ?NS_DISCO_ITEMS,
@@ -912,6 +915,8 @@ terminate(_Reason,
 	?MODULE, remove_user, 50),
     ejabberd_hooks:delete(anonymous_purge_hook, ServerHost,
 	?MODULE, remove_user, 50),
+    ejabberd_hooks:delete(c2s_handle_info, ServerHost,
+	?MODULE, c2s_handle_info, 50),
     gen_iq_handler:remove_iq_handler(ejabberd_local, Host, ?NS_DISCO_INFO),
     gen_iq_handler:remove_iq_handler(ejabberd_local, Host, ?NS_DISCO_ITEMS),
     gen_iq_handler:remove_iq_handler(ejabberd_local, Host, ?NS_PUBSUB),
@@ -2212,22 +2217,21 @@ send_items(Host, Node, _Nidx, _Type, Options, LJID, _) ->
     dispatch_items(Host, LJID, Node, Stanza).
 
 dispatch_items({FromU, FromS, FromR} = From, {ToU, ToS, ToR} = To,
-	    Node, Stanza) ->
+	       Node, Stanza) ->
     C2SPid = case ejabberd_sm:get_session_pid(ToU, ToS, ToR) of
-	ToPid when is_pid(ToPid) -> ToPid;
+		 ToPid when is_pid(ToPid) -> ToPid;
-	_ ->
+		 _ ->
-	    R = user_resource(FromU, FromS, FromR),
+		     R = user_resource(FromU, FromS, FromR),
-	    case ejabberd_sm:get_session_pid(FromU, FromS, R) of
+		     case ejabberd_sm:get_session_pid(FromU, FromS, R) of
-		FromPid when is_pid(FromPid) -> FromPid;
+			 FromPid when is_pid(FromPid) -> FromPid;
-		_ -> undefined
+			 _ -> undefined
-	    end
+		     end
-    end,
+	     end,
     if C2SPid == undefined -> ok;
-	true ->
+       true ->
-	    ejabberd_c2s:send_filtered(C2SPid,
+	    C2SPid ! {send_filtered, {pep_message, <<Node/binary, "+notify">>},
-		{pep_message, <<Node/binary, "+notify">>},
+		      service_jid(From), jid:make(To),
-		service_jid(From), jid:make(To),
+		      Stanza}
-		Stanza)
     end;
 dispatch_items(From, To, _Node, Stanza) ->
     ejabberd_router:route(service_jid(From), jid:make(To), Stanza).
@@ -2761,9 +2765,10 @@ get_resource_state({U, S, R}, ShowValues, JIDs) ->
 	    lists:append([{U, S, R}], JIDs);
 	Pid ->
 	    Show = case ejabberd_c2s:get_presence(Pid) of
-		{_, _, <<"available">>, _} -> <<"online">>;
+		       #presence{type = unavailable} -> <<"unavailable">>;
-		{_, _, State, _} -> State
+		       #presence{show = undefined} -> <<"online">>;
-	    end,
+		       #presence{show = S} -> atom_to_binary(S, latin1)
+		   end,
 	    case lists:member(Show, ShowValues) of
 		%% If yes, item can be delivered
 		true -> lists:append([{U, S, R}], JIDs);
@@ -3008,25 +3013,56 @@ broadcast_stanza({LUser, LServer, LResource}, Publisher, Node, Nidx, Type, NodeO
     broadcast_stanza({LUser, LServer, <<>>}, Node, Nidx, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM),
     %% Handles implicit presence subscriptions
     SenderResource = user_resource(LUser, LServer, LResource),
-    case ejabberd_sm:get_session_pid(LUser, LServer, SenderResource) of
+    NotificationType = get_option(NodeOptions, notification_type, headline),
-	C2SPid when is_pid(C2SPid) ->
+    Stanza = add_message_type(BaseStanza, NotificationType),
-	    NotificationType = get_option(NodeOptions, notification_type, headline),
+    %% set the from address on the notification to the bare JID of the account owner
-	    Stanza = add_message_type(BaseStanza, NotificationType),
+    %% Also, add "replyto" if entity has presence subscription to the account owner
-	    %% set the from address on the notification to the bare JID of the account owner
+    %% See XEP-0163 1.1 section 4.3.1
-	    %% Also, add "replyto" if entity has presence subscription to the account owner
+    ejabberd_sm:route(jid:make(LUser, LServer, SenderResource),
-	    %% See XEP-0163 1.1 section 4.3.1
+		      {pep_message, <<((Node))/binary, "+notify">>,
-	    ejabberd_c2s:broadcast(C2SPid,
+		       jid:make(LUser, LServer, <<"">>),
-		{pep_message, <<((Node))/binary, "+notify">>},
+		       add_extended_headers(
-		_Sender = jid:make(LUser, LServer, <<"">>),
+			 Stanza, extended_headers([Publisher]))});
-		_StanzaToSend = add_extended_headers(
-				  Stanza, 
-				  _ReplyTo = extended_headers([Publisher])));
-	_ ->
-	    ?DEBUG("~p@~p has no session; can't deliver ~p to contacts", [LUser, LServer, BaseStanza])
-    end;
 broadcast_stanza(Host, _Publisher, Node, Nidx, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) ->
     broadcast_stanza(Host, Node, Nidx, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM).
 
+-spec c2s_handle_info(ejabberd_c2s:next_state(), term()) -> ejabberd_c2s:next_state().
+c2s_handle_info({noreply, #{server := Server} = C2SState},
+		{pep_message, Feature, From, Packet}) ->
+    LServer = jid:nameprep(Server),
+    lists:foreach(
+      fun({USR, Caps}) ->
+	      Features = mod_caps:get_features(LServer, Caps),
+	      case lists:member(Feature, Features) of
+		  true ->
+		      To = jid:make(USR),
+		      NewPacket = xmpp:set_from_to(Packet, From, To),
+		      ejabberd_router:route(From, To, NewPacket);
+		  false ->
+		      ok
+	      end
+      end, mod_caps:list_features(C2SState)),
+    {noreply, C2SState};
+c2s_handle_info({noreply, #{server := Server} = C2SState},
+		{send_filtered, {pep_message, Feature}, From, To, Packet}) ->
+    LServer = jid:nameprep(Server),
+    case mod_caps:get_user_caps(To, C2SState) of
+	{ok, Caps} ->
+	    Features = mod_caps:get_features(LServer, Caps),
+	    case lists:member(Feature, Features) of
+		true ->
+		    NewPacket = xmpp:set_from_to(Packet, From, To),
+		    ejabberd_router:route(From, To, NewPacket);
+		false ->
+		    ok
+	    end;
+	error ->
+	    ok
+    end,
+    {noreply, C2SState};
+c2s_handle_info(Acc, _) ->
+    Acc.
+
 subscribed_nodes_by_jid(NotifyType, SubsByDepth) ->
     NodesToDeliver = fun (Depth, Node, Subs, Acc) ->
 	    NodeName = case Node#pubsub_node.nodeid of

src/mod_register.erl

@@ -34,7 +34,7 @@
 -behaviour(gen_mod).
 
 -export([start/2, stop/1, stream_feature_register/2,
-	 unauthenticated_iq_register/4, try_register/5,
+	 c2s_unauthenticated_packet/2, try_register/5,
 	 process_iq/1, send_registration_notifications/3,
 	 transform_options/1, transform_module_options/1,
 	 mod_opt_type/1, opt_type/1, depends/2]).
@@ -50,10 +50,10 @@ start(Host, Opts) ->
 				  ?NS_REGISTER, ?MODULE, process_iq, IQDisc),
     gen_iq_handler:add_iq_handler(ejabberd_sm, Host,
 				  ?NS_REGISTER, ?MODULE, process_iq, IQDisc),
-    ejabberd_hooks:add(c2s_stream_features, Host, ?MODULE,
+    ejabberd_hooks:add(c2s_pre_auth_features, Host, ?MODULE,
 		       stream_feature_register, 50),
-    ejabberd_hooks:add(c2s_unauthenticated_iq, Host,
+    ejabberd_hooks:add(c2s_unauthenticated_packet, Host,
-		       ?MODULE, unauthenticated_iq_register, 50),
+		       ?MODULE, c2s_unauthenticated_packet, 50),
     ejabberd_mnesia:create(?MODULE, mod_register_ip,
 			[{ram_copies, [node()]}, {local_content, true},
 			 {attributes, [key, value]}]),
@@ -62,10 +62,10 @@ start(Host, Opts) ->
     ok.
 
 stop(Host) ->
-    ejabberd_hooks:delete(c2s_stream_features, Host,
+    ejabberd_hooks:delete(c2s_pre_auth_features, Host,
 			  ?MODULE, stream_feature_register, 50),
-    ejabberd_hooks:delete(c2s_unauthenticated_iq, Host,
+    ejabberd_hooks:delete(c2s_unauthenticated_packet, Host,
-			  ?MODULE, unauthenticated_iq_register, 50),
+			  ?MODULE, c2s_unauthenticated_packet, 50),
     gen_iq_handler:remove_iq_handler(ejabberd_local, Host,
 				     ?NS_REGISTER),
     gen_iq_handler:remove_iq_handler(ejabberd_sm, Host,
@@ -86,19 +86,20 @@ stream_feature_register(Acc, Host) ->
 	    Acc
     end.
 
--spec unauthenticated_iq_register(empty | iq(), binary(), iq(),
+c2s_unauthenticated_packet({noreply, #{ip := IP, server := Server} = State},
-				  {inet:ip_address(), non_neg_integer()}) ->
+			   #iq{type = T, sub_els = [_]} = IQ)
-					 empty | iq().
+  when T == set; T == get ->
-unauthenticated_iq_register(_Acc, Server,
+    case xmpp:get_subtag(IQ, #register{}) of
-			    #iq{sub_els = [#register{}]} = IQ, IP) ->
+	#register{} ->
-    Address = case IP of
+	    {Address, _} = IP,
-		{A, _Port} -> A;
+	    IQ1 = xmpp:set_from_to(IQ, jid:make(<<>>), jid:make(Server)),
-		_ -> undefined
+	    ResIQ = process_iq(IQ1, Address),
-	      end,
+	    ResIQ1 = xmpp:set_from_to(ResIQ, jid:make(Server), undefined),
-    ResIQ = process_iq(xmpp:set_from_to(IQ, jid:make(<<>>), jid:make(Server)),
+	    {stop, ejabberd_c2s:send(State, ResIQ1)};
-		       Address),
+	false ->
-    xmpp:set_from_to(ResIQ, jid:make(Server), undefined);
+	    {noreply, State}
-unauthenticated_iq_register(Acc, _Server, _IQ, _IP) ->
+    end;
+c2s_unauthenticated_packet(Acc, _) ->
     Acc.
 
 process_iq(#iq{from = From} = IQ) ->

src/mod_roster.erl

@@ -44,7 +44,7 @@
 -export([start/2, stop/1, process_iq/1, export/1,
 	 import_info/0, process_local_iq/1, get_user_roster/2,
 	 import/5, get_subscription_lists/3, get_roster/2,
-	 import_start/2, import_stop/2,
+	 import_start/2, import_stop/2, c2s_handle_info/2,
 	 get_in_pending_subscriptions/3, in_subscription/6,
 	 out_subscription/4, set_items/3, remove_user/2,
 	 get_jid_info/4, encode_item/1, webadmin_page/3,
@@ -63,6 +63,8 @@
 
 -include("ejabberd_web_admin.hrl").
 
+-define(SETS, gb_sets).
+
 -export_type([subscription/0]).
 
 -callback init(binary(), gen_mod:opts()) -> any().
@@ -102,12 +104,14 @@ start(Host, Opts) ->
 		       remove_user, 50),
     ejabberd_hooks:add(resend_subscription_requests_hook,
 		       Host, ?MODULE, get_in_pending_subscriptions, 50),
-    ejabberd_hooks:add(roster_get_versioning_feature, Host,
+    ejabberd_hooks:add(c2s_post_auth_features, Host,
 		       ?MODULE, get_versioning_feature, 50),
     ejabberd_hooks:add(webadmin_page_host, Host, ?MODULE,
 		       webadmin_page, 50),
     ejabberd_hooks:add(webadmin_user, Host, ?MODULE,
 		       webadmin_user, 50),
+    ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE,
+		       c2s_handle_info, 50),
     gen_iq_handler:add_iq_handler(ejabberd_sm, Host,
 				  ?NS_ROSTER, ?MODULE, process_iq, IQDisc).
 
@@ -128,12 +132,14 @@ stop(Host) ->
 			  ?MODULE, remove_user, 50),
     ejabberd_hooks:delete(resend_subscription_requests_hook,
 			  Host, ?MODULE, get_in_pending_subscriptions, 50),
-    ejabberd_hooks:delete(roster_get_versioning_feature,
+    ejabberd_hooks:delete(c2s_post_auth_features,
 			  Host, ?MODULE, get_versioning_feature, 50),
     ejabberd_hooks:delete(webadmin_page_host, Host, ?MODULE,
 			  webadmin_page, 50),
     ejabberd_hooks:delete(webadmin_user, Host, ?MODULE,
 			  webadmin_user, 50),
+    ejabberd_hooks:delete(c2s_handle_info, Host, ?MODULE,
+			  c2s_handle_info, 50),
     gen_iq_handler:remove_iq_handler(ejabberd_sm, Host,
 				     ?NS_ROSTER).
 
@@ -417,10 +423,8 @@ process_iq_set(#iq{from = From, to = To,
     end.
 
 push_item(User, Server, From, Item) ->
-    ejabberd_sm:route(jid:make(<<"">>, <<"">>, <<"">>),
+    ejabberd_sm:route(jid:make(User, Server, <<"">>),
-		      jid:make(User, Server, <<"">>),
+                      {item, Item#roster.jid, Item#roster.subscription}),
-                      {broadcast, {item, Item#roster.jid,
-				   Item#roster.subscription}}),
     case roster_versioning_enabled(Server) of
       true ->
 	  push_item_version(Server, User, From, Item,
@@ -460,6 +464,66 @@ push_item_version(Server, User, From, Item,
 		  end,
 		  ejabberd_sm:get_user_resources(User, Server)).
 
+c2s_handle_info({noreply, State}, {item, JID, Sub}) ->
+    {noreply, roster_change(State, JID, Sub)};
+c2s_handle_info(Acc, _) ->
+    Acc.
+
+-spec roster_change(ejabberd_c2s:state(), jid(), subscription()) -> ejabberd_c2s:state().
+roster_change(#{user := U, server := S, resource := R} = State,
+	      IJID, ISubscription) ->
+    LIJID = jid:tolower(IJID),
+    IsFrom = (ISubscription == both) or (ISubscription == from),
+    IsTo = (ISubscription == both) or (ISubscription == to),
+    PresF = maps:get(pres_f, State, ?SETS:new()),
+    PresT = maps:get(pres_t, State, ?SETS:new()),
+    OldIsFrom = ?SETS:is_element(LIJID, PresF),
+    FSet = if IsFrom -> ?SETS:add_element(LIJID, PresF);
+	      true -> ?SETS:del_element(LIJID, PresF)
+	   end,
+    TSet = if IsTo -> ?SETS:add_element(LIJID, PresT);
+	      true -> ?SETS:del_element(LIJID, PresT)
+	   end,
+    State1 = State#{pres_f => FSet, pres_t => TSet},
+    case maps:get(pres_last, State, undefined) of
+	undefined ->
+	    State1;
+	LastPres ->
+	    From = jid:make(U, S, R),
+	    PresA = maps:get(pres_a, State1, ?SETS:new()),
+	    To = jid:make(IJID),
+	    Cond1 = IsFrom andalso not OldIsFrom,
+	    Cond2 = not IsFrom andalso OldIsFrom andalso
+		?SETS:is_element(LIJID, PresA),
+	    if Cond1 ->
+		    case ejabberd_hooks:run_fold(
+			   privacy_check_packet, allow,
+			   [State1, LastPres, out]) of
+			deny ->
+			    ok;
+			allow ->
+			    Pres = xmpp:set_from_to(LastPres, From, To),
+			    ejabberd_router:route(From, To, Pres)
+		    end,
+		    A = ?SETS:add_element(LIJID, PresA),
+		    State1#{pres_a => A};
+	     Cond2 ->
+		    PU = #presence{from = From, to = To, type = unavailable},
+		    case ejabberd_hooks:run_fold(
+			   privacy_check_packet, allow,
+			   [State1, PU, out]) of
+			deny ->
+			    ok;
+			allow ->
+			    ejabberd_router:route(From, To, PU)
+		    end,
+		    A = ?SETS:del_element(LIJID, PresA),
+		    State1#{pres_a => A};
+	       true ->
+		    State1
+	    end
+    end.
+
 -spec get_subscription_lists({[ljid()], [ljid()]}, binary(), binary())
       -> {[ljid()], [ljid()]}.
 get_subscription_lists(_Acc, User, Server) ->

src/mod_shared_roster.erl

@@ -1038,11 +1038,8 @@ split_grouphost(Host, Group) ->
     end.
 
 broadcast_subscription(User, Server, ContactJid, Subscription) ->
-    ejabberd_sm:route(
+    ejabberd_sm:route(jid:make(User, Server, <<"">>),
-		      jid:make(<<"">>, Server, <<"">>),
+                      {item, ContactJid, Subscription}).
-		      jid:make(User, Server, <<"">>),
-                      {broadcast, {item, ContactJid,
-				   Subscription}}).
 
 displayed_groups_update(Members, DisplayedGroups, Subscription) ->
     lists:foreach(fun({U, S}) ->

src/xmpp_stream_in.erl

@@ -0,0 +1,698 @@
+%%%-------------------------------------------------------------------
+%%% Created : 26 Nov 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
+%%%
+%%%
+%%% ejabberd, Copyright (C) 2002-2016   ProcessOne
+%%%
+%%% This program is free software; you can redistribute it and/or
+%%% modify it under the terms of the GNU General Public License as
+%%% published by the Free Software Foundation; either version 2 of the
+%%% License, or (at your option) any later version.
+%%%
+%%% This program is distributed in the hope that it will be useful,
+%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
+%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+%%% General Public License for more details.
+%%%
+%%% You should have received a copy of the GNU General Public License along
+%%% with this program; if not, write to the Free Software Foundation, Inc.,
+%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+%%%
+%%%-------------------------------------------------------------------
+-module(xmpp_stream_in).
+-behaviour(gen_server).
+
+-protocol({rfc, 6120}).
+
+%% API
+-export([start/3, call/3, cast/2, reply/2, send/2, send_error/3,
+	 get_transport/1, change_shaper/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_cast/2, handle_call/3, handle_info/2,
+	 terminate/2, code_change/3]).
+
+-include("xmpp.hrl").
+-type state() :: map().
+-type next_state() :: {noreply, state()} | {stop, term(), state()}.
+
+-callback init(list()) -> {ok, state()} | {stop, term()} | ignore.
+-callback handle_authenticated_packet(xmpp_element(), state()) -> next_state().
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+start(Mod, Args, Opts) ->
+    gen_server:start(?MODULE, [Mod|Args], Opts).
+
+call(Ref, Msg, Timeout) ->
+    gen_server:call(Ref, Msg, Timeout).
+
+cast(Ref, Msg) ->
+    gen_server:cast(Ref, Msg).
+
+reply(Ref, Reply) ->
+    gen_server:reply(Ref, Reply).
+
+-spec send(state(), xmpp_element()) -> next_state().
+send(State, Pkt) ->
+    send_element(State, Pkt).
+
+get_transport(#{sockmod := SockMod, socket := Socket}) ->
+    SockMod:get_transport(Socket).
+
+-spec change_shaper(state(), shaper:shaper()) -> ok.
+change_shaper(#{sockmod := SockMod, socket := Socket}, Shaper) ->
+    SockMod:change_shaper(Socket, Shaper).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+init([Module, {SockMod, Socket}, Opts]) ->
+    XMLSocket = case lists:keyfind(xml_socket, 1, Opts) of
+		    {_, XS} -> XS;
+		    false -> false
+		end,
+    TLSEnabled = proplists:get_bool(tls, Opts),
+    SocketMonitor = SockMod:monitor(Socket),
+    case peername(SockMod, Socket) of
+	{ok, IP} ->
+	    State = #{mod => Module,
+		      socket => Socket,
+		      sockmod => SockMod,
+		      socket_monitor => SocketMonitor,
+		      stream_id => new_id(),
+		      stream_state => wait_for_stream,
+		      stream_restarted => false,
+		      stream_compressed => false,
+		      stream_tlsed => TLSEnabled,
+		      stream_version => {1,0},
+		      stream_authenticated => false,
+		      xml_socket => XMLSocket,
+		      xmlns => ?NS_CLIENT,
+		      lang => <<"">>,
+		      user => <<"">>,
+		      server => <<"">>,
+		      resource => <<"">>,
+		      ip => IP},
+	    Module:init([State, Opts]);
+	{error, Reason} ->
+	    {stop, Reason}
+    end.
+
+handle_cast(Cast, #{mod := Mod} = State) ->
+    Mod:handle_cast(Cast, State).
+
+handle_call(Call, From, #{mod := Mod} = State) ->
+    Mod:handle_call(Call, From, State).
+
+handle_info({'$gen_event', {xmlstreamstart, Name, Attrs}},
+	    #{stream_state := wait_for_stream} = State) ->
+    try xmpp:decode(#xmlel{name = Name, attrs = Attrs}) of
+	#stream_start{} = Pkt ->
+	    case send_header(State, Pkt) of
+		{noreply, State1} ->
+		    process_stream(Pkt, State1);
+		Err ->
+		    Err
+	    end;
+	_ ->
+	    case send_header(State) of
+		{noreply, State1} ->
+		    send_element(State1, xmpp:serr_invalid_xml());
+		Err ->
+		    Err
+	    end
+    catch _:{xmpp_codec, Why} ->
+	    case send_header(State) of
+		{noreply, State1} -> process_invalid_xml(Why, State1);
+		Err -> Err
+	    end
+    end;
+handle_info({'$gen_event', {xmlstreamend, _}}, #{mod := Mod} = State) ->
+    try Mod:handle_stream_end(State)
+    catch _:undef -> {stop, normal, State}
+    end;
+handle_info({'$gen_event', {xmlstreamerror, Reason}}, #{lang := Lang}= State) ->
+    case send_header(State) of
+	{noreply, State1} ->
+	    Err = case Reason of
+		      <<"XML stanza is too big">> ->
+			  xmpp:serr_policy_violation(Reason, Lang);
+		      _ ->
+			  xmpp:serr_not_well_formed()
+		  end,
+	    send_element(State1, Err);
+	Err ->
+	    Err
+    end;
+handle_info({'$gen_event', {xmlstreamelement, El}},
+	    #{xmlns := NS} = State) ->
+    try xmpp:decode(El, NS, [ignore_els]) of
+	Pkt ->
+	    process_element(Pkt, State)
+    catch _:{xmpp_codec, Why} ->
+	    process_invalid_xml(Why, State)
+    end;
+handle_info({'$gen_all_state_event', {xmlstreamcdata, Data}},
+	    #{mod := Mod} = State) ->
+    try Mod:handle_cdata(Data, State)
+    catch _:undef -> {noreply, State}
+    end;
+handle_info(closed, #{mod := Mod} = State) ->
+    try Mod:handle_stream_close(State)
+    catch _:undef -> {stop, normal, State}
+    end;
+handle_info({'DOWN', MRef, _Type, _Object, _Info},
+	    #{socket_monitor := MRef, mod := Mod} = State) ->
+    try Mod:handle_stream_close(State)
+    catch _:undef -> {stop, normal, State}
+    end;
+handle_info(Info, #{mod := Mod} = State) ->
+    Mod:handle_info(Info, State).
+
+terminate(Reason, #{mod := Mod, socket := Socket,
+		    sockmod := SockMod} = State) ->
+    Mod:terminate(Reason, State),
+    send_text(State, <<"</stream:stream>">>),
+    SockMod:close(Socket).
+
+code_change(OldVsn, #{mod := Mod} = State, Extra) ->
+    Mod:code_change(OldVsn, State, Extra).
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+-spec new_id() -> binary().
+new_id() ->
+    randoms:get_string().
+
+peername(SockMod, Socket) ->
+    case SockMod of
+	gen_tcp -> inet:peername(Socket);
+	_ -> SockMod:peername(Socket)
+    end.
+
+process_invalid_xml(Reason, #{lang := Lang} = State) ->
+    Txt = xmpp:io_format_error(Reason),
+    send_element(State, xmpp:serr_invalid_xml(Txt, Lang)).
+
+process_stream(#stream_start{xmlns = XML_NS,
+			     stream_xmlns = STREAM_NS},
+	       #{xmlns := NS} = State)
+  when XML_NS /= NS; STREAM_NS /= ?NS_STREAM ->
+    send_element(State, xmpp:serr_invalid_namespace());
+process_stream(#stream_start{lang = Lang},
+	       #{xmlns := ?NS_CLIENT, lang := DefaultLang} = State)
+  when size(Lang) > 35 ->
+    %% As stated in BCP47, 4.4.1:
+    %% Protocols or specifications that specify limited buffer sizes for
+    %% language tags MUST allow for language tags of at least 35 characters.
+    %% Do not store long language tag to avoid possible DoS/flood attacks
+    Txt = <<"Too long value of 'xml:lang' attribute">>,
+    send_element(State, xmpp:serr_policy_violation(Txt, DefaultLang));
+process_stream(#stream_start{to = undefined}, #{lang := Lang} = State) ->
+    Txt = <<"Missing 'to' attribute">>,
+    send_element(State, xmpp:serr_improper_addressing(Txt, Lang));
+process_stream(#stream_start{from = undefined, version = {1,0}},
+	       #{lang := Lang, xmlns := ?NS_SERVER,
+		 stream_tlsed := true} = State) ->
+    Txt = <<"Missing 'from' attribute">>,
+    send_element(State, xmpp:serr_invalid_from(Txt, Lang));
+process_stream(#stream_start{to = #jid{luser = U, lresource = R}},
+	       #{lang := Lang} = State) when U /= <<"">>; R /= <<"">> ->
+    Txt = <<"Improper 'to' attribute">>,
+    send_element(State, xmpp:serr_improper_addressing(Txt, Lang));
+process_stream(#stream_start{to = #jid{lserver = RemoteServer}},
+	       #{xmlns := ?NS_COMPONENT, mod := Mod} = State) ->
+    State1 = State#{remote_server => RemoteServer},
+    case try Mod:handle_stream_start(State1)
+	 catch _:undef -> {noreply, State1}
+	 end of
+	{noreply, State2} ->
+	    {noreply, State2#{stream_state => wait_for_handshake}};
+	Err ->
+	    Err
+    end;
+process_stream(#stream_start{to = #jid{server = Server}, from = From},
+	       #{stream_authenticated := Authenticated,
+		 stream_restarted := StreamWasRestarted,
+		 mod := Mod, xmlns := NS, resource := Resource,
+		 stream_tlsed := TLSEnabled} = State) ->
+    case if not StreamWasRestarted ->
+		 State1 = State#{server => Server},
+		 try Mod:handle_stream_start(State1)
+		 catch _:undef -> {noreply, State1}
+		 end;
+	    true ->
+		 {noreply, State}
+	 end of
+	{noreply, State2} ->
+	    State3 = if NS == ?NS_SERVER andalso TLSEnabled ->
+			     State2#{remote_server => From#jid.lserver};
+			true ->
+			     State2
+		     end,
+	    case send_features(State3) of
+		{noreply, State4} ->
+		    TLSRequired = is_starttls_required(State4),
+		    NewStreamState =
+			if not Authenticated and
+			   (not TLSEnabled and TLSRequired) ->
+				wait_for_starttls;
+			   not Authenticated ->
+				wait_for_sasl_request;
+			   (NS == ?NS_CLIENT) and (Resource == <<"">>) ->
+				wait_for_bind;
+			   true ->
+				session_established
+			end,
+		    {noreply, State4#{stream_state => NewStreamState}};
+		Err ->
+		    Err
+	    end;
+	Err ->
+	    Err
+    end.
+
+process_element(Pkt, #{stream_state := StateName, lang := Lang} = State) ->
+    case Pkt of
+	#starttls{} when StateName == wait_for_starttls;
+			 StateName == wait_for_sasl_request ->
+	    process_starttls(State);
+	#starttls{} ->
+	    send_element(State, #starttls_failure{});
+	#sasl_auth{} when StateName == wait_for_starttls ->
+	    send_element(State, #sasl_failure{reason = 'encryption-required'});
+	#sasl_auth{} when StateName == wait_for_sasl_request ->
+	    process_sasl_request(Pkt, State);
+	#sasl_auth{} ->
+	    Txt = <<"SASL negotiation is not allowed in this state">>,
+	    send_element(State, #sasl_failure{reason = 'not-authorized',
+					      text = xmpp:mk_text(Txt, Lang)});
+	#sasl_response{} when StateName == wait_for_starttls ->
+	    send_element(State, #sasl_failure{reason = 'encryption-required'});
+	#sasl_response{} when StateName == wait_for_sasl_response ->
+	    process_sasl_response(Pkt, State);
+	#sasl_response{} ->
+	    Txt = <<"SASL negotiation is not allowed in this state">>,
+	    send_element(State, #sasl_failure{reason = 'not-authorized',
+					      text = xmpp:mk_text(Txt, Lang)});
+	#sasl_abort{} when StateName == wait_for_sasl_response ->
+	    process_sasl_abort(State);
+	#sasl_abort{} ->
+	    send_element(State, #sasl_failure{reason = 'aborted'});
+	#sasl_success{} ->
+	    {noreply, State};
+	#compress{} when StateName == wait_for_sasl_response ->
+	    send_element(State, #compress_failure{reason = 'setup-failed'});
+	#compress{} ->
+	    process_compress(Pkt, State);
+	#handshake{} when StateName == wait_for_handshake ->
+	    process_handshake(Pkt, State);
+	#handshake{} ->
+	    {noreply, State};
+	_ when StateName == wait_for_sasl_request;
+	       StateName == wait_for_handshake;
+	       StateName == wait_for_sasl_response ->
+	    process_unauthenticated_packet(Pkt, State);
+	_ when StateName == wait_for_starttls ->
+	    Txt = <<"Use of STARTTLS required">>,
+	    Err = xmpp:err_policy_violation(Txt, Lang),
+	    send_error(State, Pkt, Err);
+	_ when StateName == wait_for_bind ->
+	    process_bind(Pkt, State);
+	_ when StateName == session_established ->
+	    process_authenticated_packet(Pkt, State)
+    end.
+
+process_unauthenticated_packet(Pkt, #{mod := Mod} = State) ->
+    NewPkt = set_lang(Pkt, State),
+    try Mod:handle_unauthenticated_packet(NewPkt, State)
+    catch _:undef ->
+	    Err = xmpp:err_not_authorized(),
+	    send_error(State, Pkt, Err)
+    end.
+
+process_authenticated_packet(Pkt, #{xmlns := NS, mod := Mod} = State) ->
+    Pkt1 = set_lang(Pkt, State),
+    case set_from_to(Pkt1, State) of
+	{ok, #iq{type = set, sub_els = [_]} = Pkt2} when NS == ?NS_CLIENT ->
+	    case xmpp:get_subtag(Pkt2, #xmpp_session{}) of
+		#xmpp_session{} ->
+		    send_element(State, xmpp:make_iq_result(Pkt2));
+		_ ->
+		    Mod:handle_authenticated_packet(Pkt2, State)
+	    end;
+	{ok, Pkt2} ->
+	    Mod:handle_authenticated_packet(Pkt2, State);
+	{error, Err} ->
+	    send_element(State, Err)
+    end.
+
+process_bind(#iq{type = set, sub_els = [_]} = Pkt,
+	     #{xmlns := ?NS_CLIENT, mod := Mod, lang := Lang} = State) ->
+    case xmpp:get_subtag(Pkt, #bind{}) of
+	#bind{resource = R} ->
+	    case jid:resourceprep(R) of
+		error ->
+		    Txt = <<"Malformed resource">>,
+		    Err = xmpp:err_bad_request(Txt, Lang),
+		    send_error(State, Pkt, Err);
+		_ ->
+		    case Mod:bind(R, State) of
+			{ok, #{user := U,
+			       server := S,
+			       resource := NewR} = State1} when NewR /= <<"">> ->
+			    Reply = #bind{jid = jid:make(U, S, NewR)},
+			    State2 = State1#{stream_state => session_established},
+			    send_element(State2, xmpp:make_iq_result(Pkt, Reply));
+			{error, #stanza_error{}, State1} = Err ->
+			    send_error(State1, Pkt, Err)
+		    end
+	    end;
+	_ ->
+	    try Mod:handle_unbinded_packet(Pkt, State)
+	    catch _:undef ->
+		    Err = xmpp:err_not_authorized(),
+		    send_error(State, Pkt, Err)
+	    end
+    end;
+process_bind(Pkt, #{mod := Mod} = State) ->
+    try Mod:handle_unbinded_packet(Pkt, State)
+    catch _:undef ->
+	    Err = xmpp:err_not_authorized(),
+	    send_error(State, Pkt, Err)
+    end.
+
+process_handshake(#handshake{} = Pkt, #{mod := Mod} = State) ->
+    Mod:handle_handshake(Pkt, State).
+
+process_compress(#compress{}, #{stream_compressed := true} = State) ->
+    send_element(State, #compress_failure{reason = 'setup-failed'});
+process_compress(#compress{methods = HisMethods},
+		 #{socket := Socket, sockmod := SockMod, mod := Mod} = State) ->
+    MyMethods = try Mod:compress_methods(State)
+		catch _:undef -> []
+		end,
+    CommonMethods = lists_intersection(MyMethods, HisMethods),
+    case lists:member(<<"zlib">>, CommonMethods) of
+	true ->
+	    BCompressed = fxml:element_to_binary(xmpp:encode(#compressed{})),
+	    ZlibSocket = SockMod:compress(Socket, BCompressed),
+	    State1 = State#{socket => ZlibSocket,
+			    stream_id => new_id(),
+			    stream_restarted => true,
+			    stream_state => wait_for_stream,
+			    stream_compressed => true},
+	    {noreply, State1};
+	false ->
+	    send_element(State, #compress_failure{reason = 'unsupported-method'})
+    end.
+
+process_starttls(#{socket := Socket,
+		   sockmod := SockMod, mod := Mod} = State) ->
+    TLSOpts = try Mod:tls_options(State)
+	      catch _:undef -> []
+	      end,
+    case SockMod:starttls(Socket, TLSOpts) of
+	{ok, TLSSocket} ->
+	    case send_element(State, #starttls_proceed{}) of
+		{noreply, State1} ->
+		    {noreply, State1#{socket => TLSSocket,
+				      stream_id => new_id(),
+				      stream_restarted => true,
+				      stream_state => wait_for_stream,
+				      stream_tlsed => true}};
+		Err ->
+		    Err
+	    end;
+	{error, _Reason} ->
+	    send_element(State, #starttls_failure{})
+    end.
+
+process_sasl_request(#sasl_auth{mechanism = <<"EXTERNAL">>},
+		     #{stream_tlsed := false} = State) ->
+    process_sasl_failure('encryption-required', <<"">>, State);
+process_sasl_request(#sasl_auth{mechanism = Mech, text = ClientIn},
+		     #{mod := Mod} = State) ->
+    SASLState = Mod:init_sasl(State),
+    SASLResult = cyrsasl:server_start(SASLState, Mech, ClientIn),
+    process_sasl_result(SASLResult, State).
+
+process_sasl_response(#sasl_response{text = ClientIn},
+		      #{sasl_state := SASLState} = State) ->
+    SASLResult = cyrsasl:server_step(SASLState, ClientIn),
+    process_sasl_result(SASLResult, State).
+
+process_sasl_result({ok, Props}, State) ->
+    process_sasl_success(Props, <<"">>, State);
+process_sasl_result({ok, Props, ServerOut}, State) ->
+    process_sasl_success(Props, ServerOut, State);
+process_sasl_result({continue, ServerOut, NewSASLState}, State) ->
+    process_sasl_continue(ServerOut, NewSASLState, State);
+process_sasl_result({error, Reason, User}, State) ->
+    process_sasl_failure(Reason, User, State);
+process_sasl_result({error, Reason}, State) ->
+    process_sasl_failure(Reason, <<"">>, State).
+
+process_sasl_success(Props, ServerOut,
+		     #{socket := Socket, sockmod := SockMod,
+		       mod := Mod, sasl_state := SASLState} = State) ->
+    Mech = cyrsasl:get_mech(SASLState),
+    User = identity(Props),
+    AuthModule = proplists:get_value(auth_module, Props),
+    case try Mod:handle_auth_success(User, Mech, AuthModule, State)
+	 catch _:undef -> {noreply, State}
+	 end of
+	{noreply, State1} ->
+	    SockMod:reset_stream(Socket),
+	    case send_element(State1, #sasl_success{text = ServerOut}) of
+		{noreply, State2} ->
+		    State3 = maps:remove(sasl_state, State2),
+		    {noreply, State3#{stream_id => new_id(),
+				      stream_authenticated => true,
+				      stream_restarted => true,
+				      stream_state => wait_for_stream,
+				      user => User}};
+		Err ->
+		    Err
+	    end;
+	Err ->
+	    Err
+    end.
+
+process_sasl_continue(ServerOut, NewSASLState, State) ->
+    send_element(State, #sasl_challenge{text = ServerOut}),
+    {noreply, State#{sasl_state => NewSASLState,
+		     stream_state => wait_for_sasl_response}}.
+
+process_sasl_failure(Reason, User,
+		     #{mod := Mod, sasl_state := SASLState} = State) ->
+    Mech = cyrsasl:get_mech(SASLState),
+    case try Mod:handle_auth_failure(User, Mech, Reason, State)
+	 catch _:undef -> {noreply, State}
+	 end of
+	{noreply, State1} ->
+	    State2 = maps:remove(sasl_state, State1),
+	    State3 = State2#{stream_state => wait_for_sasl_request},
+	    send_element(State3, #sasl_failure{reason = Reason});
+	Err ->
+	    Err
+    end.
+
+process_sasl_abort(State) ->
+    process_sasl_failure('aborted', <<"">>, State).
+
+send_features(#{stream_version := {1,0},
+		stream_tlsed := TLSEnabled} = State) ->
+    TLSRequired = is_starttls_required(State),
+    Features = if TLSRequired and not TLSEnabled ->
+		       get_tls_feature(State);
+		  true ->
+		       get_sasl_feature(State) ++ get_compress_feature(State)
+			   ++ get_tls_feature(State) ++ get_bind_feature(State)
+			   ++ get_session_feature(State) ++ get_other_features(State)
+	       end,
+    send_element(State, #stream_features{sub_els = Features});
+send_features(State) ->
+    %% clients from stone age
+    {noreply, State}.
+
+get_sasl_feature(#{stream_authenticated := false,
+		   mod := Mod,
+		   stream_tlsed := TLSEnabled} = State) ->
+    TLSRequired = is_starttls_required(State),
+    if TLSEnabled or not TLSRequired ->
+	    try Mod:sasl_mechanisms(State) of
+		[] -> [];
+		List -> [#sasl_mechanisms{list = List}]
+	    catch _:undef ->
+		    []
+	    end;
+       true ->
+	    []
+    end;
+get_sasl_feature(_) ->
+    [].
+
+get_compress_feature(#{stream_compressed := false, mod := Mod} = State) ->
+    try Mod:compress_methods(State) of
+	[] -> [];
+	Ms -> [#compression{methods = Ms}]
+    catch _:undef ->
+	    []
+    end;
+get_compress_feature(_) ->
+    [].
+
+get_tls_feature(#{stream_authenticated := false,
+		  stream_tlsed := false} = State) ->
+    TLSRequired = is_starttls_required(State),
+    [#starttls{required = TLSRequired}];
+get_tls_feature(_) ->
+    [].
+
+get_bind_feature(#{stream_authenticated := true, resource := <<"">>}) ->
+    [#bind{}];
+get_bind_feature(_) ->
+    [].
+
+get_session_feature(#{stream_authenticated := true, resource := <<"">>}) ->
+    [#xmpp_session{optional = true}];
+get_session_feature(_) ->
+    [].
+
+get_other_features(#{stream_authenticated := Auth, mod := Mod} = State) ->
+    try
+	if Auth -> Mod:authenticated_stream_features(State);
+	   true -> Mod:unauthenticated_stream_features(State)
+	end
+    catch _:undef ->
+	    []
+    end.
+
+is_starttls_required(#{mod := Mod} = State) ->
+    try Mod:tls_required(State)
+    catch _:undef -> false
+    end.
+
+set_from_to(Pkt, _State) when not ?is_stanza(Pkt) ->
+    {ok, Pkt};
+set_from_to(Pkt, #{user := U, server := S, resource := R,
+		     xmlns := ?NS_CLIENT}) ->
+    JID = jid:make(U, S, R),
+    From = case xmpp:get_from(Pkt) of
+	       undefined -> JID;
+	       F -> F
+	   end,
+    if JID#jid.luser == From#jid.luser andalso
+       JID#jid.lserver == From#jid.lserver andalso
+       (JID#jid.lresource == From#jid.lresource
+	orelse From#jid.lresource == <<"">>) ->
+	    To = case xmpp:get_to(Pkt) of
+		     undefined -> jid:make(U, S);
+		     T -> T
+		 end,
+	    {ok, xmpp:set_from_to(Pkt, JID, To)};
+       true ->
+	    {error, xmpp:serr_invalid_from()}
+    end;
+set_from_to(Pkt, #{lang := Lang}) ->
+    From = xmpp:get_from(Pkt),
+    To = xmpp:get_to(Pkt),
+    if From == undefined ->
+	    Txt = <<"Missing 'from' attribute">>,
+	    {error, xmpp:serr_invalid_from(Txt, Lang)};
+       To == undefined ->
+	    Txt = <<"Missing 'to' attribute">>,
+	    {error, xmpp:serr_improper_addressing(Txt, Lang)};
+       true ->
+	    {ok, Pkt}
+    end.
+
+send_header(State) ->
+    send_header(State, #stream_start{}).
+
+send_header(#{stream_state := wait_for_stream,
+	      stream_id := StreamID,
+	      stream_version := MyVersion,
+	      lang := MyLang,
+	      xmlns := NS,
+	      server := DefaultServer} = State,
+	    #stream_start{to = To, lang = HisLang, version = HisVersion}) ->
+    Lang = choose_lang(MyLang, HisLang),
+    From = case To of
+	       #jid{} -> To;
+	       undefined -> jid:make(DefaultServer)
+	   end,
+    Version = case HisVersion of
+		  undefined -> MyVersion;
+		  _ -> HisVersion
+	      end,
+    Header = xmpp:encode(#stream_start{version = Version,
+				       lang = Lang,
+				       xmlns = NS,
+				       stream_xmlns = ?NS_STREAM,
+				       id = StreamID,
+				       from = From}),
+    State1 = State#{lang => Lang},
+    case send_text(State1, fxml:element_to_header(Header)) of
+	ok -> {noreply, State1};
+	{error, _} -> {stop, normal, State1}
+    end;
+send_header(State, _) ->
+    {noreply, State}.
+
+send_element(#{xmlns := NS, mod := Mod} = State, Pkt) ->
+    El = xmpp:encode(Pkt, NS),
+    Data = fxml:element_to_binary(El),
+    case send_text(State, Data) of
+	ok when is_record(Pkt, stream_error) ->
+	    {stop, normal, State};
+	ok when is_record(Pkt, starttls_failure) ->
+	    {stop, normal, State};
+	Res ->
+	    try Mod:handle_send(Res, Pkt, El, Data, State)
+	    catch _:undef when Res == ok ->
+		    {noreply, State};
+		  _:undef ->
+		    {stop, normal, State}
+	    end
+    end.
+
+send_error(State, Pkt, Err) when ?is_stanza(Pkt) ->
+    case xmpp:get_type(Pkt) of
+	result -> {noreply, State};
+	error -> {noreply, State};
+	_ ->
+	    ErrPkt = xmpp:make_error(Pkt, Err),
+	    send_element(State, ErrPkt)
+    end;
+send_error(State, _, _) ->
+    {noreply, State}.
+
+send_text(#{socket := Sock, sockmod := SockMod}, Data) ->
+    SockMod:send(Sock, Data).
+
+choose_lang(Lang, <<"">>) -> Lang;
+choose_lang(_, Lang) -> Lang.
+
+set_lang(Pkt, #{lang := MyLang, xmlns := ?NS_CLIENT}) when ?is_stanza(Pkt) ->
+    HisLang = xmpp:get_lang(Pkt),
+    Lang = choose_lang(MyLang, HisLang),
+    xmpp:set_lang(Pkt, Lang);
+set_lang(Pkt, _) ->
+    Pkt.
+
+lists_intersection(L1, L2) ->
+    lists:filter(
+      fun(E) ->
+	      lists:member(E, L2)
+      end, L1).
+
+identity(Props) ->
+    case proplists:get_value(authzid, Props, <<>>) of
+	<<>> -> proplists:get_value(username, Props, <<>>);
+	AuthzId -> AuthzId
+    end.