Software /
code /
prosody
File
plugins/mod_c2s.lua @ 11359:db146bf7b120
mod_c2s: Reflect stream 'from' attribute back if set (fix #1625)
Clients should *not* be setting this before TLS anyways.
author | Kim Alvefur <zash@zash.se> |
---|---|
date | Wed, 03 Feb 2021 21:41:45 +0100 |
parent | 11120:b2331f3dfeea |
child | 11477:c90ef8745779 |
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 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"); -- TODO come up with a sensible default (util.xmppstream defaults to 10M) local measure_connections = module:measure("connections", "amount"); local measure_ipv6 = module:measure("ipv6", "amount"); 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 = {}; module:hook("stats-update", function () local count = 0; local ipv6 = 0; for _, session in pairs(sessions) do count = count + 1; if session.ip and session.ip:match(":") then ipv6 = ipv6 + 1; end end measure_connections(count); measure_ipv6(ipv6); 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); (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(); if sock.info then local info = sock:info(); (session.log or log)("info", "Stream encrypted (%s with %s)", info.protocol, info.cipher); session.compressed = info.compression; 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 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 elseif reason.name then -- a stanza stream_error = reason; 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); conn:close(); end end); else sm_destroy_session(session, reason_text); conn:close(); 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 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" }; end end end end, 200); function runner_callbacks:ready() self.data.conn:resume(); end function runner_callbacks:waiting() self.data.conn:pause(); 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 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"; 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" }); 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: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 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 = "legacy_ssl"; listener = listener; encryption = "ssl"; multiplex = { pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>"; }; });