util.pposix: Compatibility with Solaris systems (thanks Filip)
[prosody.git] / core / s2smanager.lua
index 5f04fed6afe34179d804ec36f0a753db2b7d0d01..daf2c871f8da6e7deb7008c735db625eea9cb72c 100644 (file)
@@ -1,4 +1,4 @@
--- Prosody IM v0.4
+-- Prosody IM
 -- Copyright (C) 2008-2009 Matthew Wild
 -- Copyright (C) 2008-2009 Waqas Hussain
 -- 
@@ -11,6 +11,7 @@
 local hosts = hosts;
 local sessions = sessions;
 local core_process_stanza = function(a, b) core_process_stanza(a, b); end
+local add_task = require "util.timer".add_task;
 local socket = require "socket";
 local format = string.format;
 local t_insert, t_sort = table.insert, table.sort;
@@ -34,9 +35,13 @@ local log = logger_init("s2smanager");
 
 local sha256_hash = require "util.hashes".sha256;
 
-local dialback_secret = sha256_hash(tostring{} .. math.random() .. socket.gettime(), true);
+local dialback_secret = uuid_gen();
 
-local adns = require "net.adns";
+local adns, dns = require "net.adns", require "net.dns";
+
+local connect_timeout = config.get("*", "core", "s2s_timeout") or 60;
+local dns_timeout = config.get("*", "core", "dns_timeout") or 60;
+local max_dns_depth = config.get("*", "core", "dns_max_depth") or 3;
 
 incoming_s2s = {};
 local incoming_s2s = incoming_s2s;
@@ -48,7 +53,7 @@ local function compare_srv_priorities(a,b) return a.priority < b.priority or a.w
 local function bounce_sendq(session)
        local sendq = session.sendq;
        if sendq then
