1 files changed, 151 insertions, 50 deletions

diff --git a/plugins/mod_c2s.lua b/plugins/mod_c2s.lua
index f9c2e9fb..c8f54fa7 100644
--- a/plugins/mod_c2s.lua
+++ b/plugins/mod_c2s.lua
@@ -12,10 +12,12 @@ local add_task = require "util.timer".add_task;
 local new_xmpp_stream = require "util.xmppstream".new;
 local nameprep = require "util.encodings".stringprep.nameprep;
 local sessionmanager = require "core.sessionmanager";
+local statsmanager = require "core.statsmanager";
 local st = require "util.stanza";
 local sm_new_session, sm_destroy_session = sessionmanager.new_session, sessionmanager.destroy_session;
 local uuid_generate = require "util.uuid".generate;
-local runner = require "util.async".runner;
+local async = require "util.async";
+local runner = async.runner;
 
 local tostring, type = tostring, type;
 
@@ -28,8 +30,7 @@ local stream_close_timeout = module:get_option_number("c2s_close_timeout", 5);
 local opt_keepalives = module:get_option_boolean("c2s_tcp_keepalives", module:get_option_boolean("tcp_keepalives", true));
 local stanza_size_limit = module:get_option_number("c2s_stanza_size_limit", 1024*256);
 
-local measure_connections = module:measure("connections", "amount");
-local measure_ipv6 = module:measure("ipv6", "amount");
+local measure_connections = module:metric("gauge", "connections", "", "Established c2s connections", {"host", "type", "ip_family"});
 
 local sessions = module:shared("sessions");
 local core_process_stanza = prosody.core_process_stanza;
@@ -39,24 +40,46 @@ local stream_callbacks = { default_ns = "jabber:client" };
 local listener = {};
 local runner_callbacks = {};
 
+local m_tls_params = module:metric(
+	"counter", "encrypted", "",
+	"Encrypted connections",
+	{"protocol"; "cipher"}
+);
+
 module:hook("stats-update", function ()
-	local count = 0;
-	local ipv6 = 0;
+	-- for push backends, avoid sending out updates for each increment of
+	-- the metric below.
+	statsmanager.cork()
+	measure_connections:clear()
 	for _, session in pairs(sessions) do
-		count = count + 1;
-		if session.ip and session.ip:match(":") then
-			ipv6 = ipv6 + 1;
-		end
+		local host = session.host or ""
+		local type_ = session.type or "other"
+
+		-- we want to expose both v4 and v6 counters in all cases to make
+		-- queries smoother
+		local is_ipv6 = session.ip and session.ip:match(":") and 1 or 0
+		local is_ipv4 = 1 - is_ipv6
+		measure_connections:with_labels(host, type_, "ipv4"):add(is_ipv4)
+		measure_connections:with_labels(host, type_, "ipv6"):add(is_ipv6)
 	end
-	measure_connections(count);
-	measure_ipv6(ipv6);
+	statsmanager.uncork()
 end);
 
 --- Stream events handlers
 local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
 
 function stream_callbacks.streamopened(session, attr)
+	-- run _streamopened in async context
+	session.thread:run({ stream = "opened", attr = attr });
+end
+
+function stream_callbacks._streamopened(session, attr)
 	local send = session.send;
