core/s2smanager.lua

   1 -- Prosody IM v0.4
   2 -- Copyright (C) 2008-2009 Matthew Wild
   3 -- Copyright (C) 2008-2009 Waqas Hussain
   4 --
   5 -- This project is MIT/X11 licensed. Please see the
   6 -- COPYING file in the source package for more information.
   7 --
   8
   9
  10
  11 local hosts = hosts;
  12 local sessions = sessions;
  13 local core_process_stanza = function(a, b) core_process_stanza(a, b); end
  14 local socket = require "socket";
  15 local format = string.format;
  16 local t_insert, t_sort = table.insert, table.sort;
  17 local get_traceback = debug.traceback;
  18 local tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber
  19     = tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber;
  20
  21 local idna_to_ascii = require "util.encodings".idna.to_ascii;
  22 local connlisteners_get = require "net.connlisteners".get;
  23 local wrapclient = require "net.server".wrapclient;
  24 local modulemanager = require "core.modulemanager";
  25 local st = require "stanza";
  26 local stanza = st.stanza;
  27 local nameprep = require "util.encodings".stringprep.nameprep;
  28
  29 local uuid_gen = require "util.uuid".generate;
  30
  31 local logger_init = require "util.logger".init;
  32
  33 local log = logger_init("s2smanager");
  34
  35 local sha256_hash = require "util.hashes".sha256;
  36
  37 local dialback_secret = sha256_hash(tostring{} .. math.random() .. socket.gettime(), true);
  38
  39 local dns = require "net.dns";
  40
  41 incoming_s2s = {};
  42 local incoming_s2s = incoming_s2s;
  43
  44 module "s2smanager"
  45
  46 local function compare_srv_priorities(a,b) return a.priority < b.priority or a.weight < b.weight; end
  47
  48 local function bounce_sendq(session)
  49         local sendq = session.sendq;
  50         if sendq then
  51                 session.log("debug", "sending error replies for "..#sendq.." queued stanzas because of failed outgoing connection to "..tostring(session.to_host));
  52                 local dummy = {
  53                         type = "s2sin";
  54                         send = function(s)
  55                                 (session.log or log)("error", "Replying to to an s2s error reply, please report this! Traceback: %s", get_traceback());
  56                         end;
  57                         dummy = true;
  58                 };
  59                 for i, data in ipairs(sendq) do
  60                         local reply = data[2];
  61                         local xmlns = reply.attr.xmlns;
  62                         if not xmlns or xmlns == "jabber:client" or xmlns == "jabber:server" then
  63                                 reply.attr.type = "error";
  64                                 reply:tag("error", {type = "cancel"})
  65                                         :tag("remote-server-not-found", {xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas"}):up();
  66                                 core_process_stanza(dummy, reply);
  67                         end
  68                         sendq[i] = nil;
  69                 end
  70                 session.sendq = nil;
  71         end
  72 end
  73
  74 function send_to_host(from_host, to_host, data)
  75         local host = hosts[from_host].s2sout[to_host];
  76         if host then
  77                 -- We have a connection to this host already
  78                 if host.type == "s2sout_unauthed" and data.name ~= "db:verify" and ((not data.xmlns) or data.xmlns == "jabber:client" or data.xmlns == "jabber:server") then
  79                         (host.log or log)("debug", "trying to send over unauthed s2sout to "..to_host);
  80                         if not host.notopen and not host.dialback_key then
  81                                 host.log("debug", "dialback had not been initiated");
  82                                 initiate_dialback(host);
  83                         end
  84
  85                         -- Queue stanza until we are able to send it
  86                         if host.sendq then t_insert(host.sendq, {tostring(data), st.reply(data)});
  87                         else host.sendq = { {tostring(data), st.reply(data)} }; end
  88                         host.log("debug", "stanza [%s] queued ", data.name);
  89                 elseif host.type == "local" or host.type == "component" then
  90                         log("error", "Trying to send a stanza to ourselves??")
  91                         log("error", "Traceback: %s", get_traceback());
  92                         log("error", "Stanza: %s", tostring(data));
  93                 else
  94                         (host.log or log)("debug", "going to send stanza to "..to_host.." from "..from_host);
  95                         -- FIXME
  96                         if host.from_host ~= from_host then
  97                                 log("error", "WARNING! This might, possibly, be a bug, but it might not...");
  98                                 log("error", "We are going to send from %s instead of %s", tostring(host.from_host), tostring(from_host));
  99                         end
 100                         host.sends2s(data);
 101                         host.log("debug", "stanza sent over "..host.type);
 102                 end
 103         else
 104                 log("debug", "opening a new outgoing connection for this stanza");
 105                 local host_session = new_outgoing(from_host, to_host);
 106                 -- Store in buffer
 107                 host_session.sendq = { {tostring(data), st.reply(data)} };
 108                 if not host_session.conn then destroy_session(host_session); end
 109         end
 110 end
 111
 112 local open_sessions = 0;
 113
 114 function new_incoming(conn)
 115         local session = { conn = conn, type = "s2sin_unauthed", direction = "incoming", hosts = {} };
 116         if true then
 117                 session.trace = newproxy(true);
 118                 getmetatable(session.trace).__gc = function () open_sessions = open_sessions - 1; end;
 119         end
 120         open_sessions = open_sessions + 1;
 121         local w, log = conn.write, logger_init("s2sin"..tostring(conn):match("[a-f0-9]+$"));
 122         session.sends2s = function (t) log("debug", "sending: %s", tostring(t)); w(tostring(t)); end
 123         incoming_s2s[session] = true;
 124         return session;
 125 end
 126
 127 function new_outgoing(from_host, to_host)
 128                 local host_session = { to_host = to_host, from_host = from_host, notopen = true, type = "s2sout_unauthed", direction = "outgoing" };
 129                 hosts[from_host].s2sout[to_host] = host_session;
 130
 131                 local log;
 132                 do
 133                         local conn_name = "s2sout"..tostring(host_session):match("[a-f0-9]*$");
 134                         log = logger_init(conn_name);
 135                         host_session.log = log;
 136                 end
 137
 138                 attempt_connection(host_session);
 139
 140                 return host_session;
 141 end
 142
 143
 144 function attempt_connection(host_session, err)
 145         local from_host, to_host = host_session.from_host, host_session.to_host;
 146         local conn, handler = socket.tcp()
 147
 148         local connect_host, connect_port = idna_to_ascii(to_host), 5269;
 149
 150         if not err then -- This is our first attempt
 151                 local answer = dns.lookup("_xmpp-server._tcp."..connect_host..".", "SRV");
 152
 153                 if answer then
 154                         log("debug", to_host.." has SRV records, handling...");
 155                         local srv_hosts = {};
 156                         host_session.srv_hosts = srv_hosts;
 157                         for _, record in ipairs(answer) do
 158                                 t_insert(srv_hosts, record.srv);
 159                         end
 160                         t_sort(srv_hosts, compare_srv_priorities);
 161
 162                         local srv_choice = srv_hosts[1];
 163                         host_session.srv_choice = 1;
 164                         if srv_choice then
 165                                 connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
 166                                 log("debug", "Best record found, will connect to %s:%d", connect_host, connect_port);
 167                         end
 168                 end
 169         elseif host_session.srv_hosts and #host_session.srv_hosts > host_session.srv_choice then -- Not our first attempt, and we also have SRV
 170                 host_session.srv_choice = host_session.srv_choice + 1;
 171                 local srv_choice = host_session.srv_hosts[host_session.srv_choice];
 172                 connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
 173                 host_session.log("debug", "Connection failed (%s). Attempt #%d: This time to %s:%d", tostring(err), host_session.srv_choice, connect_host, connect_port);
 174         else
 175                 host_session.log("debug", "Out of connection options, can't connect to %s", tostring(host_session.to_host));
 176                 -- We're out of options
 177                 return false;
 178         end
 179
 180         -- Ok, we're going to try to connect
 181         conn:settimeout(0);
 182         local success, err = conn:connect(connect_host, connect_port);
 183         if not success and err ~= "timeout" then
 184                 log("warn", "s2s connect() failed: %s", err);
 185                 return false;
 186         end
 187
 188         local cl = connlisteners_get("xmppserver");
 189         conn = wrapclient(conn, connect_host, connect_port, cl, cl.default_mode or 1, hosts[from_host].ssl_ctx, false );
 190         host_session.conn = conn;
 191
 192         -- Register this outgoing connection so that xmppserver_listener knows about it
 193         -- otherwise it will assume it is a new incoming connection
 194         cl.register_outgoing(conn, host_session);
 195
 196         local w = conn.write;
 197         host_session.sends2s = function (t) log("debug", "sending: %s", tostring(t)); w(tostring(t)); end
 198
 199         conn.write(format([[<stream:stream xmlns='jabber:server' xmlns:db='jabber:server:dialback' xmlns:stream='http://etherx.jabber.org/streams' from='%s' to='%s' version='1.0'>]], from_host, to_host));
 200         return true;
 201 end
 202
 203 function streamopened(session, attr)
 204         local send = session.sends2s;
 205
 206         -- TODO: #29: SASL/TLS on s2s streams
 207         session.version = 0; --tonumber(attr.version) or 0;
 208
 209         if session.version >= 1.0 and not (attr.to and attr.from) then
 210                 log("warn", (session.to_host or "(unknown)").." failed to specify 'to' or 'from' hostname as per RFC");
 211         end
 212
 213         if session.direction == "incoming" then
 214                 -- Send a reply stream header
 215                 session.to_host = attr.to and nameprep(attr.to);
 216                 session.from_host = attr.from and nameprep(attr.from);
 217
 218                 session.streamid = uuid_gen();
 219                 (session.log or log)("debug", "incoming s2s received <stream:stream>");
 220                 send("<?xml version='1.0'?>");
 221                 send(stanza("stream:stream", { xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback',
 222                                 ["xmlns:stream"]='http://etherx.jabber.org/streams', id=session.streamid, from=session.to_host }):top_tag());
 223                 if session.to_host and not hosts[session.to_host] then
 224                         -- Attempting to connect to a host we don't serve
 225                         session:close({ condition = "host-unknown"; text = "This host does not serve "..session.to_host });
 226                         return;
 227                 end
 228                 if session.version >= 1.0 then
 229                         send(st.stanza("stream:features")
 230                                         :tag("dialback", { xmlns='urn:xmpp:features:dialback' }):tag("optional"):up():up());
 231                 end
 232         elseif session.direction == "outgoing" then
 233                 -- If we are just using the connection for verifying dialback keys, we won't try and auth it
 234                 if not attr.id then error("stream response did not give us a streamid!!!"); end
 235                 session.streamid = attr.id;
 236
 237                 if not session.dialback_verifying then
 238                         initiate_dialback(session);
 239                 else
 240                         mark_connected(session);
 241                 end
 242         end
 243
 244         session.notopen = nil;
 245 end
 246
 247 function initiate_dialback(session)
 248         -- generate dialback key
 249         session.dialback_key = generate_dialback(session.streamid, session.to_host, session.from_host);
 250         session.sends2s(format("<db:result from='%s' to='%s'>%s</db:result>", session.from_host, session.to_host, session.dialback_key));
 251         session.log("info", "sent dialback key on outgoing s2s stream");
 252 end
 253
 254 function generate_dialback(id, to, from)
 255         return sha256_hash(id..to..from..dialback_secret, true);
 256 end
 257
 258 function verify_dialback(id, to, from, key)
 259         return key == generate_dialback(id, to, from);
 260 end
 261
 262 function make_authenticated(session, host)
 263         if session.type == "s2sout_unauthed" then
 264                 session.type = "s2sout";
 265         elseif session.type == "s2sin_unauthed" then
 266                 session.type = "s2sin";
 267                 if host then
 268                         session.hosts[host].authed = true;
 269                 end
 270         elseif session.type == "s2sin" and host then
 271                 session.hosts[host].authed = true;
 272         else
 273                 return false;
 274         end
 275         session.log("info", "connection is now authenticated");
 276
 277         mark_connected(session);
 278
 279         return true;
 280 end
 281
 282 function mark_connected(session)
 283         local sendq, send = session.sendq, session.sends2s;
 284
 285         local from, to = session.from_host, session.to_host;
 286
 287         session.log("debug", session.direction.." s2s connection "..from.."->"..to.." is now complete");
 288
 289         local send_to_host = send_to_host;
 290         function session.send(data) send_to_host(to, from, data); end
 291
 292
 293         if session.direction == "outgoing" then
 294                 if sendq then
 295                         session.log("debug", "sending "..#sendq.." queued stanzas across new outgoing connection to "..session.to_host);
 296                         for i, data in ipairs(sendq) do
 297                                 send(data[1]);
 298                                 sendq[i] = nil;
 299                         end
 300                         session.sendq = nil;
 301                 end
 302         end
 303 end
 304
 305 function destroy_session(session)
 306         (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host));
 307
 308
 309         if session.direction == "outgoing" then
 310                 hosts[session.from_host].s2sout[session.to_host] = nil;
 311                 bounce_sendq(session);
 312         elseif session.direction == "incoming" then
 313                 incoming_s2s[session] = nil;
 314         end
 315
 316         for k in pairs(session) do
 317                 if k ~= "trace" then
 318                         session[k] = nil;
 319                 end
 320         end
 321 end
 322
 323 return _M;