From 3f2c23f051d3b06c05b201f68987b2bae8a19ef9 Mon Sep 17 00:00:00 2001 From: Victor Seva Date: Mon, 7 Apr 2014 11:17:08 +0200 Subject: [PATCH] MT#6407 prosody cluster Change-Id: I77a67d04fe88905eb137e276f9113e5ab37778ec --- core/s2scmanager.lua | 100 +++ debian/ngcp-prosody-modules.install | 1 + plugins/mod_s2s_cluster/mod_s2s_cluster.lua | 690 ++++++++++++++++++++ plugins/mod_s2s_cluster/s2scout.lib.lua | 158 +++++ plugins/mod_s2sc_dialback.lua | 185 ++++++ plugins/mod_sipwise_cluster.lua | 304 +++++++++ plugins/mod_sipwise_redis_sessions.lua | 14 +- util/table.lua | 22 +- 8 files changed, 1458 insertions(+), 16 deletions(-) create mode 100644 core/s2scmanager.lua create mode 100644 plugins/mod_s2s_cluster/mod_s2s_cluster.lua create mode 100644 plugins/mod_s2s_cluster/s2scout.lib.lua create mode 100644 plugins/mod_s2sc_dialback.lua create mode 100644 plugins/mod_sipwise_cluster.lua diff --git a/core/s2scmanager.lua b/core/s2scmanager.lua new file mode 100644 index 0000000..b948bb8 --- /dev/null +++ b/core/s2scmanager.lua @@ -0,0 +1,100 @@ +-- Prosody IM +-- Copyright (C) 2008-2010 Matthew Wild +-- Copyright (C) 2008-2010 Waqas Hussain +-- Copyright (C) 2014 Sipwise GmbH +-- +-- This project is MIT/X11 licensed. Please see the +-- COPYING file in the source package for more information. +-- + + + +local hosts = prosody.hosts; +local tostring, pairs, setmetatable + = tostring, pairs, setmetatable; + +local logger_init = require "util.logger".init; + +local log = logger_init("s2scmanager"); + +local prosody = _G.prosody; +incoming_s2sc = {}; +prosody.incoming_s2sc = incoming_s2sc; +local incoming_s2sc = incoming_s2sc; +local fire_event = prosody.events.fire_event; + +module "s2scmanager" + +function new_incoming(conn) + local session = { conn = conn, type = "s2scin_unauthed", direction = "incoming", hosts = {} }; + session.log = logger_init("s2scin"..tostring(session):match("[a-f0-9]+$")); + incoming_s2sc[session] = true; + return session; +end + +function new_outgoing(from_host, to_host) + local host_session = { to_host = to_host, from_host = from_host, host = from_host, + notopen = true, type = "s2scout_unauthed", direction = "outgoing" }; + if not hosts[from_host].s2scout then hosts[from_host].s2scout = {} end + hosts[from_host].s2scout[to_host] = host_session; + local conn_name = "s2scout"..tostring(host_session):match("[a-f0-9]*$"); + host_session.log = logger_init(conn_name); + return host_session; +end + +local resting_session = { -- Resting, not dead + destroyed = true; + type = "s2sc_destroyed"; + open_stream = function (session) + session.log("debug", "Attempt to open stream on resting session"); + end; + close = function (session) + session.log("debug", "Attempt to close already-closed session"); + end; + filter = function (type, data) return data; end; + }; resting_session.__index = resting_session; + +function retire_session(session, reason) + local log = session.log or log; + for k in pairs(session) do + if k ~= "log" and k ~= "id" and k ~= "conn" then + session[k] = nil; + end + end + + session.destruction_reason = reason; + + function session.send(data) log("debug", "Discarding data sent to resting session: %s", tostring(data)); end + function session.data(data) log("debug", "Discarding data received from resting session: %s", tostring(data)); end + return setmetatable(session, resting_session); +end + +function destroy_session(session, reason) + if session.destroyed then return; end + (session.log or log)("debug", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host)..(reason and (": "..reason) or "")); + + if session.direction == "outgoing" then + hosts[session.from_host].s2scout[session.to_host] = nil; + session:bounce_sendq(reason); + elseif session.direction == "incoming" then + incoming_s2sc[session] = nil; + end + + local event_data = { session = session, reason = reason }; + if session.type == "s2scout" then + fire_event("s2scout-destroyed", event_data); + if hosts[session.from_host] then + hosts[session.from_host].events.fire_event("s2scout-destroyed", event_data); + end + elseif session.type == "s2scin" then + fire_event("s2scin-destroyed", event_data); + if hosts[session.to_host] then + hosts[session.to_host].events.fire_event("s2scin-destroyed", event_data); + end + end + + retire_session(session, reason); -- Clean session until it is GC'd + return true; +end + +return _M; diff --git a/debian/ngcp-prosody-modules.install b/debian/ngcp-prosody-modules.install index 0c09054..1413ebd 100644 --- a/debian/ngcp-prosody-modules.install +++ b/debian/ngcp-prosody-modules.install @@ -1,2 +1,3 @@ plugins/* /usr/lib/prosody/modules/ util/* /usr/lib/prosody/util/ +core/* /usr/lib/prosody/core/ diff --git a/plugins/mod_s2s_cluster/mod_s2s_cluster.lua b/plugins/mod_s2s_cluster/mod_s2s_cluster.lua new file mode 100644 index 0000000..a846f5e --- /dev/null +++ b/plugins/mod_s2s_cluster/mod_s2s_cluster.lua @@ -0,0 +1,690 @@ +-- Prosody IM +-- Copyright (C) 2008-2010 Matthew Wild +-- Copyright (C) 2008-2010 Waqas Hussain +-- Copyright (C) 2014 Sipwise GmbH +-- +-- This project is MIT/X11 licensed. Please see the +-- COPYING file in the source package for more information. +-- + +module:set_global(); +module:depends("sipwise_cluster"); + +local ut = require "util.table"; +local prosody = prosody; +local hosts = prosody.hosts; + +local tostring, type = tostring, type; +local t_insert = table.insert; +local xpcall, traceback = xpcall, debug.traceback; +local NULL = {}; +local jid_split = require "util.jid".split; + +local add_task = require "util.timer".add_task; +local st = require "util.stanza"; +local initialize_filters = require "util.filters".initialize; +local nameprep = require "util.encodings".stringprep.nameprep; +local new_xmpp_stream = require "util.xmppstream".new; +local s2sc_new_incoming = require "core.s2scmanager".new_incoming; +local s2sc_new_outgoing = require "core.s2scmanager".new_outgoing; +local s2sc_destroy_session = require "core.s2scmanager".destroy_session; +local uuid_gen = require "util.uuid".generate; +local cert_verify_identity = require "util.x509".verify_identity; +local fire_global_event = prosody.events.fire_event; + +local s2scout = module:require("s2scout"); + +local connect_timeout = module:get_option_number("s2sc_timeout", 90); +local stream_close_timeout = module:get_option_number("s2sc_close_timeout", 5); +local opt_keepalives = module:get_option_boolean("s2sc_tcp_keepalives", module:get_option_boolean("tcp_keepalives", true)); +local secure_auth = module:get_option_boolean("s2sc_secure_auth", false); -- One day... +local secure_domains, insecure_domains = + module:get_option_set("s2sc_secure_domains", {})._items, module:get_option_set("s2sc_insecure_domains", {})._items; +local require_encryption = module:get_option_boolean("s2sc_require_encryption", false); + +local sessions = module:shared("sessions"); +local cluster = module:shared("/*/sipwise_cluster/cluster"); +local log = module._log; + +--- Handle stanzas to remote domains + +local bouncy_stanzas = { message = true, presence = true, iq = true }; +local function bounce_sendq(session, reason) + local sendq = session.sendq; + if not sendq then return; end + session.log("info", "sending error replies for "..#sendq.." queued stanzas because of failed outgoing connection to "..tostring(session.to_host)); + local dummy = { + type = "s2scin"; + send = function(s) + (session.log or log)("error", "Replying to to an s2sc error reply, please report this! Traceback: %s", traceback()); + end; + dummy = true; + }; + for i, data in ipairs(sendq) do + local reply = data[2]; + if reply and not(reply.attr.xmlns) and bouncy_stanzas[reply.name] then + reply.attr.type = "error"; + reply:tag("error", {type = "cancel"}) + :tag("remote-server-not-found", {xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas"}):up(); + if reason then + reply:tag("text", {xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas"}) + :text("Server-to-server connection failed: "..reason):up(); + end + cluster.core_process_stanza(dummy, reply); + end + sendq[i] = nil; + end + session.sendq = nil; +end + +-- Handles stanzas to existing s2sc sessions +function route_to_existing_session(event) + local from_host, to_host, stanza = event.from_host, event.to_host, event.stanza; + if not hosts[from_host] then + local st_to_node, st_to_host, st_to_resource = jid_split(stanza.attr.to); + if hosts[st_to_host] and hosts[st_to_host].s2scout then + log("debug", "[session_outside] from_host:%s to_host:%s", + tostring(from_host), tostring(to_host)); + from_host = st_to_host; + log("debug", "event.from_host:%s from_host:%s", + tostring(event.from_host), tostring(from_host)); + else + log("warn", "Attempt to send stanza from %s - a host we don't serve", from_host); + return false; + end + end + --log("debug", "from_host:%s to_host:%s", tostring(from_host), tostring(to_host)); + if hosts[from_host].s2scout then + local host = hosts[from_host].s2scout[to_host]; + if host then + (host.log or log)("debug", "host.type:"..host.type); + -- We have a connection to this host already + if host.type == "s2scout_unauthed" and (stanza.name ~= "db:verify" or not host.dialback_key) then + (host.log or log)("debug", "trying to send over unauthed s2scout to "..to_host); + + -- Queue stanza until we are able to send it + if host.sendq then t_insert(host.sendq, {tostring(stanza), stanza.attr.type ~= "error" and stanza.attr.type ~= "result" and st.reply(stanza)}); + else host.sendq = { {tostring(stanza), stanza.attr.type ~= "error" and stanza.attr.type ~= "result" and st.reply(stanza)} }; end + host.log("debug", "stanza [%s] queued ", stanza.name); + return true; + elseif host.type == "local" or host.type == "component" then + log("error", "Trying to send a stanza to ourselves??") + log("error", "Traceback: %s", traceback()); + log("error", "Stanza: %s", tostring(stanza)); + return false; + elseif host.type == "s2sc_destroyed" then + log("warn", "s2sc destroyed"); + return; + else + (host.log or log)("debug", "going to send stanza to "..to_host.." from "..from_host); + -- FIXME + if host.from_host ~= from_host then + log("error", "WARNING! This might, possibly, be a bug, but it might not..."); + log("error", "We are going to send from %s instead of %s", tostring(host.from_host), tostring(from_host)); + end + if host.sends2sc(stanza) then + host.log("debug", "stanza sent over %s", host.type); + return true; + end + end + end + end +end + +-- Create a new outgoing session for a stanza +function route_to_new_session(event) + local from_host, to_host, stanza = event.from_host, event.to_host, event.stanza; + log("debug", "opening a new outgoing connection for this stanza"); + --log("debug", "from_host:%s to_host:%s stanza:%s", + -- tostring(from_host), tostring(to_host), tostring(stanza)); + local host_session = s2sc_new_outgoing(from_host, to_host); + + -- Store in buffer + host_session.bounce_sendq = bounce_sendq; + host_session.sendq = { {tostring(stanza), stanza.attr.type ~= "error" and stanza.attr.type ~= "result" and st.reply(stanza)} }; + log("debug", "stanza [%s] queued until connection complete", tostring(stanza.name)); + s2scout.initiate_connection(host_session); + if (not host_session.connecting) and (not host_session.conn) then + log("warn", "Connection to %s failed already, destroying session...", to_host); + s2sc_destroy_session(host_session, "Connection failed"); + return false; + end + return true; +end + +function module.add_host(module) + module:hook("route/remote_cluster", route_to_existing_session, -1); + module:hook("route/remote_cluster", route_to_new_session, -10); + module:hook("s2sc-authenticated", make_authenticated, -1); +end + +-- Stream is authorised, and ready for normal stanzas +function mark_connected(session) + local sendq, send = session.sendq, session.sends2sc; + + local from, to = session.from_host, session.to_host; + + session.log("info", "%s s2sc connection %s->%s complete", session.direction, from, to); + + local event_data = { session = session }; + if session.type == "s2scout" then + fire_global_event("s2scout-established", event_data); + hosts[from].events.fire_event("s2scout-established", event_data); + else + local host_session = hosts[to]; + session.send = function(stanza) + return host_session.events.fire_event("route/remote_cluster", { from_host = to, to_host = from, stanza = stanza }); + end; + + fire_global_event("s2scin-established", event_data); + hosts[to].events.fire_event("s2scin-established", event_data); + end + + if session.direction == "outgoing" then + if sendq then + session.log("debug", "sending %d queued stanzas across new outgoing connection to %s", #sendq, session.to_host); + for i, data in ipairs(sendq) do + send(data[1]); + sendq[i] = nil; + end + session.sendq = nil; + end + + session.ip_hosts = nil; + session.srv_hosts = nil; + end +end + +function make_authenticated(event) + local session, host = event.session, event.host; + if not session.secure then + if require_encryption or (secure_auth and not(insecure_domains[host])) or secure_domains[host] then + session:close({ + condition = "policy-violation", + text = "Encrypted server-to-server communication is required but was not " + ..((session.direction == "outgoing" and "offered") or "used") + }); + end + end + if not hosts[host] then + session:close({ condition = "undefined-condition", text = "Attempt to authenticate as a host we don't serve" }); + end + if session.type == "s2scout_unauthed" then + session.type = "s2scout"; + elseif session.type == "s2scin_unauthed" then + session.type = "s2scin"; + if host then + if not session.hosts[host] then session.hosts[host] = {}; end + session.hosts[host].authed = true; + end + elseif session.type == "s2scin" and host then + if not session.hosts[host] then session.hosts[host] = {}; end + session.hosts[host].authed = true; + else + return false; + end + session.log("debug", "connection %s->%s is now authenticated for %s", session.from_host, session.to_host, host); + + mark_connected(session); + + return true; +end + +--- Helper to check that a session peer's certificate is valid +local function check_cert_status(session) + local host = session.direction == "outgoing" and session.to_host or session.from_host + local conn = session.conn:socket() + local cert + if conn.getpeercertificate then + cert = conn:getpeercertificate() + end + + if cert then + local chain_valid, errors; + if conn.getpeerverification then + chain_valid, errors = conn:getpeerverification(); + elseif conn.getpeerchainvalid then -- COMPAT mw/luasec-hg + chain_valid, errors = conn:getpeerchainvalid(); + errors = (not chain_valid) and { { errors } } or nil; + else + chain_valid, errors = false, { { "Chain verification not supported by this version of LuaSec" } }; + end + -- Is there any interest in printing out all/the number of errors here? + if not chain_valid then + (session.log or log)("debug", "certificate chain validation result: invalid"); + for depth, t in pairs(errors or NULL) do + (session.log or log)("debug", "certificate error(s) at depth %d: %s", depth-1, table.concat(t, ", ")) + end + session.cert_chain_status = "invalid"; + else + (session.log or log)("debug", "certificate chain validation result: valid"); + session.cert_chain_status = "valid"; + + -- We'll go ahead and verify the asserted identity if the + -- connecting server specified one. + if host then + if cert_verify_identity(host, "xmpp-server", cert) then + session.cert_identity_status = "valid" + else + session.cert_identity_status = "invalid" + end + (session.log or log)("debug", "certificate identity validation result: %s", session.cert_identity_status); + end + end + end + (session.log or log)("debug", "fire event:s2sc-check-certificate"); + return module:fire_event("s2sc-check-certificate", { host = host, session = session, cert = cert }); +end + +--- XMPP stream event handlers + +local stream_callbacks = { default_ns = "jabber:serverc", handlestanza = cluster.core_process_stanza }; + +local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; + +function stream_callbacks.streamopened(session, attr) + local send = session.sends2sc; + + session.version = tonumber(attr.version) or 0; + + -- TODO: Rename session.secure to session.encrypted + if session.secure == false then + session.secure = true; + + -- Check if TLS compression is used + local sock = session.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 + + if session.direction == "incoming" then + -- Send a reply stream header + + -- Validate to/from + local to, from = nameprep(attr.to), nameprep(attr.from); + if not to and attr.to then -- COMPAT: Some servers do not reliably set 'to' (especially on stream restarts) + session:close({ condition = "improper-addressing", text = "Invalid 'to' address" }); + return; + end + if not from and attr.from then -- COMPAT: Some servers do not reliably set 'from' (especially on stream restarts) + session:close({ condition = "improper-addressing", text = "Invalid 'from' address" }); + return; + end + + -- Set session.[from/to]_host if they have not been set already and if + -- this session isn't already authenticated + if session.type == "s2scin_unauthed" and from and not session.from_host then + session.from_host = from; + elseif from ~= session.from_host then + session:close({ condition = "improper-addressing", text = "New stream 'from' attribute does not match original" }); + return; + end + if session.type == "s2scin_unauthed" and to and not session.to_host then + session.to_host = to; + elseif to ~= session.to_host then + session:close({ condition = "improper-addressing", text = "New stream 'to' attribute does not match original" }); + return; + end + + -- For convenience we'll put the sanitised values into these variables + to, from = session.to_host, session.from_host; + + session.streamid = uuid_gen(); + (session.log or log)("debug", "Incoming s2sc received %s", st.stanza("stream:stream", attr):top_tag()); + if to then + if not hosts[to] then + -- Attempting to connect to a host we don't serve + session:close({ + condition = "host-unknown"; + text = "This host does not serve "..to + }); + return; + elseif not hosts[to].modules.s2s_cluster then + -- Attempting to connect to a host that disallows s2sc + session:close({ + condition = "policy-violation"; + text = "Server-to-server communication is disabled for this host"; + }); + return; + end + end + + if not hosts[from] then + session:close({ condition = "undefined-condition", text = "Attempt to connect from a host we don't serve" }); + return; + end + + if session.secure and not session.cert_chain_status then + if check_cert_status(session) == false then + return; + end + end + + session:open_stream(session.to_host, session.from_host) + if session.version >= 1.0 then + local features = st.stanza("stream:features"); + + if to then + log("debug", "fire event:s2sc-stream-features"); + hosts[to].events.fire_event("s2sc-stream-features", { origin = session, features = features }); + else + (session.log or log)("warn", "No 'to' on stream header from %s means we can't offer any features", from or session.ip or "unknown host"); + end + + log("debug", "Sending stream features: %s", tostring(features)); + send(features); + end + elseif session.direction == "outgoing" then + -- If we are just using the connection for verifying dialback keys, we won't try and auth it + if not attr.id then error("stream response did not give us a streamid!!!"); end + session.streamid = attr.id; + + if session.secure and not session.cert_chain_status then + if check_cert_status(session) == false then + return; + end + end + + -- Send unauthed buffer + -- (stanzas which are fine to send before dialback) + -- Note that this is *not* the stanza queue (which + -- we can only send if auth succeeds) :) + local send_buffer = session.send_buffer; + if send_buffer and #send_buffer > 0 then + log("debug", "Sending s2sc send_buffer now..."); + for i, data in ipairs(send_buffer) do + session.sends2sc(tostring(data)); + send_buffer[i] = nil; + end + end + session.send_buffer = nil; + + -- If server is pre-1.0, don't wait for features, just do dialback + if session.version < 1.0 then + if not session.dialback_verifying then + log("debug", "fire event:s2scout-authenticate-legacy"); + hosts[session.from_host].events.fire_event("s2scout-authenticate-legacy", { origin = session }); + else + mark_connected(session); + end + end + end + session.notopen = nil; +end + +function stream_callbacks.streamclosed(session) + (session.log or log)("debug", "Received "); + session:close(false); +end + +function stream_callbacks.error(session, error, data) + if error == "no-stream" then + session:close("invalid-namespace"); + elseif error == "parse-error" then + session.log("debug", "Server-to-server-cluster XML parse error: %s", tostring(error)); + session:close("not-well-formed"); + elseif error == "stream-error" then + local condition, text = "undefined-condition"; + for child in data:children() do + if child.attr.xmlns == xmlns_xmpp_streams then + if child.name ~= "text" then + condition = child.name; + else + text = child:get_text(); + end + if condition ~= "undefined-condition" and text then + break; + end + end + end + text = condition .. (text and (" ("..text..")") or ""); + session.log("info", "Session s2sc closed by remote with error: %s", text); + session:close(nil, text); + end +end + +local function handleerr(err) log("error", "Traceback[s2sc]: %s", traceback(tostring(err), 2)); end +function stream_callbacks.handlestanza(session, stanza) + if stanza.attr.xmlns == "jabber:client" then --COMPAT: Prosody pre-0.6.2 may send jabber:client + stanza.attr.xmlns = nil; + end + stanza = session.filter("stanzas/in", stanza); + if stanza then + return xpcall(function () return cluster.core_process_stanza(session, stanza) end, handleerr); + end +end + +local listener = {}; + +--- Session methods +local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'}; +local function session_close(session, reason, remote_reason) + local log = session.log or log; + if session.conn then + if session.notopen then + if session.direction == "incoming" then + session:open_stream(session.to_host, session.from_host); + else + session:open_stream(session.from_host, session.to_host); + end + end + if reason then -- nil == no err, initiated by us, false == initiated by remote + if type(reason) == "string" then -- assume stream error + log("debug", "Disconnecting %s[%s], is: %s", session.host or session.ip or "(unknown host)", session.type, reason); + session.sends2sc(st.stanza("stream:error"):tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' })); + elseif type(reason) == "table" then + if reason.condition then + local stanza = st.stanza("stream:error"):tag(reason.condition, stream_xmlns_attr):up(); + if reason.text then + stanza:tag("text", stream_xmlns_attr):text(reason.text):up(); + end + if reason.extra then + stanza:add_child(reason.extra); + end + log("debug", "Disconnecting %s[%s], is: %s", session.host or session.ip or "(unknown host)", session.type, tostring(stanza)); + session.sends2sc(stanza); + elseif reason.name then -- a stanza + log("debug", "Disconnecting %s->%s[%s], is: %s", session.from_host or "(unknown host)", session.to_host or "(unknown host)", session.type, tostring(reason)); + session.sends2sc(reason); + end + end + end + + session.sends2sc(""); + function session.sends2sc() return false; end + + local reason = remote_reason or (reason and (reason.text or reason.condition)) or reason; + session.log("info", "%s s2sc stream %s->%s closed: %s", session.direction, session.from_host or "(unknown host)", session.to_host or "(unknown host)", reason or "stream closed"); + + -- Authenticated incoming stream may still be sending us stanzas, so wait for from remote + local conn = session.conn; + if reason == nil and not session.notopen and session.type == "s2scin" then + add_task(stream_close_timeout, function () + if not session.destroyed then + session.log("warn", "Failed to receive a stream close response, closing connection anyway..."); + s2sc_destroy_session(session, reason); + conn:close(); + end + end); + else + s2sc_destroy_session(session, reason); + conn:close(); -- Close immediately, as this is an outgoing connection or is not authed + end + end +end + +function session_open_stream(session, from, to) + local attr = { + ["xmlns:stream"] = 'http://etherx.jabber.org/streams', + xmlns = 'jabber:serverc', + version = session.version and (session.version > 0 and "1.0" or nil), + ["xml:lang"] = 'en', + id = session.streamid, + from = from, to = to, + } + if not from or (hosts[from] and hosts[from].modules.s2sc_dialback) then + attr["xmlns:db"] = 'jabber:serverc:dialback'; + end + + session.sends2sc(""); + session.sends2sc(st.stanza("stream:stream", attr):top_tag()); + return true; +end + +-- Session initialization logic shared by incoming and outgoing +local function initialize_session(session) + local stream = new_xmpp_stream(session, stream_callbacks); + session.stream = stream; + + session.notopen = true; + + function session.reset_stream() + session.notopen = true; + session.stream:reset(); + end + + session.open_stream = session_open_stream; + + local filter = session.filter; + function session.data(data) + data = filter("bytes/in", data); + if data then + local ok, err = stream:feed(data); + if ok then return; end + (session.log or log)("warn", "Received invalid XML: %s", data); + (session.log or log)("warn", "Problem was: %s", err); + session:close("not-well-formed"); + end + end + + session.close = session_close; + + local handlestanza = stream_callbacks.handlestanza; + function session.dispatch_stanza(session, stanza) + return handlestanza(session, stanza); + end + + add_task(connect_timeout, function () + if session.type == "s2scin" or session.type == "s2scout" then + return; -- Ok, we're connected + elseif session.type == "s2sc_destroyed" then + return; -- Session already destroyed + end + -- Not connected, need to close session and clean up + (session.log or log)("debug", "Destroying incomplete session %s->%s due to inactivity", + session.from_host or "(unknown)", session.to_host or "(unknown)"); + session:close("connection-timeout"); + end); +end + +function listener.onconnect(conn) + conn:setoption("keepalive", opt_keepalives); + local session = sessions[conn]; + if not session then -- New incoming connection + session = s2sc_new_incoming(conn); + sessions[conn] = session; + session.log("debug", "Incoming s2sc connection"); + + local filter = initialize_filters(session); + local w = conn.write; + session.sends2sc = function (t) + log("debug", "sending: %s", t.top_tag and t:top_tag() or t:match("^([^>]*>?)")); + if t.name then + t = filter("stanzas/out", t); + end + if t then + t = filter("bytes/out", tostring(t)); + if t then + return w(conn, t); + end + end + end + + initialize_session(session); + else -- Outgoing session connected + session:open_stream(session.from_host, session.to_host); + end +end + +function listener.onincoming(conn, data) + local session = sessions[conn]; + if session then + session.data(data); + end +end + +function listener.onstatus(conn, status) + if status == "ssl-handshake-complete" then + local session = sessions[conn]; + if session and session.direction == "outgoing" then + session.log("debug", "Sending stream header..."); + session:open_stream(session.from_host, session.to_host); + end + end +end + +function listener.ondisconnect(conn, err) + local session = sessions[conn]; + if session then + sessions[conn] = nil; + if err and session.direction == "outgoing" and session.notopen then + (session.log or log)("debug", "s2sc connection attempt failed: %s", err); + end + (session.log or log)("debug", "s2sc disconnected: %s->%s (%s)", tostring(session.from_host), tostring(session.to_host), tostring(err or "connection closed")); + s2sc_destroy_session(session, err); + end +end + +function listener.register_outgoing(conn, session) + session.direction = "outgoing"; + sessions[conn] = session; + initialize_session(session); +end + +function check_auth_policy(event) + local host, session = event.host, event.session; + local must_secure = secure_auth; + + if not must_secure and secure_domains[host] then + must_secure = true; + elseif must_secure and insecure_domains[host] then + must_secure = false; + end + + if must_secure and (session.cert_chain_status ~= "valid" or session.cert_identity_status ~= "valid") then + module:log("warn", "Forbidding insecure connection to/from %s", host or session.ip or "(unknown host)"); + if session.direction == "incoming" then + session:close({ condition = "not-authorized", text = "Your server's certificate is invalid, expired, or not trusted by "..session.to_host }); + else -- Close outgoing connections without warning + session:close(false); + end + return false; + end +end + +module:hook("s2sc-check-certificate", check_auth_policy, -1); + +s2scout.set_listener(listener); + +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,500); + + + +module:provides("net", { + name = "s2sc"; + listener = listener; + default_port = 15269; + encryption = "starttls"; + multiplex = { + pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:serverc%1.*>"; + }; +}); + diff --git a/plugins/mod_s2s_cluster/s2scout.lib.lua b/plugins/mod_s2s_cluster/s2scout.lib.lua new file mode 100644 index 0000000..cd07c05 --- /dev/null +++ b/plugins/mod_s2s_cluster/s2scout.lib.lua @@ -0,0 +1,158 @@ +-- Prosody IM +-- Copyright (C) 2008-2010 Matthew Wild +-- Copyright (C) 2008-2010 Waqas Hussain +-- Copyright (C) 2014 Sipwise GmbH +-- +-- This project is MIT/X11 licensed. Please see the +-- COPYING file in the source package for more information. +-- + +--- Module containing all the logic for connecting to a remote server + +local portmanager = require "core.portmanager"; +local wrapclient = require "net.server".wrapclient; +local initialize_filters = require "util.filters".initialize; +local idna_to_ascii = require "util.encodings".idna.to_ascii; +local new_ip = require "util.ip".new_ip; +local rfc6724_dest = require "util.rfc6724".destination; +local socket = require "socket"; +local t_insert, t_sort, ipairs = table.insert, table.sort, ipairs; +local local_addresses = require "util.net".local_addresses; + +local s2sc_destroy_session = require "core.s2scmanager".destroy_session; + +local log = module._log; + +local sources = {}; +local has_ipv4, has_ipv6; + +local s2scout = {}; + +local s2sc_listener; + + +function s2scout.set_listener(listener) + s2sc_listener = listener; +end + +function s2scout.initiate_connection(host_session) + initialize_filters(host_session); + host_session.version = 1; + + -- Kick the connection attempting machine into life + local connect_host = { addr=host_session.to_host, proto="IPv4" }; + host_session.to_host = host_session.from_host; + if not s2scout.make_connect(host_session, connect_host, 15269) then + -- Intentionally not returning here, the + -- session is needed, connected or not + s2sc_destroy_session(host_session); + end + + if not host_session.sends2sc then + host_session.log("debug", "adding sends2sc") + -- A sends2sc which buffers data (until the stream is opened) + -- note that data in this buffer will be sent before the stream is authed + -- and will not be ack'd in any way, successful or otherwise + local buffer; + function host_session.sends2sc(data) + if not buffer then + buffer = {}; + host_session.send_buffer = buffer; + end + log("debug", "[sends2sc] Buffering data on unconnected s2scout to %s", tostring(host_session.to_host)); + buffer[#buffer+1] = data; + log("debug", "[sends2sc] Buffered item %d: %s", #buffer, tostring(data)); + end + end +end + +function s2scout.make_connect(host_session, connect_host, connect_port) + (host_session.log or log)("info", "Beginning new connection attempt to %s ([%s]:%d)", host_session.to_host, connect_host.addr, connect_port); + + -- Reset secure flag in case this is another + -- connection attempt after a failed STARTTLS + host_session.secure = nil; + + local conn, handler; + local proto = connect_host.proto; + if proto == "IPv4" then + conn, handler = socket.tcp(); + elseif proto == "IPv6" and socket.tcp6 then + conn, handler = socket.tcp6(); + else + handler = "Unsupported protocol: "..tostring(proto); + end + + if not conn then + log("warn", "Failed to create outgoing connection, system error: %s", handler); + return false, handler; + end + + conn:settimeout(0); + local success, err = conn:connect(connect_host.addr, connect_port); + if not success and err ~= "timeout" then + log("warn", "s2sc connect() to %s (%s:%d) failed: %s", host_session.to_host, connect_host.addr, connect_port, err); + return false, err; + end + + conn = wrapclient(conn, connect_host.addr, connect_port, s2sc_listener, "*a"); + host_session.conn = conn; + + local filter = initialize_filters(host_session); + local w, log = conn.write, host_session.log; + host_session.sends2sc = function (t) + log("debug", "sends2sc: sending[%s (%s:%d)]: %s", + host_session.to_host, connect_host.addr, connect_port, + (t.top_tag and t:top_tag()) or t:match("^[^>]*>?")); + if t.name then + t = filter("stanzas/out", t); + end + if t then + t = filter("bytes/out", tostring(t)); + if t then + return w(conn, tostring(t)); + end + end + end + -- Register this outgoing connection so that xmppserver_listener knows about it + -- otherwise it will assume it is a new incoming connection + s2sc_listener.register_outgoing(conn, host_session); + + log("debug", "Connection attempt in progress..."); + return true; +end + +module:hook_global("service-added", function (event) + if event.name ~= "s2sc" then return end + + local s2sc_sources = portmanager.get_active_services():get("s2sc"); + if not s2sc_sources then + module:log("warn", "s2sc not listening on any ports, outgoing connections may fail"); + return; + end + for source, _ in pairs(s2sc_sources) do + if source == "*" or source == "0.0.0.0" then + for _, addr in ipairs(local_addresses("ipv4", true)) do + sources[#sources + 1] = new_ip(addr, "IPv4"); + end + elseif source == "::" then + for _, addr in ipairs(local_addresses("ipv6", true)) do + sources[#sources + 1] = new_ip(addr, "IPv6"); + end + else + sources[#sources + 1] = new_ip(source, (source:find(":") and "IPv6") or "IPv4"); + end + end + for i = 1,#sources do + if sources[i].proto == "IPv6" then + has_ipv6 = true; + elseif sources[i].proto == "IPv4" then + has_ipv4 = true; + end + end + if not (has_ipv4 or has_ipv6) then + module:log("warn", "No local IPv4 or IPv6 addresses detected, outgoing connections may fail"); + end +end); + +return s2scout; diff --git a/plugins/mod_s2sc_dialback.lua b/plugins/mod_s2sc_dialback.lua new file mode 100644 index 0000000..9de195e --- /dev/null +++ b/plugins/mod_s2sc_dialback.lua @@ -0,0 +1,185 @@ +-- Prosody IM +-- Copyright (C) 2008-2010 Matthew Wild +-- Copyright (C) 2008-2010 Waqas Hussain +-- Copyright (C) 2014 Sipwise GmbH +-- +-- This project is MIT/X11 licensed. Please see the +-- COPYING file in the source package for more information. +-- + +local hosts = _G.hosts; + +local log = module._log; + +local st = require "util.stanza"; +local ut = require "util.table"; +local sha256_hash = require "util.hashes".sha256; +local nameprep = require "util.encodings".stringprep.nameprep; + +local xmlns_stream = "http://etherx.jabber.org/streams"; + +local dialback_requests = setmetatable({}, { __mode = 'v' }); + +function generate_dialback(id, to, from) + return sha256_hash(id..to..from..hosts[from].dialback_secret, true); +end + +function initiate_dialback(session) + -- generate dialback key + session.dialback_key = generate_dialback(session.streamid, session.to_host, session.from_host); + session.sends2sc(st.stanza("db:result", { from = session.from_host, to = session.to_host }):text(session.dialback_key)); + session.log("info", "sent dialback key on outgoing s2sc stream"); +end + +function verify_dialback(id, to, from, key) + return key == generate_dialback(id, to, from); +end + +module:hook("stanza/jabber:serverc:dialback:verify", function(event) + local origin, stanza = event.origin, event.stanza; + + if origin.type == "s2scin_unauthed" or origin.type == "s2scin" then + -- We are being asked to verify the key, to ensure it was generated by us + origin.log("debug", "verifying that dialback key is ours..."); + local attr = stanza.attr; + if attr.type then + module:log("warn", "Ignoring incoming session from %s claiming a dialback key for %s is %s", + origin.from_host or "(unknown)", attr.from or "(unknown)", attr.type); + return true; + end + -- COMPAT: Grr, ejabberd breaks this one too?? it is black and white in XEP-220 example 34 + --if attr.from ~= origin.to_host then error("invalid-from"); end + local type; + if verify_dialback(attr.id, attr.from, attr.to, stanza[1]) then + type = "valid" + else + type = "invalid" + origin.log("warn", "Asked to verify a dialback key that was incorrect. An imposter is claiming to be %s?", attr.to); + end + origin.log("debug", "verified dialback key... it is %s", type); + origin.sends2sc(st.stanza("db:verify", { from = attr.to, to = attr.from, id = attr.id, type = type }):text(stanza[1])); + return true; + end +end); + +module:hook("stanza/jabber:serverc:dialback:result", function(event) + local origin, stanza = event.origin, event.stanza; + + if origin.type == "s2scin_unauthed" or origin.type == "s2scin" then + -- he wants to be identified through dialback + -- We need to check the key with the Authoritative server + local attr = stanza.attr; + local to, from = nameprep(attr.to), nameprep(attr.from); + + if not hosts[to] then + -- Not a host that we serve + origin.log("info", "%s tried to connect to %s, which we don't serve", from, to); + origin:close("host-unknown"); + return true; + elseif not from then + origin:close("improper-addressing"); + end + + origin.hosts[from] = { dialback_key = stanza[1] }; + + dialback_requests[from.."/"..origin.streamid] = origin; + + -- COMPAT: ejabberd, gmail and perhaps others do not always set 'to' and 'from' + -- on streams. We fill in the session's to/from here instead. + if not origin.from_host then + origin.from_host = from; + end + if not origin.to_host then + origin.to_host = to; + end + + origin.log("debug", "asking %s if key %s belongs to them", from, stanza[1]); + module:fire_event("route/remote_cluster", { + from_host = to, to_host = origin.conn.ip(); + stanza = st.stanza("db:verify", { from = to, to = from, id = origin.streamid }):text(stanza[1]); + }); + return true; + end +end); + +module:hook("stanza/jabber:serverc:dialback:verify", function(event) + local origin, stanza = event.origin, event.stanza; + + if origin.type == "s2scout_unauthed" or origin.type == "s2scout" then + local attr = stanza.attr; + local dialback_verifying = dialback_requests[attr.from.."/"..(attr.id or "")]; + if dialback_verifying and attr.from == origin.to_host then + local valid; + if attr.type == "valid" then + module:fire_event("s2sc-authenticated", { session = dialback_verifying, host = attr.from }); + valid = "valid"; + else + -- Warn the original connection that is was not verified successfully + log("warn", "authoritative server for %s denied the key", attr.from or "(unknown)"); + valid = "invalid"; + end + if dialback_verifying.destroyed then + log("warn", "Incoming s2sc session %s was closed in the meantime, so we can't notify it of the db result", tostring(dialback_verifying):match("%w+$")); + else + dialback_verifying.sends2sc( + st.stanza("db:result", { from = attr.to, to = attr.from, id = attr.id, type = valid }) + :text(dialback_verifying.hosts[attr.from].dialback_key)); + end + dialback_requests[attr.from.."/"..(attr.id or "")] = nil; + end + return true; + end +end); + +module:hook("stanza/jabber:serverc:dialback:result", function(event) + local origin, stanza = event.origin, event.stanza; + + if origin.type == "s2scout_unauthed" or origin.type == "s2scout" then + -- Remote server is telling us whether we passed dialback + + local attr = stanza.attr; + if not hosts[attr.to] then + origin:close("host-unknown"); + return true; + elseif hosts[attr.to].s2scout[origin.conn.ip()] ~= origin then + -- This isn't right + origin:close("invalid-id"); + return true; + end + if stanza.attr.type == "valid" then + module:fire_event("s2sc-authenticated", { session = origin, host = attr.from }); + else + origin:close("not-authorized", "s2sc dialback authentication failed"); + end + return true; + end +end); + +module:hook_stanza("urn:ietf:params:xml:ns:xmpp-sasl", "failure", function (origin, stanza) + if origin.external_auth == "failed" then + module:log("debug", "SASL EXTERNAL failed, falling back to s2sc dialback"); + initiate_dialback(origin); + return true; + end +end, 100); + +module:hook_stanza(xmlns_stream, "features", function (origin, stanza) + if origin.sends2sc then + if not origin.external_auth or origin.external_auth == "failed" then + module:log("debug", "Initiating s2sc dialback..."); + initiate_dialback(origin); + return true; + end + end +end, 200); + +module:hook("s2scout-authenticate-legacy", function (event) + module:log("debug", "Initiating s2sc dialback..."); + initiate_dialback(event.origin); + return true; +end, 100); + +-- Offer dialback to incoming hosts +module:hook("s2sc-stream-features", function (data) + data.features:tag("dialback", { xmlns='urn:xmpp:features:dialback' }):up(); +end); diff --git a/plugins/mod_sipwise_cluster.lua b/plugins/mod_sipwise_cluster.lua new file mode 100644 index 0000000..d94fa83 --- /dev/null +++ b/plugins/mod_sipwise_cluster.lua @@ -0,0 +1,304 @@ +-- Prosody IM +-- Copyright (C) 2014 Sipwise GmbH +-- +-- This project is MIT/X11 licensed. Please see the +-- COPYING file in the source package for more information. +-- + +module:set_global(); +module:depends("sipwise_redis_sessions"); + +local st = require "util.stanza"; +local ut = require "util.table"; +local jid_split = require "util.jid".split; +local jid_prepped_split = require "util.jid".prepped_split; + +local redis_sessions = module:shared("/*/sipwise_redis_sessions/redis_sessions"); +local cluster = module:shared("cluster"); +local core = { + process_stanza = prosody.core_process_stanza, + route_stanza = prosody.core_route_stanza, + post_stanza = prosody.core_post_stanza, + hosts = prosody.hosts +}; + +local cluster_config = { + me = "localhost", + hosts = {}, + dialback_secret = nil +}; + +local function handle_unhandled_stanza(host, origin, stanza) + local name, xmlns, origin_type = stanza.name, stanza.attr.xmlns or "jabber:client", origin.type; + if name == "iq" and xmlns == "jabber:client" then + if stanza.attr.type == "get" or stanza.attr.type == "set" then + xmlns = stanza.tags[1].attr.xmlns or "jabber:client"; + log("debug", "Stanza of type %s from %s has xmlns: %s", name, origin_type, xmlns); + else + log("debug", "Discarding %s from %s of type: %s", name, origin_type, stanza.attr.type); + return true; + end + end + if stanza.attr.xmlns == nil and origin.send then + log("debug", "Unhandled %s stanza: %s; xmlns=%s", origin.type, stanza.name, xmlns); -- we didn't handle it + if stanza.attr.type ~= "error" and stanza.attr.type ~= "result" then + origin.send(st.error_reply(stanza, "cancel", "service-unavailable")); + end + elseif not((name == "features" or name == "error") and xmlns == "http://etherx.jabber.org/streams") then -- FIXME remove check once we handle S2S features + log("warn", "Unhandled %s stream element or stanza: %s; xmlns=%s: %s", origin.type, stanza.name, xmlns, tostring(stanza)); -- we didn't handle it + origin:close("unsupported-stanza-type"); + end +end + +local iq_types = { set=true, get=true, result=true, error=true }; +function cluster.core_process_stanza(origin, stanza) + module:log("debug", "--- sipwise_cluster.core_process_staza --"); + --module:log("debug", "--- origin:%s stanza:%s", + -- ut.table.tostring(origin), tostring(stanza)); + + if not ut.string.starts(origin.type, "s2sc") then return core.process_stanza(origin, stanza) end + (origin.log or log)("debug", "--- Received[%s]: %s", origin.type, stanza:top_tag()) + -- TODO verify validity of stanza (as well as JID validity) + if stanza.attr.type == "error" and #stanza.tags == 0 then return; end -- TODO invalid stanza, log + if stanza.name == "iq" then + if not stanza.attr.id then stanza.attr.id = ""; end -- COMPAT Jabiru doesn't send the id attribute on roster requests + if not iq_types[stanza.attr.type] or ((stanza.attr.type == "set" or stanza.attr.type == "get") and (#stanza.tags ~= 1)) then + origin.sends2sc(st.error_reply(stanza, "modify", "bad-request", "Invalid IQ type or incorrect number of children")); + return; + end + end + + if origin.type == "c2scin" and not stanza.attr.xmlns then + if not origin.full_jid + and not(stanza.name == "iq" and stanza.attr.type == "set" and stanza.tags[1] and stanza.tags[1].name == "bind" + and stanza.tags[1].attr.xmlns == "urn:ietf:params:xml:ns:xmpp-bind") then + -- authenticated client isn't bound and current stanza is not a bind request + if stanza.attr.type ~= "result" and stanza.attr.type ~= "error" then + origin.sends2sc(st.error_reply(stanza, "auth", "not-authorized")); -- FIXME maybe allow stanzas to account or server + end + return; + end + + -- TODO also, stanzas should be returned to their original state before the function ends + stanza.attr.from = origin.full_jid; + end + local to, xmlns = stanza.attr.to, stanza.attr.xmlns; + local from = stanza.attr.from; + local node, host, resource; + local from_node, from_host, from_resource; + local to_bare, from_bare; + if to then + if full_sessions[to] or bare_sessions[to] or hosts[to] then + node, host = jid_split(to); -- TODO only the host is needed, optimize + else + node, host, resource = jid_prepped_split(to); + if not host then + log("warn", "Received stanza with invalid destination JID: %s", to); + if stanza.attr.type ~= "error" and stanza.attr.type ~= "result" then + origin.send(st.error_reply(stanza, "modify", "jid-malformed", "The destination address is invalid: "..to)); + end + return; + end + to_bare = node and (node.."@"..host) or host; -- bare JID + if resource then to = to_bare.."/"..resource; else to = to_bare; end + stanza.attr.to = to; + end + end + if from and not origin.full_jid then + -- We only stamp the 'from' on c2s stanzas, so we still need to check validity + from_node, from_host, from_resource = jid_prepped_split(from); + if not from_host then + log("warn", "Received stanza with invalid source JID: %s", from); + if stanza.attr.type ~= "error" and stanza.attr.type ~= "result" then + origin.send(st.error_reply(stanza, "modify", "jid-malformed", "The source address is invalid: "..from)); + end + return; + end + from_bare = from_node and (from_node.."@"..from_host) or from_host; -- bare JID + if from_resource then from = from_bare.."/"..from_resource; else from = from_bare; end + stanza.attr.from = from; + end + + if (origin.type == "s2scin") and xmlns == nil then + if origin.type == "s2csin" and not origin.dummy then + local host_status = origin.hosts[from_host]; + if not host_status or not host_status.authed then -- remote server trying to impersonate some other server? + log("warn", "Received a stanza claiming to be from %s, over a stream authed for %s!", from_host, origin.from_host); + origin:close("not-authorized"); + return; + elseif not hosts[host] then + log("warn", "Remote server %s sent us a stanza for %s, closing stream", origin.from_host, host); + origin:close("host-unknown"); + return; + end + end + cluster.core_post_stanza(origin, stanza, origin.full_jid); + else + local h = hosts[stanza.attr.to or origin.host or origin.to_host]; + if h then + local event; + if xmlns == nil then + if stanza.name == "iq" and (stanza.attr.type == "set" or stanza.attr.type == "get") then + event = "stanza/iq/"..stanza.tags[1].attr.xmlns..":"..stanza.tags[1].name; + else + event = "stanza/"..stanza.name; + end + else + event = "stanza/"..xmlns..":"..stanza.name; + end + module:log("debug", "--- event:%s stanza:%s", event, tostring(stanza)); + if h.events.fire_event(event, {origin = origin, stanza = stanza}) then return; end + end + if host and not hosts[host] then host = nil; end -- COMPAT: workaround for a Pidgin bug which sets 'to' to the SRV result + handle_unhandled_stanza(host or origin.host or origin.to_host, origin, stanza); + end +end + +function cluster.core_route_stanza(origin, stanza) + module:log("debug", "--- sipwise_cluster.core_route_staza --"); + --module:log("debug", "--- origin:%s stanza:%s", + -- ut.table.tostring(origin), tostring(stanza)); + if not ut.string.starts(origin.type, "s2sc") then return core.route_stanza(origin, stanza) end + module:log("debug", "---------------- send error ---"); + --origin.sends2sc(st.error_reply(stanza, "cancel", "service-unavailable")); +end + +function cluster.core_post_stanza(origin, stanza, preevents) + module:log("debug", "--- sipwise_cluster.core_post_staza --"); + --module:log("debug", "--- origin:%s stanza:%s", + -- ut.table.tostring(origin), tostring(stanza)); + if not ut.string.starts(origin.type, "s2sc") then return core.post_stanza(origin, stanza) end + local to = stanza.attr.to; + local node, host, resource = jid_split(to); + local to_bare = node and (node.."@"..host) or host; -- bare JID + + local to_type, to_self; + if node then + if resource then + to_type = '/full'; + else + to_type = '/bare'; + if node == origin.username and host == origin.host then + stanza.attr.to = nil; + to_self = true; + end + end + else + if host then + to_type = '/host'; + else + to_type = '/bare'; + to_self = true; + end + end + + local event_data = {origin=origin, stanza=stanza}; + if preevents then -- c2s connection + module:log("debug", "fire event:%s stanza:%s", 'pre-'..stanza.name..to_type, tostring(stanza)); + if hosts[origin.host].events.fire_event('pre-'..stanza.name..to_type, event_data) then return; end -- do preprocessing + end + local h = hosts[to_bare] or hosts[host or origin.host]; + if h then + module:log("debug", "fire event:%s stanza:%s", stanza.name..to_type, tostring(stanza)); + if h.events.fire_event(stanza.name..to_type, event_data) then return; end -- do processing + module:log("debug", "fire event:%s stanza:%s", stanza.name..'/self', tostring(stanza)); + if to_self and h.events.fire_event(stanza.name..'/self', event_data) then return; end -- do processing + handle_unhandled_stanza(h.host, origin, stanza); + else + cluster.core_route_stanza(origin, stanza); + end +end + +function set_dialback_secret(host) + local h = core.hosts[host]; + module:log("debug", "[%s] set cluster dialback_secret", host); + --module:log("debug", "[%s] %s->%s", host, h.dialback_secret, + -- cluster_config.dialback_secret); + h.dialback_secret = cluster_config.dialback_secret; +end + +function module.load() + cluster_config = module:get_option("cluster", cluster_config); + cluster.dialback_secret = cluster_config.dialback_secret; + -- TODO check cluster_config.hosts is a set and does not have me + if cluster.dialback_secret then + module:hook("host-activated", set_dialback_secret, 100); + end +end + +local function route_cluster(origin, stanza, dest) + local from_node, from_host, from_resource = jid_split(stanza.attr.from); + local to_node, to_host, to_resource = jid_split(stanza.attr.to); + + -- Auto-detect origin if not specified + origin = origin or hosts[from_host]; + if not origin then return false; end + + log("debug", "Routing[%s] to remote cluster[%s]...", tostring(from_host), dest); + local host_session = hosts[from_host] or hosts[to_host]; + if not host_session then + log("error", "No hosts[from_host] or hosts[to_host] (please report): %s", tostring(stanza)); + else + local xmlns = stanza.attr.xmlns; + stanza.attr.xmlns = nil; + local routed = host_session.events.fire_event("route/remote_cluster", { origin = origin, stanza = stanza, from_host = from_host, to_host = dest }); + stanza.attr.xmlns = xmlns; -- reset + if not routed then + log("debug", "... no, just kidding."); + if stanza.attr.type == "error" or (stanza.name == "iq" and stanza.attr.type == "result") then return; end + cluster.core_route_stanza(host_session, st.error_reply(stanza, "cancel", "not-allowed", "Communication with remote domains is not enabled")); + end + end +end + +local function outbound_handler(event) + local origin, stanza = event.origin, event.stanza; + local h, resorces, res; + local to = stanza.attr.to; + local stanza_c; + + if origin and ut.string.starts(origin.type, "s2sc") then + module:log("[outbound] stanza coming from the cluster"); + return + end + if to then + local node, host, resource = jid_split(to); + if core.hosts[host] then + module:log("debug", "[outbound] stanza from:%s to:%s", + tostring(stanza.attr.from), tostring(to)); + --module:log("debug", "--- origin:%s stanza:%s", + -- ut.table.tostring(origin), tostring(stanza)); + local rhosts = redis_sessions.get_hosts(stanza.attr.to); + for h,resources in pairs(rhosts) do + if h and h ~= cluster_config.me then + for _,res in pairs(resources) do + stanza_c = st.clone(stanza); + stanza_c.attr.to = node..'@'..host..'/'..res; + module:log("debug", "[outbound] send:%s to hosts:%s/%s", + tostring(stanza_c), h, res); + route_cluster(origin, stanza_c, h); + end + end + end + else + module:log("debug", "[outbound] stanza[%s] not for cluster", host); + end + end +end + +function module.add_host(module) + module:log("debug", "cluster hooks host %s!", module.host); + module:hook("pre-presence/full", outbound_handler, 20); + module:hook("pre-presence/bare", outbound_handler, 20); + module:hook("pre-message/full", outbound_handler, 20); + module:hook("pre-message/bare", outbound_handler, 20); + module:hook("pre-iq/full", outbound_handler, 20); + module:hook("pre-iq/bare", outbound_handler, 20); + -- Stanszas to local clients ?? + module:hook("presence/full", outbound_handler, 20); + module:hook("presence/bare", outbound_handler, 20); + module:hook("message/full", outbound_handler, 20); + module:hook("message/bare", outbound_handler, 20); + module:hook("iq/full", outbound_handler, 20); + module:hook("iq/bare", outbound_handler, 20); +end diff --git a/plugins/mod_sipwise_redis_sessions.lua b/plugins/mod_sipwise_redis_sessions.lua index 4edf919..6308fd0 100644 --- a/plugins/mod_sipwise_redis_sessions.lua +++ b/plugins/mod_sipwise_redis_sessions.lua @@ -6,7 +6,7 @@ -- module:set_global(); -require "util.table"; +local ut = require "util.table"; local jid = require "util.jid"; local array = require "util.array"; local redis = require 'redis'; @@ -28,8 +28,8 @@ end local function client_connect() redis_client = redis.connect(redis_config.host, redis_config.port); - module:log("debug", "connected to redis server %s:%d", - redis_config.host, redis_config.port); + --module:log("debug", "connected to redis server %s:%d", + --redis_config.host, redis_config.port); if redis_config.redis_db then redis_client:select(redis_config.redis_db); end @@ -46,7 +46,6 @@ local function resource_bind(event) redis_client:set(full_jid, redis_config.server_id); module:log("debug", "append [%s]=>%s:%s", bare_jid, redis_config.server_id, resource); redis_client:sadd(bare_jid, redis_config.server_id..":"..resource); - module:log("debug", "done"); end local function resource_unbind(event) @@ -60,11 +59,10 @@ local function resource_unbind(event) redis_client:del(full_jid); module:log("debug", "remove [%s]=>%s:%s", bare_jid, redis_config.server_id, resource); redis_client:srem(bare_jid, redis_config.server_id..":"..resource); - module:log("debug", "done"); end local function split_key(key) - local t = explode(':', key); + local t = ut.string.explode(':', key); return t[1], t[2]; end @@ -77,10 +75,10 @@ function redis_sessions.get_hosts(j) module:log("debug", "search session:%s host", bare_jid); if not test_connection() then client_connect() end l = redis_client:smembers(bare_jid); - module:log("debug", "l:%s", table.tostring(l)); + --module:log("debug", "l:%s", ut.table.tostring(l)); for _,v in pairs(l) do h, r = split_key(v); - module:log("debug", "h:%s r:%s", tostring(h), tostring(r)); + --module:log("debug", "h:%s r:%s", tostring(h), tostring(r)); if not res[h] then res[h] = array() end res[h]:push(r); end diff --git a/util/table.lua b/util/table.lua index c131ae3..7236536 100644 --- a/util/table.lua +++ b/util/table.lua @@ -19,6 +19,10 @@ -- -- Lua utils +local type = type; +local string = string; +local t_insert, t_concat, t_remove, t_sort = table.insert, table.concat, table.remove, table.sort; + -- copy a table function table.deepcopy(object) local lookup_table = {} @@ -52,7 +56,7 @@ end -- add if element is not in table function table.add(t, element) if not table.contains(t, element) then - table.insert(t, element) + t_insert(t, element) end end @@ -80,16 +84,16 @@ end function table.tostring( tbl ) local result, done = {}, {} for k, v in ipairs( tbl ) do - table.insert( result, table.val_to_str( v ) ) + t_insert( result, table.val_to_str( v ) ) done[ k ] = true end for k, v in pairs( tbl ) do if not done[ k ] then - table.insert( result, + t_insert( result, table.key_to_str( k ) .. "=" .. table.val_to_str( v ) ) end end - return "{" .. table.concat( result, "," ) .. "}" + return "{" .. t_concat( result, "," ) .. "}" end -- from table to string @@ -98,7 +102,7 @@ end -- "'a','b'" -- implode("#",t) -- "a#b" -function implode(delimiter, list, quoter) +function table.implode(delimiter, list, quoter) local len = #list if not delimiter then error("delimiter is nil") @@ -117,7 +121,7 @@ function implode(delimiter, list, quoter) end -- from string to table -function explode(delimiter, text) +function string.explode(delimiter, text) local list = {} local pos = 1 @@ -135,10 +139,10 @@ function explode(delimiter, text) local first, last = string.find(text, delimiter, pos) -- print (first, last) if first then - table.insert(list, string.sub(text, pos, first-1)) + t_insert(list, string.sub(text, pos, first-1)) pos = last+1 else - table.insert(list, string.sub(text, pos)) + t_insert(list, string.sub(text, pos)) break end end @@ -152,3 +156,5 @@ end function string.ends(String,End) return End=='' or string.sub(String,-string.len(End))==End end + +return {table=table, string=string}