diff options
Diffstat (limited to 'plugins/mod_c2s.lua')
| -rw-r--r-- | plugins/mod_c2s.lua | 86 |
1 files changed, 66 insertions, 20 deletions
diff --git a/plugins/mod_c2s.lua b/plugins/mod_c2s.lua index f19f9df5..38a275f5 100644 --- a/plugins/mod_c2s.lua +++ b/plugins/mod_c2s.lua @@ -12,6 +12,7 @@ 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; @@ -28,8 +29,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; @@ -40,23 +40,39 @@ local listener = {}; local runner_callbacks = {}; 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 +96,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; @@ -98,7 +117,6 @@ function stream_callbacks.streamopened(session, attr) session.compressed = info.compression; else (session.log or log)("info", "Stream encrypted"); - session.compressed = sock.compression and sock:compression(); --COMPAT mw/luasec-hg end end @@ -107,12 +125,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"); + 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 @@ -122,7 +151,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"; @@ -260,8 +289,6 @@ function listener.onconnect(conn) 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 end end @@ -281,7 +308,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; @@ -292,8 +325,12 @@ 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" }); + else + session:close("not-well-formed"); + end end end end @@ -302,6 +339,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); @@ -336,6 +374,13 @@ 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 + local function keepalive(event) local session = event.session; if not session.notopen then @@ -368,6 +413,7 @@ module:provides("net", { default_port = 5222; encryption = "starttls"; multiplex = { + protocol = "xmpp-client"; pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>"; }; }); |