plugins/mod_s2s/s2sout.lib.lua

   1 -- Prosody IM
   2 -- Copyright (C) 2008-2010 Matthew Wild
   3 -- Copyright (C) 2008-2010 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 --- Module containing all the logic for connecting to a remote server
  10
  11 -- luacheck: ignore 432/err
  12
  13 local portmanager = require "core.portmanager";
  14 local wrapclient = require "net.server".wrapclient;
  15 local initialize_filters = require "util.filters".initialize;
  16 local idna_to_ascii = require "util.encodings".idna.to_ascii;
  17 local new_ip = require "util.ip".new_ip;
  18 local rfc6724_dest = require "util.rfc6724".destination;
  19 local socket = require "socket";
  20 local adns = require "net.adns";
  21 local t_insert, t_sort, ipairs = table.insert, table.sort, ipairs;
  22 local local_addresses = require "util.net".local_addresses;
  23
  24 local s2s_destroy_session = require "core.s2smanager".destroy_session;
  25
  26 local default_mode = module:get_option("network_default_read_size", 4096);
  27
  28 local log = module._log;
  29
  30 local sources = {};
  31 local has_ipv4, has_ipv6;
  32
  33 local dns_timeout = module:get_option_number("dns_timeout", 15);
  34 local resolvers = module:get_option_set("s2s_dns_resolvers")
  35
  36 local s2sout = {};
  37
  38 local s2s_listener;
  39
  40
  41 function s2sout.set_listener(listener)
  42         s2s_listener = listener;
  43 end
  44
  45 local function compare_srv_priorities(a,b)
  46         return a.priority < b.priority or (a.priority == b.priority and a.weight > b.weight);
  47 end
  48
  49 function s2sout.initiate_connection(host_session)
  50         local log = host_session.log or log;
  51
  52         initialize_filters(host_session);
  53         host_session.version = 1;
  54
  55         host_session.resolver = adns.resolver();
  56         host_session.resolver._resolver:settimeout(dns_timeout);
  57         if resolvers then
  58                 for resolver in resolvers do
  59                         host_session.resolver._resolver:addnameserver(resolver);
  60                 end
  61         end
  62
  63         -- Kick the connection attempting machine into life
  64         if not s2sout.attempt_connection(host_session) then
  65                 -- Intentionally not returning here, the
  66                 -- session is needed, connected or not
  67                 s2s_destroy_session(host_session);
  68         end
  69
  70         if not host_session.sends2s then
  71                 -- A sends2s which buffers data (until the stream is opened)
  72                 -- note that data in this buffer will be sent before the stream is authed
  73                 -- and will not be ack'd in any way, successful or otherwise
  74                 local buffer;
  75                 function host_session.sends2s(data)
  76                         if not buffer then
  77                                 buffer = {};
  78                                 host_session.send_buffer = buffer;
  79                         end
  80                         log("debug", "Buffering data on unconnected s2sout to %s", host_session.to_host);
  81                         buffer[#buffer+1] = data;
  82                         log("debug", "Buffered item %d: %s", #buffer, data);
  83                 end
  84         end
  85 end
  86
  87 function s2sout.attempt_connection(host_session, err)
  88         local to_host = host_session.to_host;
  89         local connect_host, connect_port = to_host and idna_to_ascii(to_host), 5269;
  90         local log = host_session.log or log;
  91
  92         if not connect_host then
  93                 return false;
  94         end
  95
  96         if not err then -- This is our first attempt
  97                 log("debug", "First attempt to connect to %s, starting with SRV lookup...", to_host);
  98                 host_session.connecting = true;
  99                 host_session.resolver:lookup(function (answer)
 100                         local srv_hosts = { answer = answer };
 101                         host_session.srv_hosts = srv_hosts;
 102                         host_session.srv_choice = 0;
 103                         host_session.connecting = nil;
 104                         if answer and #answer > 0 then
 105                                 log("debug", "%s has SRV records, handling...", to_host);
 106                                 for _, record in ipairs(answer) do
 107                                         t_insert(srv_hosts, record.srv);
 108                                 end
 109                                 if #srv_hosts == 1 and srv_hosts[1].target == "." then
 110                                         log("debug", "%s does not provide a XMPP service", to_host);
 111                                         s2s_destroy_session(host_session, err); -- Nothing to see here
 112                                         return;
 113                                 end
 114                                 t_sort(srv_hosts, compare_srv_priorities);
 115
 116                                 local srv_choice = srv_hosts[1];
 117                                 host_session.srv_choice = 1;
 118                                 if srv_choice then
 119                                         connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
 120                                         log("debug", "Best record found, will connect to %s:%d", connect_host, connect_port);
 121                                 end
 122                         else
 123                                 log("debug", "%s has no SRV records, falling back to A/AAAA", to_host);
 124                         end
 125                         -- Try with SRV, or just the plain hostname if no SRV
 126                         local ok, err = s2sout.try_connect(host_session, connect_host, connect_port);
 127                         if not ok then
 128                                 if not s2sout.attempt_connection(host_session, err) then
 129                                         -- No more attempts will be made
 130                                         s2s_destroy_session(host_session, err);
 131                                 end
 132                         end
 133                 end, "_xmpp-server._tcp."..connect_host..".", "SRV");
 134
 135                 return true; -- Attempt in progress
 136         elseif host_session.ip_hosts then
 137                 return s2sout.try_connect(host_session, connect_host, connect_port, err);
 138         elseif host_session.srv_hosts and #host_session.srv_hosts > host_session.srv_choice then -- Not our first attempt, and we also have SRV
 139                 host_session.srv_choice = host_session.srv_choice + 1;
 140                 local srv_choice = host_session.srv_hosts[host_session.srv_choice];
 141                 connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
 142                 host_session.log("info", "Connection failed (%s). Attempt #%d: This time to %s:%d", err, host_session.srv_choice, connect_host, connect_port);
 143         else
 144                 host_session.log("info", "Failed in all attempts to connect to %s", host_session.to_host);
 145                 -- We're out of options
 146                 return false;
 147         end
 148
 149         if not (connect_host and connect_port) then
 150                 -- Likely we couldn't resolve DNS
 151                 log("warn", "Hmm, we're without a host (%s) and port (%s) to connect to for %s, giving up :(", connect_host, connect_port, to_host);
 152                 return false;
 153         end
 154
 155         return s2sout.try_connect(host_session, connect_host, connect_port);
 156 end
 157
 158 function s2sout.try_next_ip(host_session)
 159         host_session.connecting = nil;
 160         host_session.ip_choice = host_session.ip_choice + 1;
 161         local ip = host_session.ip_hosts[host_session.ip_choice];
 162         local ok, err= s2sout.make_connect(host_session, ip.ip, ip.port);
 163         if not ok then
 164                 if not s2sout.attempt_connection(host_session, err or "closed") then
 165                         err = err and (": "..err) or "";
 166                         s2s_destroy_session(host_session, "Connection failed"..err);
 167                 end
 168         end
 169 end
 170
 171 function s2sout.try_connect(host_session, connect_host, connect_port, err)
 172         host_session.connecting = true;
 173         local log = host_session.log or log;
 174
 175         if not err then
 176                 local IPs = {};
 177                 host_session.ip_hosts = IPs;
 178                 -- luacheck: ignore 231/handle4 231/handle6
 179                 local handle4, handle6;
 180                 local have_other_result = not(has_ipv4) or not(has_ipv6) or false;
 181
 182                 if has_ipv4 then
 183                         handle4 = host_session.resolver:lookup(function (reply, err)
 184                                 handle4 = nil;
 185
 186                                 if reply and reply[#reply] and reply[#reply].a then
 187                                         for _, ip in ipairs(reply) do
 188                                                 log("debug", "DNS reply for %s gives us %s", connect_host, ip.a);
 189                                                 IPs[#IPs+1] = new_ip(ip.a, "IPv4");
 190                                         end
 191                                 elseif err then
 192                                         log("debug", "Error in DNS lookup: %s", err);
 193                                 end
 194
 195                                 if have_other_result then
 196                                         if #IPs > 0 then
 197                                                 rfc6724_dest(host_session.ip_hosts, sources);
 198                                                 for i = 1, #IPs do
 199                                                         IPs[i] = {ip = IPs[i], port = connect_port};
 200                                                 end
 201                                                 host_session.ip_choice = 0;
 202                                                 s2sout.try_next_ip(host_session);
 203                                         else
 204                                                 log("debug", "DNS lookup failed to get a response for %s", connect_host);
 205                                                 host_session.ip_hosts = nil;
 206                                                 if not s2sout.attempt_connection(host_session, "name resolution failed") then -- Retry if we can
 207                                                         log("debug", "No other records to try for %s - destroying", host_session.to_host);
 208                                                         err = err and (": "..err) or "";
 209                                                         s2s_destroy_session(host_session, "DNS resolution failed"..err); -- End of the line, we can't
 210                                                 end
 211                                         end
 212                                 else
 213                                         have_other_result = true;
 214                                 end
 215                         end, connect_host, "A", "IN");
 216                 else
 217                         have_other_result = true;
 218                 end
 219
 220                 if has_ipv6 then
 221                         handle6 = host_session.resolver:lookup(function (reply, err)
 222                                 handle6 = nil;
 223
 224                                 if reply and reply[#reply] and reply[#reply].aaaa then
 225                                         for _, ip in ipairs(reply) do
 226                                                 log("debug", "DNS reply for %s gives us %s", connect_host, ip.aaaa);
 227                                                 IPs[#IPs+1] = new_ip(ip.aaaa, "IPv6");
 228                                         end
 229                                 elseif err then
 230                                         log("debug", "Error in DNS lookup: %s", err);
 231                                 end
 232
 233                                 if have_other_result then
 234                                         if #IPs > 0 then
 235                                                 rfc6724_dest(host_session.ip_hosts, sources);
 236                                                 for i = 1, #IPs do
 237                                                         IPs[i] = {ip = IPs[i], port = connect_port};
 238                                                 end
 239                                                 host_session.ip_choice = 0;
 240                                                 s2sout.try_next_ip(host_session);
 241                                         else
 242                                                 log("debug", "DNS lookup failed to get a response for %s", connect_host);
 243                                                 host_session.ip_hosts = nil;
 244                                                 if not s2sout.attempt_connection(host_session, "name resolution failed") then -- Retry if we can
 245                                                         log("debug", "No other records to try for %s - destroying", host_session.to_host);
 246                                                         err = err and (": "..err) or "";
 247                                                         s2s_destroy_session(host_session, "DNS resolution failed"..err); -- End of the line, we can't
 248                                                 end
 249                                         end
 250                                 else
 251                                         have_other_result = true;
 252                                 end
 253                         end, connect_host, "AAAA", "IN");
 254                 else
 255                         have_other_result = true;
 256                 end
 257                 return true;
 258         elseif host_session.ip_hosts and #host_session.ip_hosts > host_session.ip_choice then -- Not our first attempt, and we also have IPs left to try
 259                 s2sout.try_next_ip(host_session);
 260         else
 261                 log("debug", "Out of IP addresses, trying next SRV record (if any)");
 262                 host_session.ip_hosts = nil;
 263                 if not s2sout.attempt_connection(host_session, "out of IP addresses") then -- Retry if we can
 264                         log("debug", "No other records to try for %s - destroying", host_session.to_host);
 265                         err = err and (": "..err) or "";
 266                         s2s_destroy_session(host_session, "Connecting failed"..err); -- End of the line, we can't
 267                         return false;
 268                 end
 269         end
 270
 271         return true;
 272 end
 273
 274 function s2sout.make_connect(host_session, connect_host, connect_port)
 275         local log = host_session.log or log;
 276         log("debug", "Beginning new connection attempt to %s ([%s]:%d)", host_session.to_host, connect_host.addr, connect_port);
 277
 278         -- Reset secure flag in case this is another
 279         -- connection attempt after a failed STARTTLS
 280         host_session.secure = nil;
 281         host_session.encrypted = nil;
 282
 283         local conn, handler;
 284         local proto = connect_host.proto;
 285         if proto == "IPv4" then
 286                 conn, handler = socket.tcp();
 287         elseif proto == "IPv6" and socket.tcp6 then
 288                 conn, handler = socket.tcp6();
 289         else
 290                 handler = "Unsupported protocol: "..tostring(proto);
 291         end
 292
 293         if not conn then
 294                 log("warn", "Failed to create outgoing connection, system error: %s", handler);
 295                 return false, handler;
 296         end
 297
 298         conn:settimeout(0);
 299         local success, err = conn:connect(connect_host.addr, connect_port);
 300         if not success and err ~= "timeout" then
 301                 log("warn", "s2s connect() to %s (%s:%d) failed: %s", host_session.to_host, connect_host.addr, connect_port, err);
 302                 return false, err;
 303         end
 304
 305         conn = wrapclient(conn, connect_host.addr, connect_port, s2s_listener, default_mode);
 306         host_session.conn = conn;
 307
 308         -- Register this outgoing connection so that xmppserver_listener knows about it
 309         -- otherwise it will assume it is a new incoming connection
 310         s2s_listener.register_outgoing(conn, host_session);
 311
 312         log("debug", "Connection attempt in progress...");
 313         return true;
 314 end
 315
 316 module:hook_global("service-added", function (event)
 317         if event.name ~= "s2s" then return end
 318
 319         local s2s_sources = portmanager.get_active_services():get("s2s");
 320         if not s2s_sources then
 321                 module:log("warn", "s2s not listening on any ports, outgoing connections may fail");
 322                 return;
 323         end
 324         for source, _ in pairs(s2s_sources) do
 325                 if source == "*" or source == "0.0.0.0" then
 326                         for _, addr in ipairs(local_addresses("ipv4", true)) do
 327                                 sources[#sources + 1] = new_ip(addr, "IPv4");
 328                         end
 329                 elseif source == "::" then
 330                         for _, addr in ipairs(local_addresses("ipv6", true)) do
 331                                 sources[#sources + 1] = new_ip(addr, "IPv6");
 332                         end
 333                 else
 334                         sources[#sources + 1] = new_ip(source, (source:find(":") and "IPv6") or "IPv4");
 335                 end
 336         end
 337         for i = 1,#sources do
 338                 if sources[i].proto == "IPv6" then
 339                         has_ipv6 = true;
 340                 elseif sources[i].proto == "IPv4" then
 341                         has_ipv4 = true;
 342                 end
 343         end
 344         if not (has_ipv4 or has_ipv6)  then
 345                 module:log("warn", "No local IPv4 or IPv6 addresses detected, outgoing connections may fail");
 346         end
 347 end);
 348
 349 return s2sout;