0990a0aae327c14bde22500e1e57955295bd2626
[prosody.git] / core / s2smanager.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
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 add_task = require "util.timer".add_task;
15 local socket = require "socket";
16 local format = string.format;
17 local t_insert, t_sort = table.insert, table.sort;
18 local get_traceback = debug.traceback;
19 local tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber, setmetatable
20     = tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber, setmetatable;
21
22 local idna_to_ascii = require "util.encodings".idna.to_ascii;
23 local connlisteners_get = require "net.connlisteners".get;
24 local initialize_filters = require "util.filters".initialize;
25 local wrapclient = require "net.server".wrapclient;
26 local modulemanager = require "core.modulemanager";
27 local st = require "stanza";
28 local stanza = st.stanza;
29 local nameprep = require "util.encodings".stringprep.nameprep;
30 local cert_verify_identity = require "util.certverification".verify_identity;
31
32 local fire_event = prosody.events.fire_event;
33 local uuid_gen = require "util.uuid".generate;
34
35 local logger_init = require "util.logger".init;
36
37 local log = logger_init("s2smanager");
38
39 local sha256_hash = require "util.hashes".sha256;
40
41 local adns, dns = require "net.adns", require "net.dns";
42 local config = require "core.configmanager";
43 local connect_timeout = config.get("*", "core", "s2s_timeout") or 60;
44 local dns_timeout = config.get("*", "core", "dns_timeout") or 15;
45 local max_dns_depth = config.get("*", "core", "dns_max_depth") or 3;
46
47 dns.settimeout(dns_timeout);
48
49 local prosody = _G.prosody;
50 incoming_s2s = {};
51 prosody.incoming_s2s = incoming_s2s;
52 local incoming_s2s = incoming_s2s;
53
54 module "s2smanager"
55
56 function compare_srv_priorities(a,b)
57         return a.priority < b.priority or (a.priority == b.priority and a.weight > b.weight);
58 end
59
60 local bouncy_stanzas = { message = true, presence = true, iq = true };
61 local function bounce_sendq(session, reason)
62         local sendq = session.sendq;
63         if sendq then
64                 session.log("info", "sending error replies for "..#sendq.." queued stanzas because of failed outgoing connection to "..tostring(session.to_host));
65                 local dummy = {
66                         type = "s2sin";
67                         send = function(s)
68                                 (session.log or log)("error", "Replying to to an s2s error reply, please report this! Traceback: %s", get_traceback());
69                         end;
70                         dummy = true;
71                 };
72                 for i, data in ipairs(sendq) do
73                         local reply = data[2];
74                         local xmlns = reply.attr.xmlns;
75                         if not(xmlns) and bouncy_stanzas[reply.name] then
76                                 reply.attr.type = "error";
77                                 reply:tag("error", {type = "cancel"})
78                                         :tag("remote-server-not-found", {xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas"}):up();
79                                 if reason then
80                                         reply:tag("text", {xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas"}):text("Connection failed: "..reason):up();
81                                 end
82                                 core_process_stanza(dummy, reply);
83                         end
84                         sendq[i] = nil;
85                 end
86                 session.sendq = nil;
87         end
88 end
89
90 function send_to_host(from_host, to_host, data)
91         if not hosts[from_host] then
92                 log("warn", "Attempt to send stanza from %s - a host we don't serve", from_host);
93                 return false;
94         end
95         local host = hosts[from_host].s2sout[to_host];
96         if host then
97                 -- We have a connection to this host already
98                 if host.type == "s2sout_unauthed" and (data.name ~= "db:verify" or not host.dialback_key) then
99                         (host.log or log)("debug", "trying to send over unauthed s2sout to "..to_host);
100                         
101                         -- Queue stanza until we are able to send it
102                         if host.sendq then t_insert(host.sendq, {tostring(data), st.reply(data)});
103                         else host.sendq = { {tostring(data), st.reply(data)} }; end
104                         host.log("debug", "stanza [%s] queued ", data.name);
105                 elseif host.type == "local" or host.type == "component" then
106                         log("error", "Trying to send a stanza to ourselves??")
107                         log("error", "Traceback: %s", get_traceback());
108                         log("error", "Stanza: %s", tostring(data));
109                 else
110                         (host.log or log)("debug", "going to send stanza to "..to_host.." from "..from_host);
111                         -- FIXME
112                         if host.from_host ~= from_host then
113                                 log("error", "WARNING! This might, possibly, be a bug, but it might not...");
114                                 log("error", "We are going to send from %s instead of %s", tostring(host.from_host), tostring(from_host));
115                         end
116                         host.sends2s(data);
117                         host.log("debug", "stanza sent over "..host.type);
118                 end
119         else
120                 log("debug", "opening a new outgoing connection for this stanza");
121                 local host_session = new_outgoing(from_host, to_host);
122
123                 -- Store in buffer
124                 host_session.sendq = { {tostring(data), st.reply(data)} };
125                 log("debug", "stanza [%s] queued until connection complete", tostring(data.name));
126                 if (not host_session.connecting) and (not host_session.conn) then
127                         log("warn", "Connection to %s failed already, destroying session...", to_host);
128                         destroy_session(host_session);
129                 end
130         end
131 end
132
133 local open_sessions = 0;
134
135 function new_incoming(conn)
136         local session = { conn = conn, type = "s2sin_unauthed", direction = "incoming", hosts = {} };
137         if true then
138                 session.trace = newproxy(true);
139                 getmetatable(session.trace).__gc = function () open_sessions = open_sessions - 1; end;
140         end
141         open_sessions = open_sessions + 1;
142         local w, log = conn.write, logger_init("s2sin"..tostring(conn):match("[a-f0-9]+$"));
143         session.log = log;
144         local filter = initialize_filters(session);
145         session.sends2s = function (t)
146                 log("debug", "sending: %s", t.top_tag and t:top_tag() or t:match("^([^>]*>?)"));
147                 if t.name then
148                         t = filter("stanzas/out", t);
149                 end
150                 if t then
151                         t = filter("bytes/out", tostring(t));
152                         if t then
153                                 return w(conn, t);
154                         end
155                 end
156         end
157         incoming_s2s[session] = true;
158         add_task(connect_timeout, function ()
159                 if session.conn ~= conn or
160                    session.type == "s2sin" then
161                         return; -- Ok, we're connect[ed|ing]
162                 end
163                 -- Not connected, need to close session and clean up
164                 (session.log or log)("warn", "Destroying incomplete session %s->%s due to inactivity",
165                     session.from_host or "(unknown)", session.to_host or "(unknown)");
166                 session:close("connection-timeout");
167         end);
168         return session;
169 end
170
171 function new_outgoing(from_host, to_host, connect)
172                 local host_session = { to_host = to_host, from_host = from_host, host = from_host,
173                                        notopen = true, type = "s2sout_unauthed", direction = "outgoing",
174                                        open_stream = session_open_stream };
175                 
176                 hosts[from_host].s2sout[to_host] = host_session;
177                 
178                 local log;
179                 do
180                         local conn_name = "s2sout"..tostring(host_session):match("[a-f0-9]*$");
181                         log = logger_init(conn_name);
182                         host_session.log = log;
183                 end
184                 
185                 initialize_filters(host_session);
186                 
187                 if connect ~= false then
188                         -- Kick the connection attempting machine into life
189                         attempt_connection(host_session);
190                 end
191                 
192                 if not host_session.sends2s then
193                         -- A sends2s which buffers data (until the stream is opened)
194                         -- note that data in this buffer will be sent before the stream is authed
195                         -- and will not be ack'd in any way, successful or otherwise
196                         local buffer;
197                         function host_session.sends2s(data)
198                                 if not buffer then
199                                         buffer = {};
200                                         host_session.send_buffer = buffer;
201                                 end
202                                 log("debug", "Buffering data on unconnected s2sout to %s", to_host);
203                                 buffer[#buffer+1] = data;
204                                 log("debug", "Buffered item %d: %s", #buffer, tostring(data));
205                         end
206                 end
207
208                 return host_session;
209 end
210
211
212 function attempt_connection(host_session, err)
213         local from_host, to_host = host_session.from_host, host_session.to_host;
214         local connect_host, connect_port = to_host and idna_to_ascii(to_host), 5269;
215         
216         if not connect_host then
217                 return false;
218         end
219         
220         if not err then -- This is our first attempt
221                 log("debug", "First attempt to connect to %s, starting with SRV lookup...", to_host);
222                 host_session.connecting = true;
223                 local handle;
224                 handle = adns.lookup(function (answer)
225                         handle = nil;
226                         host_session.connecting = nil;
227                         if answer then
228                                 log("debug", to_host.." has SRV records, handling...");
229                                 local srv_hosts = {};
230                                 host_session.srv_hosts = srv_hosts;
231                                 for _, record in ipairs(answer) do
232                                         t_insert(srv_hosts, record.srv);
233                                 end
234                                 t_sort(srv_hosts, compare_srv_priorities);
235                                 
236                                 local srv_choice = srv_hosts[1];
237                                 host_session.srv_choice = 1;
238                                 if srv_choice then
239                                         connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
240                                         log("debug", "Best record found, will connect to %s:%d", connect_host, connect_port);
241                                 end
242                         else
243                                 log("debug", to_host.." has no SRV records, falling back to A");
244                         end
245                         -- Try with SRV, or just the plain hostname if no SRV
246                         local ok, err = try_connect(host_session, connect_host, connect_port);
247                         if not ok then
248                                 if not attempt_connection(host_session, err) then
249                                         -- No more attempts will be made
250                                         destroy_session(host_session, err);
251                                 end
252                         end
253                 end, "_xmpp-server._tcp."..connect_host..".", "SRV");
254                 
255                 return true; -- Attempt in progress
256         elseif host_session.srv_hosts and #host_session.srv_hosts > host_session.srv_choice then -- Not our first attempt, and we also have SRV
257                 host_session.srv_choice = host_session.srv_choice + 1;
258                 local srv_choice = host_session.srv_hosts[host_session.srv_choice];
259                 connect_host, connect_port = srv_choice.target or to_host, srv_choice.port or connect_port;
260                 host_session.log("info", "Connection failed (%s). Attempt #%d: This time to %s:%d", tostring(err), host_session.srv_choice, connect_host, connect_port);
261         else
262                 host_session.log("info", "Out of connection options, can't connect to %s", tostring(host_session.to_host));
263                 -- We're out of options
264                 return false;
265         end
266         
267         if not (connect_host and connect_port) then
268                 -- Likely we couldn't resolve DNS
269                 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));
270                 return false;
271         end
272         
273         return try_connect(host_session, connect_host, connect_port);
274 end
275
276 function try_connect(host_session, connect_host, connect_port)
277         host_session.connecting = true;
278         local handle;
279         handle = adns.lookup(function (reply)
280                 handle = nil;
281                 host_session.connecting = nil;
282                 
283                 -- COMPAT: This is a compromise for all you CNAME-(ab)users :)
284                 if not (reply and reply[#reply] and reply[#reply].a) then
285                         local count = max_dns_depth;
286                         reply = dns.peek(connect_host, "CNAME", "IN");
287                         while count > 0 and reply and reply[#reply] and not reply[#reply].a and reply[#reply].cname do
288                                 log("debug", "Looking up %s (DNS depth is %d)", tostring(reply[#reply].cname), count);
289                                 reply = dns.peek(reply[#reply].cname, "A", "IN") or dns.peek(reply[#reply].cname, "CNAME", "IN");
290                                 count = count - 1;
291                         end
292                 end
293                 -- end of CNAME resolving
294                 
295                 if reply and reply[#reply] and reply[#reply].a then
296                         log("debug", "DNS reply for %s gives us %s", connect_host, reply[#reply].a);
297                         return make_connect(host_session, reply[#reply].a, connect_port);
298                 else
299                         log("debug", "DNS lookup failed to get a response for %s", connect_host);
300                         if not attempt_connection(host_session, "name resolution failed") then -- Retry if we can
301                                 log("debug", "No other records to try for %s - destroying", host_session.to_host);
302                                 destroy_session(host_session, "DNS resolution failed"); -- End of the line, we can't
303                         end
304                 end
305         end, connect_host, "A", "IN");
306
307         return true;
308 end
309
310 function make_connect(host_session, connect_host, connect_port)
311         (host_session.log or log)("info", "Beginning new connection attempt to %s (%s:%d)", host_session.to_host, connect_host, connect_port);
312         -- Ok, we're going to try to connect
313         
314         local from_host, to_host = host_session.from_host, host_session.to_host;
315         
316         local conn, handler = socket.tcp()
317         
318         if not conn then
319                 log("warn", "Failed to create outgoing connection, system error: %s", handler);
320                 return false, handler;
321         end
322
323         conn:settimeout(0);
324         local success, err = conn:connect(connect_host, connect_port);
325         if not success and err ~= "timeout" then
326                 log("warn", "s2s connect() to %s (%s:%d) failed: %s", host_session.to_host, connect_host, connect_port, err);
327                 return false, err;
328         end
329         
330         local cl = connlisteners_get("xmppserver");
331         conn = wrapclient(conn, connect_host, connect_port, cl, cl.default_mode or 1 );
332         host_session.conn = conn;
333         
334         local filter = initialize_filters(host_session);
335         local w, log = conn.write, host_session.log;
336         host_session.sends2s = function (t)
337                 log("debug", "sending: %s", (t.top_tag and t:top_tag()) or t:match("^[^>]*>?"));
338                 if t.name then
339                         t = filter("stanzas/out", t);
340                 end
341                 if t then
342                         t = filter("bytes/out", tostring(t));
343                         if t then
344                                 return w(conn, tostring(t));
345                         end
346                 end
347         end
348         
349         -- Register this outgoing connection so that xmppserver_listener knows about it
350         -- otherwise it will assume it is a new incoming connection
351         cl.register_outgoing(conn, host_session);
352         
353         host_session:open_stream(from_host, to_host);
354         
355         log("debug", "Connection attempt in progress...");
356         add_task(connect_timeout, function ()
357                 if host_session.conn ~= conn or
358                    host_session.type == "s2sout" or
359                    host_session.connecting then
360                         return; -- Ok, we're connect[ed|ing]
361                 end
362                 -- Not connected, need to close session and clean up
363                 (host_session.log or log)("warn", "Destroying incomplete session %s->%s due to inactivity",
364                     host_session.from_host or "(unknown)", host_session.to_host or "(unknown)");
365                 host_session:close("connection-timeout");
366         end);
367         return true;
368 end
369
370 function session_open_stream(session, from, to)
371         session.sends2s(st.stanza("stream:stream", {
372                 xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback',
373                 ["xmlns:stream"]='http://etherx.jabber.org/streams',
374                 from=from, to=to, version='1.0', ["xml:lang"]='en'}):top_tag());
375 end
376
377 local function check_cert_status(session)
378         local conn = session.conn:socket()
379         local cert
380         if conn.getpeercertificate then
381                 cert = conn:getpeercertificate()
382         end
383
384         if cert then
385                 local chain_valid, err = conn:getpeerchainvalid()
386                 if not chain_valid then
387                         session.cert_chain_status = "invalid";
388                         (session.log or log)("debug", "certificate chain validation result: %s", err);
389                 else
390                         session.cert_chain_status = "valid";
391
392                         local host = session.direction == "incoming" and session.from_host or session.to_host
393
394                         -- We'll go ahead and verify the asserted identity if the
395                         -- connecting server specified one.
396                         if host then
397                                 if cert_verify_identity(host, "xmpp-server", cert) then
398                                         session.cert_identity_status = "valid"
399                                 else
400                                         session.cert_identity_status = "invalid"
401                                 end
402                         end
403                 end
404         end
405 end
406
407 function streamopened(session, attr)
408         local send = session.sends2s;
409         
410         -- TODO: #29: SASL/TLS on s2s streams
411         session.version = tonumber(attr.version) or 0;
412         
413         -- TODO: Rename session.secure to session.encrypted
414         if session.secure == false then
415                 session.secure = true;
416         end
417
418         if session.direction == "incoming" then
419                 -- Send a reply stream header
420                 session.to_host = attr.to and nameprep(attr.to);
421                 session.from_host = attr.from and nameprep(attr.from);
422         
423                 session.streamid = uuid_gen();
424                 (session.log or log)("debug", "incoming s2s received <stream:stream>");
425                 if session.to_host then
426                         if not hosts[session.to_host] then
427                                 -- Attempting to connect to a host we don't serve
428                                 session:close({
429                                         condition = "host-unknown";
430                                         text = "This host does not serve "..session.to_host
431                                 });
432                                 return;
433                         elseif hosts[session.to_host].disallow_s2s then
434                                 -- Attempting to connect to a host that disallows s2s
435                                 session:close({
436                                         condition = "policy-violation";
437                                         text = "Server-to-server communication is not allowed to this host";
438                                 });
439                                 return;
440                         end
441                 end
442
443                 if session.secure and not session.cert_chain_status then check_cert_status(session); end
444
445                 send("<?xml version='1.0'?>");
446                 send(stanza("stream:stream", { xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback',
447                                 ["xmlns:stream"]='http://etherx.jabber.org/streams', id=session.streamid, from=session.to_host, to=session.from_host, version=(session.version > 0 and "1.0" or nil) }):top_tag());
448                 if session.version >= 1.0 then
449                         local features = st.stanza("stream:features");
450                         
451                         if session.to_host then
452                                 hosts[session.to_host].events.fire_event("s2s-stream-features", { origin = session, features = features });
453                         else
454                                 (session.log or log)("warn", "No 'to' on stream header from %s means we can't offer any features", session.from_host or "unknown host");
455                         end
456                         
457                         log("debug", "Sending stream features: %s", tostring(features));
458                         send(features);
459                 end
460         elseif session.direction == "outgoing" then
461                 -- If we are just using the connection for verifying dialback keys, we won't try and auth it
462                 if not attr.id then error("stream response did not give us a streamid!!!"); end
463                 session.streamid = attr.id;
464
465                 if session.secure and not session.cert_chain_status then check_cert_status(session); end
466
467                 -- Send unauthed buffer
468                 -- (stanzas which are fine to send before dialback)
469                 -- Note that this is *not* the stanza queue (which
470                 -- we can only send if auth succeeds) :)
471                 local send_buffer = session.send_buffer;
472                 if send_buffer and #send_buffer > 0 then
473                         log("debug", "Sending s2s send_buffer now...");
474                         for i, data in ipairs(send_buffer) do
475                                 session.sends2s(tostring(data));
476                                 send_buffer[i] = nil;
477                         end
478                 end
479                 session.send_buffer = nil;
480         
481                 -- If server is pre-1.0, don't wait for features, just do dialback
482                 if session.version < 1.0 then
483                         if not session.dialback_verifying then
484                                 log("debug", "Initiating dialback...");
485                                 initiate_dialback(session);
486                         else
487                                 mark_connected(session);
488                         end
489                 end
490         end
491         session.notopen = nil;
492 end
493
494 function streamclosed(session)
495         (session.log or log)("debug", "Received </stream:stream>");
496         session:close();
497 end
498
499 function initiate_dialback(session)
500         -- generate dialback key
501         session.dialback_key = generate_dialback(session.streamid, session.to_host, session.from_host);
502         session.sends2s(format("<db:result from='%s' to='%s'>%s</db:result>", session.from_host, session.to_host, session.dialback_key));
503         session.log("info", "sent dialback key on outgoing s2s stream");
504 end
505
506 function generate_dialback(id, to, from)
507         return sha256_hash(id..to..from..hosts[from].dialback_secret, true);
508 end
509
510 function verify_dialback(id, to, from, key)
511         return key == generate_dialback(id, to, from);
512 end
513
514 function make_authenticated(session, host)
515         if not session.secure then
516                 local local_host = session.direction == "incoming" and session.to_host or session.from_host;
517                 if config.get(local_host, "core", "s2s_require_encryption") then
518                         session:close({
519                                 condition = "policy-violation",
520                                 text = "Encrypted server-to-server communication is required but was not "
521                                        ..((session.direction == "outgoing" and "offered") or "used")
522                         });
523                 end
524         end
525         if session.type == "s2sout_unauthed" then
526                 session.type = "s2sout";
527         elseif session.type == "s2sin_unauthed" then
528                 session.type = "s2sin";
529                 if host then
530                         if not session.hosts[host] then session.hosts[host] = {}; end
531                         session.hosts[host].authed = true;
532                 end
533         elseif session.type == "s2sin" and host then
534                 if not session.hosts[host] then session.hosts[host] = {}; end
535                 session.hosts[host].authed = true;
536         else
537                 return false;
538         end
539         session.log("debug", "connection %s->%s is now authenticated", session.from_host or "(unknown)", session.to_host or "(unknown)");
540         
541         mark_connected(session);
542         
543         return true;
544 end
545
546 -- Stream is authorised, and ready for normal stanzas
547 function mark_connected(session)
548         local sendq, send = session.sendq, session.sends2s;
549         
550         local from, to = session.from_host, session.to_host;
551         
552         session.log("info", session.direction.." s2s connection "..from.."->"..to.." complete");
553         
554         local send_to_host = send_to_host;
555         function session.send(data) send_to_host(to, from, data); end
556         
557         local event_data = { session = session };
558         if session.type == "s2sout" then
559                 prosody.events.fire_event("s2sout-established", event_data);
560                 hosts[session.from_host].events.fire_event("s2sout-established", event_data);
561         else
562                 prosody.events.fire_event("s2sin-established", event_data);
563                 hosts[session.to_host].events.fire_event("s2sin-established", event_data);
564         end
565         
566         if session.direction == "outgoing" then
567                 if sendq then
568                         session.log("debug", "sending "..#sendq.." queued stanzas across new outgoing connection to "..session.to_host);
569                         for i, data in ipairs(sendq) do
570                                 send(data[1]);
571                                 sendq[i] = nil;
572                         end
573                         session.sendq = nil;
574                 end
575                 
576                 session.srv_hosts = nil;
577         end
578 end
579
580 local resting_session = { -- Resting, not dead
581                 destroyed = true;
582                 type = "s2s_destroyed";
583                 open_stream = function (session)
584                         session.log("debug", "Attempt to open stream on resting session");
585                 end;
586                 close = function (session)
587                         session.log("debug", "Attempt to close already-closed session");
588                 end;
589                 filter = function (type, data) return data; end;
590         }; resting_session.__index = resting_session;
591
592 function retire_session(session)
593         local log = session.log or log;
594         for k in pairs(session) do
595                 if k ~= "trace" and k ~= "log" and k ~= "id" then
596                         session[k] = nil;
597                 end
598         end
599
600         function session.send(data) log("debug", "Discarding data sent to resting session: %s", tostring(data)); end
601         function session.data(data) log("debug", "Discarding data received from resting session: %s", tostring(data)); end
602         return setmetatable(session, resting_session);
603 end
604
605 function destroy_session(session, reason)
606         if session.destroyed then return; end
607         (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host));
608         
609         if session.direction == "outgoing" then
610                 hosts[session.from_host].s2sout[session.to_host] = nil;
611                 bounce_sendq(session, reason);
612         elseif session.direction == "incoming" then
613                 incoming_s2s[session] = nil;
614         end
615         
616         local event_data = { session = session, reason = reason };
617         if session.type == "s2sout" then
618                 prosody.events.fire_event("s2sout-destroyed", event_data);
619                 if hosts[session.from_host] then
620                         hosts[session.from_host].events.fire_event("s2sout-destroyed", event_data);
621                 end
622         elseif session.type == "s2sin" then
623                 prosody.events.fire_event("s2sin-destroyed", event_data);
624                 if hosts[session.to_host] then
625                         hosts[session.to_host].events.fire_event("s2sin-destroyed", event_data);
626                 end
627         end
628         
629         retire_session(session); -- Clean session until it is GC'd
630 end
631
632 return _M;