-- Prosody IM

-- 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.

--

module:set_global();

local prosody = prosody;

local hosts = prosody.hosts;

local core_process_stanza = prosody.core_process_stanza;

local tostring, type = tostring, type;

local t_insert = table.insert;

local traceback = debug.traceback;

local add_task = require "prosody.util.timer".add_task;

local stop_timer = require "prosody.util.timer".stop;

local st = require "prosody.util.stanza";

local initialize_filters = require "prosody.util.filters".initialize;

local nameprep = require "prosody.util.encodings".stringprep.nameprep;

local new_xmpp_stream = require "prosody.util.xmppstream".new;

local s2s_new_incoming = require "prosody.core.s2smanager".new_incoming;

local s2s_new_outgoing = require "prosody.core.s2smanager".new_outgoing;

local s2s_destroy_session = require "prosody.core.s2smanager".destroy_session;

local uuid_gen = require "prosody.util.uuid".generate;

local async = require "prosody.util.async";

local runner = async.runner;

local connect = require "prosody.net.connect".connect;

local service = require "prosody.net.resolvers.service";

local resolver_chain = require "prosody.net.resolvers.chain";

local errors = require "prosody.util.error";

local set = require "prosody.util.set";

local connect_timeout = module:get_option_period("s2s_timeout", 90);

local stream_close_timeout = module:get_option_period("s2s_close_timeout", 5);

local opt_keepalives = module:get_option_boolean("s2s_tcp_keepalives", module:get_option_boolean("tcp_keepalives", true));

local secure_auth = module:get_option_boolean("s2s_secure_auth", false); -- One day...

local secure_domains, insecure_domains =

	module:get_option_set("s2s_secure_domains", {})._items, module:get_option_set("s2s_insecure_domains", {})._items;

local require_encryption = module:get_option_boolean("s2s_require_encryption", true);

local stanza_size_limit = module:get_option_integer("s2s_stanza_size_limit", 1024*512, 10000);

local measure_connections_inbound = module:metric(

	"gauge", "connections_inbound", "",

	"Established incoming s2s connections",

	{"host", "type", "ip_family"}

);

local measure_connections_outbound = module:metric(

	"gauge", "connections_outbound", "",

	"Established outgoing s2s connections",

	{"host", "type", "ip_family"}

);

local m_accepted_tcp_connections = module:metric(

	"counter", "accepted_tcp", "",

	"Accepted incoming connections on the TCP layer"

);

local m_authn_connections = module:metric(

	"counter", "authenticated", "",

	"Authenticated incoming connections",

	{"host", "direction", "mechanism"}

);

local m_initiated_connections = module:metric(

	"counter", "initiated", "",

	"Initiated outbound connections",

	{"host"}

);

local m_closed_connections = module:metric(

	"counter", "closed", "",

	"Closed connections",

	{"host", "direction", "error"}

);

local m_tls_params = module:metric(

	"counter", "encrypted", "",

	"Encrypted connections",

	{"protocol"; "cipher"}

);

local sessions = module:shared("sessions");

local runner_callbacks = {};

local listener = {};

local log = module._log;

local s2s_service_options = {

	default_port = 5269;

	use_ipv4 = module:get_option_boolean("use_ipv4", true);

	use_ipv6 = module:get_option_boolean("use_ipv6", true);

	use_dane = module:get_option_boolean("use_dane", false);

};

local s2s_service_options_mt = { __index = s2s_service_options }

if module:get_option_boolean("use_dane", false) then

	-- DANE is supported in net.connect but only for outgoing connections,

	-- to authenticate incoming connections with DANE we need

	module:depends("s2s_auth_dane_in");

end

module:hook("stats-update", function ()

	measure_connections_inbound:clear()

	measure_connections_outbound:clear()

	-- TODO: init all expected metrics once?

	-- or maybe create/delete them in host-activate/host-deactivate? requires

	-- extra API in openmetrics.lua tho

	for _, session in pairs(sessions) do

		local is_inbound = string.sub(session.type, 4, 5) == "in"

		local metric_family = is_inbound and measure_connections_inbound or measure_connections_outbound

		local host = is_inbound and session.to_host or session.from_host or ""

		local type_ = session.type or "other"

		-- we want to expose both v4 and v6 counters in all cases to make

		-- queries smoother

		local is_ipv6 = session.ip and session.ip:match(":") and 1 or 0

		local is_ipv4 = 1 - is_ipv6

		metric_family:with_labels(host, type_, "ipv4"):add(is_ipv4)

		metric_family:with_labels(host, type_, "ipv6"):add(is_ipv6)

	end

end);

