mirror of https://github.com/sipwise/prosody.git
Victor Seva
13 years ago
parent
aae45dab2b
commit
3f2c23f051
@ -0,0 +1,100 @@
|
||||
-- Prosody IM
|
||||
-- Copyright (C) 2008-2010 Matthew Wild
|
||||
-- Copyright (C) 2008-2010 Waqas Hussain
|
||||
-- Copyright (C) 2014 Sipwise GmbH <development@sipwise.com>
|
||||
--
|
||||
-- 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;
|
||||
@ -1,2 +1,3 @@
|
||||
plugins/* /usr/lib/prosody/modules/
|
||||
util/* /usr/lib/prosody/util/
|
||||
core/* /usr/lib/prosody/core/
|
||||
|
||||
@ -0,0 +1,690 @@
|
||||
-- Prosody IM
|
||||
-- Copyright (C) 2008-2010 Matthew Wild
|
||||
-- Copyright (C) 2008-2010 Waqas Hussain
|
||||
-- Copyright (C) 2014 Sipwise GmbH <development@sipwise.com>
|
||||
--
|
||||
-- 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 </stream:stream>");
|
||||
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], <stream:error> 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], <stream:error> 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], <stream:error> 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("</stream:stream>");
|
||||
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 </stream:stream> 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("<?xml version='1.0'?>");
|
||||
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.*>";
|
||||
};
|
||||
});
|
||||
|
||||
@ -0,0 +1,158 @@
|
||||
-- Prosody IM
|
||||
-- Copyright (C) 2008-2010 Matthew Wild
|
||||
-- Copyright (C) 2008-2010 Waqas Hussain
|
||||
-- Copyright (C) 2014 Sipwise GmbH <development@sipwise.com>
|
||||
--
|
||||
-- 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;
|
||||
@ -0,0 +1,185 @@
|
||||
-- Prosody IM
|
||||
-- Copyright (C) 2008-2010 Matthew Wild
|
||||
-- Copyright (C) 2008-2010 Waqas Hussain
|
||||
-- Copyright (C) 2014 Sipwise GmbH <development@sipwise.com>
|
||||
--
|
||||
-- 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);
|
||||
@ -0,0 +1,304 @@
|
||||
-- Prosody IM
|
||||
-- Copyright (C) 2014 Sipwise GmbH <development@sipwise.com>
|
||||
--
|
||||
-- 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
|
||||
Loading…
Reference in new issue