-               session.log("debug", "sending error replies for "..#sendq.." queued stanzas because of failed outgoing connection to "..tostring(session.to_host));
+               session.log("info", "sending error replies for "..#sendq.." queued stanzas because of failed outgoing connection to "..tostring(session.to_host));
                local dummy = {
                        type = "s2sin";
                        send = function(s)
@@ -77,7 +82,7 @@ function send_to_host(from_host, to_host, data)
                -- We have a connection to this host already
                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
                        (host.log or log)("debug", "trying to send over unauthed s2sout to "..to_host);
-                       if not host.notopen and not host.dialback_key then
+                       if not host.notopen and not host.dialback_key and host.sends2s then
                                host.log("debug", "dialback had not been initiated");
                                initiate_dialback(host);
                        end
@@ -105,7 +110,11 @@ function send_to_host(from_host, to_host, data)
                local host_session = new_outgoing(from_host, to_host);
                -- Store in buffer
                host_session.sendq = { {tostring(data), st.reply(data)} };
-               if (not host_session.connecting) and (not host_session.conn) then destroy_session(host_session); end
+               log("debug", "stanza [%s] queued until connection complete", tostring(data.name));
+               if (not host_session.connecting) and (not host_session.conn) then
+                       log("warn", "Connection to %s failed already, destroying session...", to_host);
+                       destroy_session(host_session);
+               end
        end
 end
 
@@ -119,8 +128,19 @@ function new_incoming(conn)
        end
        open_sessions = open_sessions + 1;
        local w, log = conn.write, logger_init("s2sin"..tostring(conn):match("[a-f0-9]+$"));
+       session.log = log;
        session.sends2s = function (t) log("debug", "sending: %s", tostring(t)); w(tostring(t)); end
        incoming_s2s[session] = true;
+       add_task(connect_timeout, function ()
+               if session.conn ~= conn or
+                  session.type == "s2sin" then
+                       return; -- Ok, we're connect[ed|ing]
+               end
+               -- Not connected, need to close session and clean up
+               (session.log or log)("warn", "Destroying incomplete session %s->%s due to inactivity", 
+                   session.from_host or "(unknown)", session.to_host or "(unknown)");
+               session:close("connection-timeout");
+       end);
        return session;
 end
 
@@ -135,8 +155,26 @@ function new_outgoing(from_host, to_host)
                        host_session.log = log;
                end
                
+               -- This is the first call, can't fail (the first step is DNS lookup)
                attempt_connection(host_session);
                
+               if not host_session.sends2s then                
+                       -- A sends2s 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.sends2s(data)
+                               if not buffer then
+                                       buffer = {};
+                                       host_session.send_buffer = buffer;
+                               end
+                               log("debug", "Buffering data on unconnected s2sout to %s", to_host);
+                               buffer[#buffer+1] = data;
+                               log("debug", "Buffered item %d: %s", #buffer, tostring(data));
+                       end
+                       
+               end
+
                return host_session;
 end
 
@@ -146,9 +184,11 @@ function attempt_connection(host_session, err)
        local connect_host, connect_port = idna_to_ascii(to_host), 5269;
        
        if not err then -- This is our first attempt
+               log("debug", "First attempt to connect to %s, starting with SRV lookup...", to_host);
                host_session.connecting = true;
-               local answer = 
-               adns.lookup(function (answer)
+               local handle;
+               handle = adns.lookup(function (answer)
+                       handle = nil;
                        host_session.connecting = nil;
                        if answer then
                                log("debug", to_host.." has SRV records, handling...");
@@ -169,22 +209,38 @@ function attempt_connection(host_session, err)
                                log("debug", to_host.." has no SRV records, falling back to A");
                        end
                        -- Try with SRV, or just the plain hostname if no SRV
-                       return try_connect(host_session, connect_host, connect_port);
+                       local ok, err = try_connect(host_session, connect_host, connect_port);
+                       if not ok then
+                               if not attempt_connection(host_session, err) then
+                                       -- No more attempts will be made
+                                       destroy_session(host_session);
+                               end
+                       end
                end, "_xmpp-server._tcp."..connect_host..".", "SRV");
+               
+               -- Set handler for DNS timeout
+               add_task(dns_timeout, function ()
+                       if handle then
+                               adns.cancel(handle, true);
+                       end
+               end);
+               
+               log("debug", "DNS lookup for %s sent, waiting for response before we can connect", to_host);
                return true; -- Attempt in progress
        elseif host_session.srv_hosts and #host_session.srv_hosts > host_session.srv_choice then -- Not our first attempt, and we also have SRV
                host_session.srv_choice = host_session.srv_choice + 1;
                local srv_choice = host_session.srv_hosts[host_session.srv_choice];
                connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
-               host_session.log("debug", "Connection failed (%s). Attempt #%d: This time to %s:%d", tostring(err), host_session.srv_choice, connect_host, connect_port);
+               host_session.log("info", "Connection failed (%s). Attempt #%d: This time to %s:%d", tostring(err), host_session.srv_choice, connect_host, connect_port);
        else
-               host_session.log("debug", "Out of connection options, can't connect to %s", tostring(host_session.to_host));
+               host_session.log("info", "Out of connection options, can't connect to %s", tostring(host_session.to_host));
                -- We're out of options
                return false;
        end
        
        if not (connect_host and connect_port) then
                -- Likely we couldn't resolve DNS
+               log("warn", "Hmm, we're without a host (%s) and port (%s) to connect to for %s, giving up :(", tostring(connect_host), tostring(connect_port), tostring(to_host));
                return false;
        end
        
@@ -192,7 +248,48 @@ function attempt_connection(host_session, err)
 end
 
 function try_connect(host_session, connect_host, connect_port)
-       log("debug", "Beginning new connection attempt to %s (%s:%d)", host_session.to_host, connect_host, connect_port);
+       host_session.connecting = true;
+       local handle;
+       handle = adns.lookup(function (reply)
+               handle = nil;
+               host_session.connecting = nil;
+               
+               -- COMPAT: This is a compromise for all you CNAME-(ab)users :)
+               if not (reply and reply[#reply] and reply[#reply].a) then
+                       local count = max_dns_depth;
+                       reply = dns.peek(connect_host, "CNAME", "IN");
+                       while count > 0 and reply and reply[#reply] and not reply[#reply].a and reply[#reply].cname do
+                               log("debug", "Looking up %s (DNS depth is %d)", tostring(reply[#reply].cname), count);
+                               reply = dns.peek(reply[#reply].cname, "A", "IN") or dns.peek(reply[#reply].cname, "CNAME", "IN");
+                               count = count - 1;
+                       end
+               end
+               -- end of CNAME resolving
+               
+               if reply and reply[#reply] and reply[#reply].a then
+                       log("debug", "DNS reply for %s gives us %s", connect_host, reply[#reply].a);
+                       return make_connect(host_session, reply[#reply].a, connect_port);
+               else
+                       log("debug", "DNS lookup failed to get a response for %s", connect_host);
+                       if not attempt_connection(host_session, "name resolution failed") then -- Retry if we can
+                               log("debug", "No other records to try for %s - destroying", host_session.to_host);
+                               destroy_session(host_session); -- End of the line, we can't
+                       end
+               end
+       end, connect_host, "A", "IN");
+
+       -- Set handler for DNS timeout
+       add_task(dns_timeout, function ()
+               if handle then
+                       adns.cancel(handle, true);
+               end
+       end);
+               
+       return true;
+end
+
+function make_connect(host_session, connect_host, connect_port)
+       host_session.log("info", "Beginning new connection attempt to %s (%s:%d)", host_session.to_host, connect_host, connect_port);
        -- Ok, we're going to try to connect
        
        local from_host, to_host = host_session.from_host, host_session.to_host;
@@ -202,8 +299,8 @@ function try_connect(host_session, connect_host, connect_port)
        conn:settimeout(0);
        local success, err = conn:connect(connect_host, connect_port);
        if not success and err ~= "timeout" then
-               log("warn", "s2s connect() failed: %s", err);
-               return false;
+               log("warn", "s2s connect() to %s (%s:%d) failed: %s", host_session.to_host, connect_host, connect_port, err);
+               return false, err;
        end
        
        local cl = connlisteners_get("xmppserver");
@@ -217,8 +314,19 @@ function try_connect(host_session, connect_host, connect_port)
        local w = conn.write;
        host_session.sends2s = function (t) log("debug", "sending: %s", tostring(t)); w(tostring(t)); end
        
-       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));
+       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' xml:lang='en'>]], from_host, to_host));
        log("debug", "Connection attempt in progress...");
+       add_task(connect_timeout, function ()
+               if host_session.conn ~= conn or
+                  host_session.type == "s2sout" or
+                  host_session.connecting then
+                       return; -- Ok, we're connect[ed|ing]
+               end
+               -- Not connected, need to close session and clean up
+               (host_session.log or log)("warn", "Destroying incomplete session %s->%s due to inactivity", 
+                   host_session.from_host or "(unknown)", host_session.to_host or "(unknown)");
+               host_session:close("connection-timeout");
+       end);
        return true;
 end
 
@@ -256,6 +364,20 @@ function streamopened(session, attr)
                if not attr.id then error("stream response did not give us a streamid!!!"); end
                session.streamid = attr.id;
        
+               -- 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 s2s send_buffer now...");
+                       for i, data in ipairs(send_buffer) do
+                               session.sends2s(tostring(data));
+                               send_buffer[i] = nil;
+                       end
+               end
+               session.send_buffer = nil;
+       
                if not session.dialback_verifying then
                        initiate_dialback(session);
                else
@@ -267,7 +389,10 @@ function streamopened(session, attr)
 end
 
 function streamclosed(session)
-       session.sends2s("</stream:stream>");
+       (session.log or log)("debug", "</stream:stream>");
+       if session.sends2s then
+               session.sends2s("</stream:stream>");
+       end
        session.notopen = true;
 end
 
@@ -299,7 +424,7 @@ function make_authenticated(session, host)
        else
                return false;
        end
-       session.log("info", "connection is now authenticated");
+       session.log("debug", "connection %s->%s is now authenticated", session.from_host or "(unknown)", session.to_host or "(unknown)");
        
        mark_connected(session);
        
@@ -311,7 +436,7 @@ function mark_connected(session)
        
        local from, to = session.from_host, session.to_host;
        
-       session.log("debug", session.direction.." s2s connection "..from.."->"..to.." is now complete");
+       session.log("info", session.direction.." s2s connection "..from.."->"..to.." complete");
        
        local send_to_host = send_to_host;
        function session.send(data) send_to_host(to, from, data); end
@@ -326,6 +451,8 @@ function mark_connected(session)
                        end
                        session.sendq = nil;
                end
+               
+               session.srv_hosts = nil;
        end
 end