+	if not attr.to then
+		session:close{ condition = "improper-addressing",
+			text = "A 'to' attribute is required on stream headers" };
+		return;
+	end
 	local host = nameprep(attr.to);
 	if not host then
 		session:close{ condition = "improper-addressing",
@@ -80,7 +103,10 @@ function stream_callbacks.streamopened(session, attr)
 		return;
 	end
 
-	session:open_stream();
+	session:open_stream(host, attr.from);
+
+	-- Opening the stream can cause the stream to be closed
+	if session.destroyed then return end
 
 	(session.log or log)("debug", "Sent reply <stream:stream> to client");
 	session.notopen = nil;
@@ -92,13 +118,13 @@ function stream_callbacks.streamopened(session, attr)
 		session.encrypted = true;
 
 		local sock = session.conn:socket();
-		if sock.info then
-			local info = sock:info();
+		local info = sock.info and sock:info();
+		if type(info) == "table" then
 			(session.log or log)("info", "Stream encrypted (%s with %s)", info.protocol, info.cipher);
 			session.compressed = info.compression;
+			m_tls_params:with_labels(info.protocol, info.cipher):add(1)
 		else
 			(session.log or log)("info", "Stream encrypted");
-			session.compressed = sock.compression and sock:compression(); --COMPAT mw/luasec-hg
 		end
 	end
 
@@ -107,15 +133,23 @@ function stream_callbacks.streamopened(session, attr)
 	if features.tags[1] or session.full_jid then
 		send(features);
 	else
-		(session.log or log)("warn", "No stream features to offer");
-		session:close({
-			condition = "undefined-condition";
-			text = "No stream features to proceed with on "..(session.secure and "" or "in").."secure stream";
-		});
+		if session.secure then
+			-- Here SASL should be offered
+			(session.log or log)("warn", "No stream features to offer on secure session. Check authentication settings.");
+		else
+			-- Normally STARTTLS would be offered
+			(session.log or log)("warn", "No stream features to offer on insecure session. Check encryption and security settings.");
+		end
+		session:close{ condition = "undefined-condition", text = "No stream features to proceed with" };
 	end
 end
 
-function stream_callbacks.streamclosed(session)
+function stream_callbacks.streamclosed(session, attr)
+	-- run _streamclosed in async context
+	session.thread:run({ stream = "closed", attr = attr });
+end
+
+function stream_callbacks._streamclosed(session)
 	session.log("debug", "Received </stream:stream>");
 	session:close(false);
 end
@@ -125,7 +159,7 @@ function stream_callbacks.error(session, error, data)
 		session.log("debug", "Invalid opening stream header (%s)", (data:gsub("^([^\1]+)\1", "{%1}")));
 		session:close("invalid-namespace");
 	elseif error == "parse-error" then
-		(session.log or log)("debug", "Client XML parse error: %s", tostring(data));
+		(session.log or log)("debug", "Client XML parse error: %s", data);
 		session:close("not-well-formed");
 	elseif error == "stream-error" then
 		local condition, text = "undefined-condition";
@@ -153,6 +187,9 @@ end
 --- Session methods
 local function session_close(session, reason)
 	local log = session.log or log;
+	local close_event_payload = { session = session, reason = reason };
+	module:context(session.host):fire_event("pre-session-close", close_event_payload);
+	reason = close_event_payload.reason;
 	if session.conn then
 		if session.notopen then
 			session:open_stream();
@@ -161,6 +198,8 @@ local function session_close(session, reason)
 			local stream_error = st.stanza("stream:error");
 			if type(reason) == "string" then -- assume stream error
 				stream_error:tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' });
+			elseif st.is_stanza(reason) then
+				stream_error = reason;
 			elseif type(reason) == "table" then
 				if reason.condition then
 					stream_error:tag(reason.condition, stream_xmlns_attr):up();
@@ -170,8 +209,6 @@ local function session_close(session, reason)
 					if reason.extra then
 						stream_error:add_child(reason.extra);
 					end
-				elseif reason.name then -- a stanza
-					stream_error = reason;
 				end
 			end
 			stream_error = tostring(stream_error);
@@ -206,27 +243,25 @@ local function session_close(session, reason)
 	end
 end
 
-module:hook_global("user-deleted", function(event)
-	local username, host = event.username, event.host;
-	local user = hosts[host].sessions[username];
-	if user and user.sessions then
-		for _, session in pairs(user.sessions) do
-			session:close{ condition = "not-authorized", text = "Account deleted" };
-		end
-	end
-end, 200);
-
-module:hook_global("user-password-changed", function(event)
-	local username, host, resource = event.username, event.host, event.resource;
-	local user = hosts[host].sessions[username];
-	if user and user.sessions then
-		for r, session in pairs(user.sessions) do
-			if r ~= resource then
-				session:close{ condition = "reset", text = "Password changed" };
+-- Close all user sessions with the specified reason. If leave_resource is
+-- true, the resource named by event.resource will not be closed.
+local function disconnect_user_sessions(reason, leave_resource)
+	return function (event)
+		local username, host, resource = event.username, event.host, event.resource;
+		local user = hosts[host].sessions[username];
+		if user and user.sessions then
+			for r, session in pairs(user.sessions) do
+				if not leave_resource or r ~= resource then
+					session:close(reason);
+				end
 			end
 		end
 	end
-end, 200);
+end
+
+module:hook_global("user-password-changed", disconnect_user_sessions({ condition = "reset", text = "Password changed" }, true), 200);
+module:hook_global("user-roles-changed", disconnect_user_sessions({ condition = "reset", text = "Roles changed" }), 200);
+module:hook_global("user-deleted", disconnect_user_sessions({ condition = "not-authorized", text = "Account deleted" }), 200);
 
 function runner_callbacks:ready()
 	if self.data.conn then
@@ -254,17 +289,20 @@ function listener.onconnect(conn)
 
 	session.log("info", "Client connected");
 
-	-- Client is using legacy SSL (otherwise mod_tls sets this flag)
+	-- Client is using Direct TLS or legacy SSL (otherwise mod_tls sets this flag)
 	if conn:ssl() then
 		session.secure = true;
 		session.encrypted = true;
 
 		-- Check if TLS compression is used
 		local sock = conn:socket();
-		if sock.info then
-			session.compressed = sock:info"compression";
-		elseif sock.compression then
-			session.compressed = sock:compression(); --COMPAT mw/luasec-hg
+		local info = sock.info and sock:info();
+		if type(info) == "table" then
+			(session.log or log)("info", "Stream encrypted (%s with %s)", info.protocol, info.cipher);
+			session.compressed = info.compression;
+			m_tls_params:with_labels(info.protocol, info.cipher):add(1)
+		else
+			(session.log or log)("info", "Stream encrypted");
 		end
 	end
 
@@ -284,7 +322,13 @@ function listener.onconnect(conn)
 	end
 
 	session.thread = runner(function (stanza)
-		core_process_stanza(session, stanza);
+		if st.is_stanza(stanza) then
+			core_process_stanza(session, stanza);
+		elseif stanza.stream == "opened" then
+			stream_callbacks._streamopened(session, stanza.attr);
+		elseif stanza.stream == "closed" then
+			stream_callbacks._streamclosed(session, stanza.attr);
+		end
 	end, runner_callbacks, session);
 
 	local filter = session.filter;
@@ -295,8 +339,16 @@ function listener.onconnect(conn)
 			if data then
 				local ok, err = stream:feed(data);
 				if not ok then
-					log("debug", "Received invalid XML (%s) %d bytes: %s", tostring(err), #data, data:sub(1, 300):gsub("[\r\n]+", " "):gsub("[%z\1-\31]", "_"));
-					session:close("not-well-formed");
+					log("debug", "Received invalid XML (%s) %d bytes: %q", err, #data, data:sub(1, 300));
+					if err == "stanza-too-large" then
+						session:close({
+							condition = "policy-violation",
+							text = "XML stanza is too big",
+							extra = st.stanza("stanza-too-big", { xmlns = 'urn:xmpp:errors' }),
+						});
+					else
+						session:close("not-well-formed");
+					end
 				end
 			end
 		end
@@ -305,6 +357,7 @@ function listener.onconnect(conn)
 	if c2s_timeout then
 		add_task(c2s_timeout, function ()
 			if session.type == "c2s_unauthed" then
+				(session.log or log)("debug", "Connection still not authenticated after c2s_timeout=%gs, closing it", c2s_timeout);
 				session:close("connection-timeout");
 			end
 		end);
@@ -330,6 +383,7 @@ function listener.ondisconnect(conn, err)
 		session.conn = nil;
 		sessions[conn]  = nil;
 	end
+	module:fire_event("c2s-closed", { session = session; conn = conn });
 end
 
 function listener.onreadtimeout(conn)
@@ -339,6 +393,20 @@ function listener.onreadtimeout(conn)
 	end
 end
 
+function listener.ondrain(conn)
+	local session = sessions[conn];
+	if session then
+		return (hosts[session.host] or prosody).events.fire_event("c2s-ondrain", { session = session });
+	end
+end
+
+function listener.onpredrain(conn)
+	local session = sessions[conn];
+	if session then
+		return (hosts[session.host] or prosody).events.fire_event("c2s-pre-ondrain", { session = session });
+	end
+end
+
 local function keepalive(event)
 	local session = event.session;
 	if not session.notopen then
@@ -356,11 +424,33 @@ end
 
 module:hook("c2s-read-timeout", keepalive, -1);
 
+module:hook("server-stopping", function(event) -- luacheck: ignore 212/event
+	-- Close ports
+	local pm = require "core.portmanager";
+	for _, netservice in pairs(module.items["net-provider"]) do
+		pm.unregister_service(netservice.name, netservice);
+	end
+end, -80);
+
 module:hook("server-stopping", function(event)
+	local wait, done = async.waiter(1, true);
+	module:hook("c2s-closed", function ()
+		if next(sessions) == nil then done(); end
+	end)
+
+	-- Close sessions
 	local reason = event.reason;
 	for _, session in pairs(sessions) do
 		session:close{ condition = "system-shutdown", text = reason };
 	end
+
+	-- Wait for them to close properly if they haven't already
+	if next(sessions) ~= nil then
+		add_task(stream_close_timeout+1, function () done() end);
+		module:log("info", "Waiting for sessions to close");
+		wait();
+	end
+
 end, -100);
 
 
@@ -371,10 +461,21 @@ module:provides("net", {
 	default_port = 5222;
 	encryption = "starttls";
 	multiplex = {
+		protocol = "xmpp-client";
+		pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>";
+	};
+});
+
+module:provides("net", {
+	name = "c2s_direct_tls";
+	listener = listener;
+	encryption = "ssl";
+	multiplex = {
 		pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>";
 	};
 });
 
+-- COMPAT
 module:provides("net", {
 	name = "legacy_ssl";
 	listener = listener;