X-Git-Url: https://git.enpas.org/?a=blobdiff_plain;f=core%2Fs2smanager.lua;h=7e6f8135791df3a6d0734fbace1ce92898608223;hb=83b669b80dc14c47edae26a1f0e180565ee1f203;hp=053dbbe773d290ff227587bbaba758de777715a9;hpb=4643eae0bf41b461c9337bfb66f9ac80f7e84f11;p=prosody.git diff --git a/core/s2smanager.lua b/core/s2smanager.lua index 053dbbe7..7e6f8135 100644 --- a/core/s2smanager.lua +++ b/core/s2smanager.lua @@ -16,20 +16,20 @@ 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, - setmetatable - = tostring, pairs, ipairs, getmetatable, newproxy, error, tonumber, - setmetatable; +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"; local stanza = st.stanza; local nameprep = require "util.encodings".stringprep.nameprep; +local cert_verify_identity = require "util.x509".verify_identity; -local fire_event = require "core.eventmanager".fire_event; +local fire_event = prosody.events.fire_event; local uuid_gen = require "util.uuid".generate; local logger_init = require "util.logger".init; @@ -41,11 +41,14 @@ local sha256_hash = require "util.hashes".sha256; 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); + +local prosody = _G.prosody; incoming_s2s = {}; -_G.prosody.incoming_s2s = incoming_s2s; +prosody.incoming_s2s = incoming_s2s; local incoming_s2s = incoming_s2s; module "s2smanager" @@ -54,6 +57,7 @@ function compare_srv_priorities(a,b) return a.priority < b.priority or (a.priority == b.priority and a.weight > b.weight); end +local bouncy_stanzas = { message = true, presence = true, iq = true }; local function bounce_sendq(session, reason) local sendq = session.sendq; if sendq then @@ -68,7 +72,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) and bouncy_stanzas[reply.name] 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 +95,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 @@ -102,6 +106,7 @@ function send_to_host(from_host, to_host, data) log("error", "Trying to send a stanza to ourselves??") log("error", "Traceback: %s", get_traceback()); log("error", "Stanza: %s", tostring(data)); + return false; else (host.log or log)("debug", "going to send stanza to "..to_host.." from "..from_host); -- FIXME @@ -121,9 +126,14 @@ function send_to_host(from_host, to_host, data) 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); + if not destroy_session(host_session, "Connection failed") then + -- Already destroyed, we need to bounce our stanza + bounce_sendq(host_session, host_session.destruction_reason); + end + return false; end end + return true; end local open_sessions = 0; @@ -137,7 +147,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", t.top_tag and t:top_tag() or t:match("^([^>]*>?)")); w(conn, 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 @@ -159,6 +181,8 @@ function new_outgoing(from_host, to_host, connect) hosts[from_host].s2sout[to_host] = host_session; + host_session.close = destroy_session; -- This gets replaced by xmppserver_listener later + local log; do local conn_name = "s2sout"..tostring(host_session):match("[a-f0-9]*$"); @@ -166,9 +190,15 @@ function new_outgoing(from_host, to_host, connect) host_session.log = log; end + initialize_filters(host_session); + if connect ~= false then -- Kick the connection attempting machine into life - attempt_connection(host_session); + if not attempt_connection(host_session) then + -- Intentionally not returning here, the + -- session is needed, connected or not + destroy_session(host_session); + end end if not host_session.sends2s then @@ -193,7 +223,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; @@ -234,14 +264,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; @@ -266,7 +288,7 @@ end function try_connect(host_session, connect_host, connect_port) host_session.connecting = true; local handle; - handle = adns.lookup(function (reply) + handle = adns.lookup(function (reply, err) handle = nil; host_session.connecting = nil; @@ -284,23 +306,23 @@ function try_connect(host_session, connect_host, connect_port) 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); + local ok, err = make_connect(host_session, reply[#reply].a, connect_port); + if not ok then + if not attempt_connection(host_session, err or "closed") then + err = err and (": "..err) or ""; + destroy_session(host_session, "Connection failed"..err); + end + end 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, "DNS resolution failed"); -- End of the line, we can't + err = err and (": "..err) or ""; + destroy_session(host_session, "DNS resolution failed"..err); -- 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 @@ -310,7 +332,7 @@ function make_connect(host_session, connect_host, connect_port) local from_host, to_host = host_session.from_host, host_session.to_host; - local conn, handler = socket.tcp() + local conn, handler = socket.tcp(); if not conn then log("warn", "Failed to create outgoing connection, system error: %s", handler); @@ -328,13 +350,25 @@ function make_connect(host_session, connect_host, connect_port) 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", (t.top_tag and t:top_tag()) or t:match("^[^>]*>?")); w(conn, tostring(t)); end - host_session:open_stream(from_host, to_host); log("debug", "Connection attempt in progress..."); @@ -359,16 +393,47 @@ function session_open_stream(session, from, to) from=from, to=to, version='1.0', ["xml:lang"]='en'}):top_tag()); end +local function check_cert_status(session) + local conn = session.conn:socket() + local cert + if conn.getpeercertificate then + cert = conn:getpeercertificate() + end + + if cert then + local chain_valid, err = conn:getpeerchainvalid() + if not chain_valid then + session.cert_chain_status = "invalid"; + (session.log or log)("debug", "certificate chain validation result: %s", err); + else + session.cert_chain_status = "valid"; + + local host = session.direction == "incoming" and session.from_host or session.to_host + + -- We'll go ahead and verify the asserted identity if the + -- connecting server specified one. + if host then + if cert_verify_identity(host, "xmpp-server", cert) then + session.cert_identity_status = "valid" + else + session.cert_identity_status = "invalid" + end + end + end + end +end + function streamopened(session, attr) local send = session.sends2s; -- TODO: #29: SASL/TLS on s2s streams session.version = tonumber(attr.version) or 0; + -- TODO: Rename session.secure to session.encrypted if session.secure == false then session.secure = true; end - + if session.direction == "incoming" then -- Send a reply stream header session.to_host = attr.to and nameprep(attr.to); @@ -376,11 +441,26 @@ 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 + + if session.secure and not session.cert_chain_status then check_cert_status(session); 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()); @@ -400,7 +480,9 @@ function streamopened(session, attr) -- If we are just using the connection for verifying dialback keys, we won't try and auth it if not attr.id then error("stream response did not give us a streamid!!!"); end session.streamid = attr.id; - + + if session.secure and not session.cert_chain_status then check_cert_status(session); end + -- Send unauthed buffer -- (stanzas which are fine to send before dialback) -- Note that this is *not* the stanza queue (which @@ -464,9 +546,11 @@ function make_authenticated(session, host) elseif session.type == "s2sin_unauthed" then session.type = "s2sin"; if host then + if not session.hosts[host] then session.hosts[host] = {}; end session.hosts[host].authed = true; end elseif session.type == "s2sin" and host then + if not session.hosts[host] then session.hosts[host] = {}; end session.hosts[host].authed = true; else return false; @@ -487,8 +571,16 @@ function mark_connected(session) 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 + function session.send(data) return send_to_host(to, from, data); end + local event_data = { session = session }; + if session.type == "s2sout" then + prosody.events.fire_event("s2sout-established", event_data); + hosts[session.from_host].events.fire_event("s2sout-established", event_data); + else + prosody.events.fire_event("s2sin-established", event_data); + hosts[session.to_host].events.fire_event("s2sin-established", event_data); + end if session.direction == "outgoing" then if sendq then @@ -513,9 +605,10 @@ local resting_session = { -- Resting, not dead close = function (session) session.log("debug", "Attempt to close already-closed session"); end; + filter = function (type, data) return data; end; }; resting_session.__index = resting_session; -function retire_session(session) +function retire_session(session, reason) local log = session.log or log; for k in pairs(session) do if k ~= "trace" and k ~= "log" and k ~= "id" then @@ -523,6 +616,8 @@ function retire_session(session) end end + session.destruction_reason = reason; + 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); @@ -539,7 +634,21 @@ function destroy_session(session, reason) incoming_s2s[session] = nil; end - retire_session(session); -- Clean session until it is GC'd + local event_data = { session = session, reason = reason }; + if session.type == "s2sout" then + prosody.events.fire_event("s2sout-destroyed", event_data); + if hosts[session.from_host] then + hosts[session.from_host].events.fire_event("s2sout-destroyed", event_data); + end + elseif session.type == "s2sin" then + prosody.events.fire_event("s2sin-destroyed", event_data); + if hosts[session.to_host] then + hosts[session.to_host].events.fire_event("s2sin-destroyed", event_data); + end + end + + retire_session(session, reason); -- Clean session until it is GC'd + return true; end return _M;