Software /
code /
prosody
File
plugins/mod_c2s.lua @ 11906:ba3344926e18
MUC: Add option to include form in registration query
This was originally not done based on my interpretation of XEP-0045. Today's
reading, however, revealed that it actually says the result
> SHOULD contain **at least** a <username/> element
(emphasis mine)
I take this to mean that including a form **is** allowed (and I think this is
sensible). Tigase already includes the form I believe.
I've gated the new behaviour behind a (default off) option, because it hasn't
been tested for compatibility with clients. My primary desire for it is in
Snikket, where the clients will be tested to ensure compatibility with this.
I don't anticipate that (m)any clients would break, so maybe after 0.12 we can
experiment with enabling it by default and eventually remove the option.
author | Matthew Wild <mwild1@gmail.com> |
---|---|
date | Mon, 15 Nov 2021 16:11:03 +0000 |
parent | 11899:6d06068363aa |
child | 11974:a7c183bb4e64 |
line wrap: on
line source
-- Prosody IM -- Copyright (C) 2008-2010 Matthew Wild -- Copyright (C) 2008-2010 Waqas Hussain -- -- This project is MIT/X11 licensed. Please see the -- COPYING file in the source package for more information. -- module:set_global(); 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 tostring, type = tostring, type; local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; local log = module._log; local c2s_timeout = module:get_option_number("c2s_timeout", 300); 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:metric("gauge", "connections", "", "Established c2s connections", {"host", "type", "ip_family"}); local sessions = module:shared("sessions"); local core_process_stanza = prosody.core_process_stanza; local hosts = prosody.hosts; 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 () -- 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 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 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", text = "A valid 'to' attribute is required on stream headers" }; return; end if not session.host then session.host = host; elseif session.host ~= host then session:close{ condition = "not-authorized", text = "The 'to' attribute must remain the same across stream restarts" }; return; end session.version = tonumber(attr.version) or 0; session.streamid = uuid_generate(); (session.log or log)("debug", "Client sent opening <stream:stream> to %s", session.host); if not hosts[session.host] or not hosts[session.host].modules.c2s then -- We don't serve this host... session:close{ condition = "host-unknown", text = "This server does not serve "..tostring(session.host)}; return; end 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; -- If session.secure is *false* (not nil) then it means we /were/ encrypting -- since we now have a new stream header, session is secured if session.secure == false then session.secure = true; session.encrypted = true; local sock = session.conn:socket(); 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 local features = st.stanza("stream:features"); hosts[session.host].events.fire_event("stream-features", { origin = session, features = features }); if features.tags[1] or session.full_jid then send(features); else 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, 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 function stream_callbacks.error(session, error, data) if error == "no-stream" then 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", data); session:close("not-well-formed"); elseif error == "stream-error" then local condition, text = "undefined-condition"; for child in data:childtags(nil, xmlns_xmpp_streams) do if child.name ~= "text" then condition = child.name; else text = child:get_text(); end if condition ~= "undefined-condition" and text then break; end end text = condition .. (text and (" ("..text..")") or ""); session.log("info", "Session closed by remote with error: %s", text); session:close(nil, text); end end function stream_callbacks.handlestanza(session, stanza) stanza = session.filter("stanzas/in", stanza); session.thread:run(stanza); end --- Session methods local function session_close(session, reason) local log = session.log or log; if session.conn then if session.notopen then session:open_stream(); end if reason then -- nil == no err, initiated by us, false == initiated by client 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(); if reason.text then stream_error:tag("text", stream_xmlns_attr):text(reason.text):up(); end if reason.extra then stream_error:add_child(reason.extra); end end end stream_error = tostring(stream_error); log("debug", "Disconnecting client, <stream:error> is: %s", stream_error); session.send(stream_error); end session.send("</stream:stream>"); function session.send() return false; end local reason_text = (reason and (reason.name or reason.text or reason.condition)) or reason; session.log("debug", "c2s stream for %s closed: %s", session.full_jid or session.ip or "<unknown>", reason_text or "session closed"); -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote local conn = session.conn; if reason_text == nil and not session.notopen and session.type == "c2s" then -- Grace time to process data from authenticated cleanly-closed stream add_task(stream_close_timeout, function () if not session.destroyed then session.log("warn", "Failed to receive a stream close response, closing connection anyway..."); sm_destroy_session(session, reason_text); if conn then conn:close(); end end end); else sm_destroy_session(session, reason_text); if conn then conn:close(); end end else local reason_text = (reason and (reason.name or reason.text or reason.condition)) or reason; sm_destroy_session(session, reason_text); end end -- 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 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 self.data.conn:resume(); else (self.data.log or log)("debug", "Session has no connection to resume"); end end function runner_callbacks:waiting() if self.data.conn then self.data.conn:pause(); else (self.data.log or log)("debug", "Session has no connection to pause while waiting"); end end function runner_callbacks:error(err) (self.data.log or log)("error", "Traceback[c2s]: %s", err); end --- Port listener function listener.onconnect(conn) local session = sm_new_session(conn); session.log("info", "Client connected"); -- 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(); 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 if opt_keepalives then conn:setoption("keepalive", opt_keepalives); end session.close = session_close; local stream = new_xmpp_stream(session, stream_callbacks, stanza_size_limit); session.stream = stream; session.notopen = true; function session.reset_stream() session.notopen = true; session.stream:reset(); end session.thread = runner(function (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; function session.data(data) -- Parse the data, which will store stanzas in session.pending_stanzas if data then data = filter("bytes/in", data); if data then local ok, err = stream:feed(data); if not ok then 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 end 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); end session.dispatch_stanza = stream_callbacks.handlestanza; sessions[conn] = session; end function listener.onincoming(conn, data) local session = sessions[conn]; if session then session.data(data); end end function listener.ondisconnect(conn, err) local session = sessions[conn]; if session then (session.log or log)("info", "Client disconnected: %s", err or "connection closed"); sm_destroy_session(session, err); session.conn = nil; sessions[conn] = nil; end end function listener.onreadtimeout(conn) local session = sessions[conn]; if session then return (hosts[session.host] or prosody).events.fire_event("c2s-read-timeout", { session = session }); 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 return event.session.send(' '); end end function listener.associate_session(conn, session) sessions[conn] = session; end function module.add_host(module) module:hook("c2s-read-timeout", keepalive, -1); end module:hook("c2s-read-timeout", keepalive, -1); module:hook("server-stopping", function(event) local reason = event.reason; for _, session in pairs(sessions) do session:close{ condition = "system-shutdown", text = reason }; end end, -100); module:provides("net", { name = "c2s"; listener = listener; 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; encryption = "ssl"; multiplex = { pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>"; }; });