X-Git-Url: https://git.enpas.org/?a=blobdiff_plain;f=core%2Fs2smanager.lua;h=af5e91e3585705eb946b68a2ca3759a53848232c;hb=9a42b61e1f6522baf13a72f10d4c9fc24d1325f1;hp=4c61eaa3fc972bbc77e0302926d3547f976ee2c5;hpb=f5b1bba779693b7f501b5ec5c6c91ea029e082a0;p=prosody.git diff --git a/core/s2smanager.lua b/core/s2smanager.lua index 4c61eaa3..af5e91e3 100644 --- a/core/s2smanager.lua +++ b/core/s2smanager.lua @@ -36,8 +36,6 @@ local log = logger_init("s2smanager"); local sha256_hash = require "util.hashes".sha256; -local dialback_secret = uuid_gen(); - local adns, dns = require "net.adns", require "net.dns"; local config = require "core.configmanager"; local connect_timeout = config.get("*", "core", "s2s_timeout") or 60; @@ -52,7 +50,7 @@ module "s2smanager" local function compare_srv_priorities(a,b) return a.priority < b.priority or a.weight < b.weight; end -local function bounce_sendq(session) +local function bounce_sendq(session, reason) local sendq = session.sendq; if sendq then session.log("info", "sending error replies for "..#sendq.." queued stanzas because of failed outgoing connection to "..tostring(session.to_host)); @@ -70,6 +68,9 @@ local function bounce_sendq(session) 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("Connection failed: "..reason):up(); + end core_process_stanza(dummy, reply); end sendq[i] = nil; @@ -79,6 +80,10 @@ local function bounce_sendq(session) end function send_to_host(from_host, to_host, data) + if not hosts[from_host] then + log("warn", "Attempt to send stanza from %s - a host we don't serve", from_host); + return false; + end local host = hosts[from_host].s2sout[to_host]; if host then -- We have a connection to this host already @@ -128,7 +133,7 @@ function new_incoming(conn) 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 + session.sends2s = function (t) log("debug", "sending: %s", t.top_tag and t:top_tag() or t:match("^([^>]*>?)")); w(conn, tostring(t)); end incoming_s2s[session] = true; add_task(connect_timeout, function () if session.conn ~= conn or @@ -136,16 +141,17 @@ function new_incoming(conn) 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.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 -function new_outgoing(from_host, to_host) - local host_session = { to_host = to_host, from_host = from_host, host = from_host, - notopen = true, type = "s2sout_unauthed", direction = "outgoing" }; +function new_outgoing(from_host, to_host, connect) + local host_session = { to_host = to_host, from_host = from_host, host = from_host, + notopen = true, type = "s2sout_unauthed", direction = "outgoing", + open_stream = session_open_stream }; hosts[from_host].s2sout[to_host] = host_session; @@ -156,10 +162,12 @@ function new_outgoing(from_host, to_host) host_session.log = log; end - -- Kick the connection attempting machine - attempt_connection(host_session); + if connect ~= false then + -- Kick the connection attempting machine into life + attempt_connection(host_session); + end - if not host_session.sends2s then + 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 @@ -218,7 +226,7 @@ function attempt_connection(host_session, err) if not ok then if not attempt_connection(host_session, err) then -- No more attempts will be made - destroy_session(host_session); + destroy_session(host_session, err); end end end, "_xmpp-server._tcp."..connect_host..".", "SRV"); @@ -278,7 +286,7 @@ function try_connect(host_session, connect_host, connect_port) 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 + destroy_session(host_session, "DNS resolution failed"); -- End of the line, we can't end end end, connect_host, "A", "IN"); @@ -289,7 +297,7 @@ function try_connect(host_session, connect_host, connect_port) adns.cancel(handle, true); end end); - + return true; end @@ -317,9 +325,10 @@ function make_connect(host_session, connect_host, connect_port) cl.register_outgoing(conn, host_session); local w, log = conn.write, host_session.log; - host_session.sends2s = function (t) log("debug", "sending: %s", tostring(t)); w(tostring(t)); end + host_session.sends2s = function (t) log("debug", "sending: %s", (t.top_tag and t:top_tag()) or t:match("^[^>]*>?")); w(conn, tostring(t)); end + + host_session:open_stream(); - conn.write(format([[]], from_host, to_host)); log("debug", "Connection attempt in progress..."); add_task(connect_timeout, function () if host_session.conn ~= conn or @@ -328,13 +337,20 @@ function make_connect(host_session, connect_host, connect_port) 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.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 +function session_open_stream(session, from, to) + session.sends2s(st.stanza("stream:stream", { + xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback', + ["xmlns:stream"]='http://etherx.jabber.org/streams', + from=from, to=to, version='1.0', ["xml:lang"]='en'}):top_tag()); +end + function streamopened(session, attr) local send = session.sends2s; @@ -346,7 +362,6 @@ function streamopened(session, attr) end if session.version >= 1.0 and not (attr.to and attr.from) then - (session.log or log)("warn", "Remote of stream "..(session.from_host or "(unknown)").."->"..(session.to_host or "(unknown)") .." failed to specify to (%s) and/or from (%s) hostname as per RFC", tostring(attr.to), tostring(attr.from)); end @@ -358,17 +373,17 @@ function streamopened(session, attr) session.streamid = uuid_gen(); (session.log or log)("debug", "incoming s2s received "); - send(""); - send(stanza("stream:stream", { xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback', - ["xmlns:stream"]='http://etherx.jabber.org/streams', id=session.streamid, from=session.to_host, version=(session.version > 0 and "1.0" or nil) }):top_tag()); if session.to_host and not hosts[session.to_host] then -- Attempting to connect to a host we don't serve session:close({ condition = "host-unknown"; text = "This host does not serve "..session.to_host }); return; end + send(""); + send(stanza("stream:stream", { xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback', + ["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()); if session.version >= 1.0 then local features = st.stanza("stream:features"); - + if session.to_host then hosts[session.to_host].events.fire_event("s2s-stream-features", { session = session, features = features }); else @@ -385,7 +400,7 @@ function streamopened(session, attr) -- Send unauthed buffer -- (stanzas which are fine to send before dialback) - -- Note that this is *not* the stanza queue (which + -- 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 @@ -407,7 +422,6 @@ function streamopened(session, attr) end end end - session.notopen = nil; end @@ -427,7 +441,7 @@ function initiate_dialback(session) end function generate_dialback(id, to, from) - return sha256_hash(id..to..from..dialback_secret, true); + return sha256_hash(id..to..from..hosts[from].dialback_secret, true); end function verify_dialback(id, to, from, key) @@ -480,12 +494,12 @@ function mark_connected(session) end end -function destroy_session(session) +function destroy_session(session, reason) (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host)); if session.direction == "outgoing" then hosts[session.from_host].s2sout[session.to_host] = nil; - bounce_sendq(session); + bounce_sendq(session, reason); elseif session.direction == "incoming" then incoming_s2s[session] = nil; end