X-Git-Url: https://git.enpas.org/?a=blobdiff_plain;f=core%2Fs2smanager.lua;h=e5fb699bacfc648ae8a3de869fe98dafd75a2061;hb=3ac2f4b559505492eb767ac4238b5d505ce5e6b0;hp=10c27746c19bd37edd16b381caea37a9135293a2;hpb=2e222d9f13dcde2c62265103795ea548aa8a9bed;p=prosody.git diff --git a/core/s2smanager.lua b/core/s2smanager.lua index 10c27746..e5fb699b 100644 --- a/core/s2smanager.lua +++ b/core/s2smanager.lua @@ -1,6 +1,6 @@ -- Prosody IM --- Copyright (C) 2008-2009 Matthew Wild --- Copyright (C) 2008-2009 Waqas Hussain +-- Copyright (C) 2008-2010 Matthew Wild +-- Copyright (C) 2008-2010 Waqas Hussain -- -- This project is MIT/X11 licensed. Please see the -- COPYING file in the source package for more information. @@ -16,11 +16,14 @@ local socket = require "socket"; local format = string.format; local t_insert, t_sort = table.insert, table.sort; local get_traceback = debug.traceback; -local tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber - = tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber; +local tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber, + setmetatable + = tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber, + setmetatable; local idna_to_ascii = require "util.encodings".idna.to_ascii; local connlisteners_get = require "net.connlisteners".get; +local initialize_filters = require "util.filters".initialize; local wrapclient = require "net.server".wrapclient; local modulemanager = require "core.modulemanager"; local st = require "stanza"; @@ -36,14 +39,14 @@ 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; -local dns_timeout = config.get("*", "core", "dns_timeout") or 60; +local dns_timeout = config.get("*", "core", "dns_timeout") or 15; local max_dns_depth = config.get("*", "core", "dns_max_depth") or 3; +dns.settimeout(dns_timeout); + incoming_s2s = {}; _G.prosody.incoming_s2s = incoming_s2s; local incoming_s2s = incoming_s2s; @@ -68,7 +71,7 @@ local function bounce_sendq(session, reason) for i, data in ipairs(sendq) do local reply = data[2]; local xmlns = reply.attr.xmlns; - if not xmlns or xmlns == "jabber:client" or xmlns == "jabber:server" then + if not xmlns then reply.attr.type = "error"; reply:tag("error", {type = "cancel"}) :tag("remote-server-not-found", {xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas"}):up(); @@ -91,7 +94,7 @@ function send_to_host(from_host, to_host, data) local host = hosts[from_host].s2sout[to_host]; if host then -- We have a connection to this host already - if host.type == "s2sout_unauthed" and (data.name ~= "db:verify" or not host.dialback_key) and ((not data.xmlns) or data.xmlns == "jabber:client" or data.xmlns == "jabber:server") then + if host.type == "s2sout_unauthed" and (data.name ~= "db:verify" or not host.dialback_key) then (host.log or log)("debug", "trying to send over unauthed s2sout to "..to_host); -- Queue stanza until we are able to send it @@ -137,7 +140,19 @@ 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 + local filter = initialize_filters(session); + session.sends2s = function (t) + log("debug", "sending: %s", t.top_tag and t:top_tag() or t:match("^([^>]*>?)")); + if t.name then + t = filter("stanzas/out", t); + end + if t then + t = filter("bytes/out", tostring(t)); + if t then + return w(conn, t); + end + end + end incoming_s2s[session] = true; add_task(connect_timeout, function () if session.conn ~= conn or @@ -145,16 +160,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; @@ -165,10 +181,14 @@ function new_outgoing(from_host, to_host) host_session.log = log; end - -- Kick the connection attempting machine - attempt_connection(host_session); + initialize_filters(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 @@ -182,7 +202,6 @@ function new_outgoing(from_host, to_host) buffer[#buffer+1] = data; log("debug", "Buffered item %d: %s", #buffer, tostring(data)); end - end return host_session; @@ -191,7 +210,7 @@ end function attempt_connection(host_session, err) local from_host, to_host = host_session.from_host, host_session.to_host; - local connect_host, connect_port = idna_to_ascii(to_host), 5269; + local connect_host, connect_port = to_host and idna_to_ascii(to_host), 5269; if not connect_host then return false; @@ -232,14 +251,6 @@ function attempt_connection(host_session, err) 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; @@ -292,18 +303,11 @@ function try_connect(host_session, connect_host, connect_port) 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); + (host_session.log or 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; @@ -323,17 +327,30 @@ function make_connect(host_session, connect_host, connect_port) end local cl = connlisteners_get("xmppserver"); - conn = wrapclient(conn, connect_host, connect_port, cl, cl.default_mode or 1, hosts[from_host].ssl_ctx, false ); + conn = wrapclient(conn, connect_host, connect_port, cl, cl.default_mode or 1 ); host_session.conn = conn; + local filter = initialize_filters(host_session); + local w, log = conn.write, host_session.log; + host_session.sends2s = function (t) + log("debug", "sending: %s", (t.top_tag and t:top_tag()) or t:match("^[^>]*>?")); + if t.name then + t = filter("stanzas/out", t); + end + if t then + t = filter("bytes/out", tostring(t)); + if t then + return w(conn, tostring(t)); + end + end + end + -- Register this outgoing connection so that xmppserver_listener knows about it -- otherwise it will assume it is a new incoming connection 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:open_stream(from_host, to_host); - conn.write(format([[]], from_host, to_host)); log("debug", "Connection attempt in progress..."); add_task(connect_timeout, function () if host_session.conn ~= conn or @@ -342,13 +359,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; @@ -359,12 +383,6 @@ function streamopened(session, attr) session.secure = true; 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 - if session.direction == "incoming" then -- Send a reply stream header session.to_host = attr.to and nameprep(attr.to); @@ -372,19 +390,31 @@ function streamopened(session, attr) session.streamid = uuid_gen(); (session.log or log)("debug", "incoming s2s received "); - 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; + if session.to_host then + if 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; + elseif hosts[session.to_host].disallow_s2s then + -- Attempting to connect to a host that disallows s2s + session:close({ + condition = "policy-violation"; + text = "Server-to-server communication is not allowed to this host"; + }); + return; + end end send(""); - send(stanza("stream:stream", { xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback', + 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 }); + hosts[session.to_host].events.fire_event("s2s-stream-features", { origin = session, features = features }); else (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"); end @@ -399,13 +429,13 @@ 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 log("debug", "Sending s2s send_buffer now..."); for i, data in ipairs(send_buffer) do - session.sends2s(data); + session.sends2s(tostring(data)); send_buffer[i] = nil; end end @@ -421,16 +451,12 @@ function streamopened(session, attr) end end end - session.notopen = nil; end function streamclosed(session) - (session.log or log)("debug", ""); - if session.sends2s then - session.sends2s(""); - end - session.notopen = true; + (session.log or log)("debug", "Received "); + session:close(); end function initiate_dialback(session) @@ -441,7 +467,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) @@ -504,7 +530,32 @@ function mark_connected(session) end end +local resting_session = { -- Resting, not dead + destroyed = true; + type = "s2s_destroyed"; + open_stream = function (session) + session.log("debug", "Attempt to open stream on resting session"); + end; + close = function (session) + session.log("debug", "Attempt to close already-closed session"); + end; + }; resting_session.__index = resting_session; + +function retire_session(session) + local log = session.log or log; + for k in pairs(session) do + if k ~= "trace" and k ~= "log" and k ~= "id" then + session[k] = nil; + end + end + + function session.send(data) log("debug", "Discarding data sent to resting session: %s", tostring(data)); end + function session.data(data) log("debug", "Discarding data received from resting session: %s", tostring(data)); end + return setmetatable(session, resting_session); +end + function destroy_session(session, reason) + if session.destroyed then return; end (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host)); if session.direction == "outgoing" then @@ -514,11 +565,7 @@ function destroy_session(session, reason) incoming_s2s[session] = nil; end - for k in pairs(session) do - if k ~= "trace" then - session[k] = nil; - end - end + retire_session(session); -- Clean session until it is GC'd end return _M;