--- Handle stanzas to remote domains

local bouncy_stanzas = { message = true, presence = true, iq = true };

local function bounce_sendq(session, reason)

	local sendq = session.sendq;

	if not sendq then return; end

	session.log("info", "Sending error replies for %d queued stanzas because of failed outgoing connection to %s", #sendq, session.to_host);

	local dummy = {

		type = "s2sin";

		send = function ()

			(session.log or log)("error", "Replying to to an s2s error reply, please report this! Traceback: %s", traceback());

		end;

		dummy = true;

		close = function ()

			(session.log or log)("error", "Attempting to close the dummy origin of s2s error replies, please report this! Traceback: %s", traceback());

		end;

	};

	-- FIXME Allow for more specific error conditions

	-- TODO use util.error ?

	local error_type = "cancel";

	local condition = "remote-server-not-found";

	local reason_text;

	if session.had_stream then -- set when a stream is opened by the remote

		error_type, condition = "wait", "remote-server-timeout";

	end

	if errors.is_err(reason) then

		error_type, condition, reason_text = reason.type, reason.condition, reason.text;

	elseif type(reason) == "string" then

		reason_text = reason;

	end

	for i, stanza in ipairs(sendq) do

		if not stanza.attr.xmlns and bouncy_stanzas[stanza.name] and stanza.attr.type ~= "error" and stanza.attr.type ~= "result" then

			local reply = st.error_reply(

				stanza,

				error_type,

				condition,

				reason_text and ("Server-to-server connection failed: "..reason_text) or nil

			);

			core_process_stanza(dummy, reply);

		else

			(session.log or log)("debug", "Not eligible for bouncing, discarding %s", stanza:top_tag());

		end

		sendq[i] = nil;

	end

	session.sendq = nil;

end

-- Handles stanzas to existing s2s sessions

function route_to_existing_session(event)

	local from_host, to_host, stanza = event.from_host, event.to_host, event.stanza;

	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

	if hosts[to_host] then

		log("warn", "Attempt to route stanza to a remote %s - a host we do serve?!", from_host);

		return false;

	end

	local host = hosts[from_host].s2sout[to_host];

	if not host then return end

	-- We have a connection to this host already

	if host.type == "s2sout_unauthed" and (stanza.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

		if host.sendq then

			t_insert(host.sendq, st.clone(stanza));

		else

			-- luacheck: ignore 122

			host.sendq = { st.clone(stanza) };

		end

		host.log("debug", "stanza [%s] queued ", stanza.name);

		return true;

	elseif host.type == "local" or host.type == "component" then

		log("error", "Trying to send a stanza to ourselves??")

		log("error", "Traceback: %s", traceback());

		log("error", "Stanza: %s", stanza);

		return false;

	else

		if host.sends2s(stanza) then

			return true;

		end

	end

end

-- Create a new outgoing session for a stanza

function route_to_new_session(event)

	local from_host, to_host, stanza = event.from_host, event.to_host, event.stanza;

	log("debug", "opening a new outgoing connection for this stanza");

	local host_session = s2s_new_outgoing(from_host, to_host);

	host_session.version = 1;

	-- Store in buffer

	host_session.bounce_sendq = bounce_sendq;

	host_session.sendq = { st.clone(stanza) };

	log("debug", "stanza [%s] queued until connection complete", stanza.name);

	-- FIXME Cleaner solution to passing extra data from resolvers to net.server

	-- This mt-clone allows resolvers to add extra data, currently used for DANE TLSA records

	module:context(from_host):fire_event("s2sout-created", { session = host_session });

	local xmpp_extra = setmetatable({}, s2s_service_options_mt);

	local resolver = service.new(to_host, "xmpp-server", "tcp", xmpp_extra);

	if host_session.ssl_ctx then

		local sslctx = host_session.ssl_ctx;

		local xmpps_extra = setmetatable({ default_port = false; servername = to_host; sslctx = sslctx }, s2s_service_options_mt);

		resolver = resolver_chain.new({

			service.new(to_host, "xmpps-server", "tcp", xmpps_extra);

			resolver;

		});

	end

	local pre_event = { session = host_session; resolver = resolver };

	module:context(from_host):fire_event("s2sout-pre-connect", pre_event);

	resolver = pre_event.resolver;

	connect(resolver, listener, nil, { session = host_session });

	m_initiated_connections:with_labels(from_host):add(1)

	return true;

end

local function keepalive(event)

	local session = event.session;

	if not session.notopen then

		return event.session.sends2s(' ');

	end

end

module:hook("s2s-read-timeout", keepalive, -1);

function module.add_host(module)

	if module:get_option_boolean("disallow_s2s", false) then

		module:log("warn", "The 'disallow_s2s' config option is deprecated, please see https://prosody.im/doc/s2s#disabling");

		return nil, "This host has disallow_s2s set";

	end

	module:hook("route/remote", route_to_existing_session, -1);

	module:hook("route/remote", route_to_new_session, -10);

	module:hook("s2sout-stream-features", function (event)

		if stanza_size_limit then

			event.features:tag("limits", { xmlns = "urn:xmpp:stream-limits:0" })

				:text_tag("max-bytes", string.format("%d", stanza_size_limit)):up();

		end

	end);

	module:hook_tag("urn:xmpp:bidi", "bidi", function(session, stanza)

		-- Advertising features on bidi connections where no <stream:features> is sent in the other direction

		local limits = stanza:get_child("limits", "urn:xmpp:stream-limits:0");

		if limits then

			session.outgoing_stanza_size_limit = tonumber(limits:get_child_text("max-bytes"));

		end

	end, 100);

	module:hook("s2s-authenticated", make_authenticated, -1);

	module:hook("s2s-read-timeout", keepalive, -1);

	module:hook("smacks-ack-delayed", function (event)

		if event.origin.type == "s2sin" or event.origin.type == "s2sout" then

			event.origin:close("connection-timeout");

			return true;

		end

	end, -1);

	module:hook_stanza("http://etherx.jabber.org/streams", "features", function (session, stanza) -- luacheck: ignore 212/stanza

		local limits = stanza:get_child("limits", "urn:xmpp:stream-limits:0");

		if limits then

			session.outgoing_stanza_size_limit = tonumber(limits:get_child_text("max-bytes"));

		end

		if session.type == "s2sout" then

			-- Stream is authenticated and we are seem to be done with feature negotiation,

			-- so the stream is ready for stanzas.  RFC 6120 Section 4.3

			mark_connected(session);

			return true;

		elseif require_encryption and not session.secure then

			session.log("warn", "Encrypted server-to-server communication is required but was not offered by %s", session.to_host);

			session:close({

					condition = "policy-violation",

					text = "Encrypted server-to-server communication is required but was not offered",

				}, nil, "Could not establish encrypted connection to remote server");

			return true;

		elseif not session.dialback_verifying then

			session.log("warn", "No SASL EXTERNAL offer and Dialback doesn't seem to be enabled, giving up");

			session:close({

					condition = "unsupported-feature",

					text = "No viable authentication method offered",

				}, nil, "No viable authentication method offered by remote server");

			return true;

		end

	end, -1);

	function module.unload()

		if module.reloading then return end

		for _, session in pairs(sessions) do

			if session.host == module.host then

				session:close("host-gone");

			end

		end

	end

end

-- Stream is authorised, and ready for normal stanzas

function mark_connected(session)

	local sendq = session.sendq;

	local from, to = session.from_host, session.to_host;

	session.log("info", "%s s2s connection %s->%s complete", session.direction:gsub("^.", string.upper), from, to);

	local event_data = { session = session };

	if session.type == "s2sout" then

		module:fire_event("s2sout-established", event_data);

		module:context(from):fire_event("s2sout-established", event_data);

		if session.incoming then

			session.send = function(stanza)

				return module:context(from):fire_event("route/remote", { from_host = from, to_host = to, stanza = stanza });

			end;

		end

	else

		if session.outgoing and not hosts[to].s2sout[from] then

			session.log("debug", "Setting up to handle route from %s to %s", to, from);

			hosts[to].s2sout[from] = session; -- luacheck: ignore 122

		end

		local host_session = hosts[to];

		session.send = function(stanza)

			return host_session.events.fire_event("route/remote", { from_host = to, to_host = from, stanza = stanza });

		end;

		module:fire_event("s2sin-established", event_data);

		module:context(to):fire_event("s2sin-established", event_data);

	end

	if session.direction == "outgoing" then

		if sendq then

			session.log("debug", "sending %d queued stanzas across new outgoing connection to %s", #sendq, session.to_host);

			local send = session.sends2s;

			for i, stanza in ipairs(sendq) do

				send(stanza);

				sendq[i] = nil;

			end

			session.sendq = nil;