aboutsummaryrefslogtreecommitdiffstats
path: root/plugins/mod_c2s.lua
blob: cd4b0f71a997e47e7114f8fa215d32ccb89e450e (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
-- 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 add_task = require "util.timer".add_task;
local new_xmpp_stream = require "util.xmppstream".new;
local nameprep = require "util.encodings".stringprep.nameprep;
local sessionmanager = require "core.sessionmanager";
local statsmanager = require "core.statsmanager";
local st = require "util.stanza";
local sm_new_session, sm_destroy_session = sessionmanager.new_session, sessionmanager.destroy_session;
local uuid_generate = require "util.uuid".generate;
local runner = require "util.async".runner;

local tostring, type = tostring, type;

local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams";

local log = module._log;

local c2s_timeout = module:get_option_number("c2s_timeout", 300);
local stream_close_timeout = module:get_option_number("c2s_close_timeout", 5);
local opt_keepalives = module:get_option_boolean("c2s_tcp_keepalives", module:get_option_boolean("tcp_keepalives", true));
local stanza_size_limit = module:get_option_number("c2s_stanza_size_limit", 1024*256);

local measure_connections = module:metric("gauge", "connections", "", "Established c2s connections", {"host", "type", "ip_family"});

local sessions = module:shared("sessions");
local core_process_stanza = prosody.core_process_stanza;
local hosts = prosody.hosts;

local stream_callbacks = { default_ns = "jabber:client" };
local listener = {};
local runner_callbacks = {};

local m_tls_params = module:metric(
	"counter", "encrypted", "",
	"Encrypted connections",
	{"protocol"; "cipher"}
);

module:hook("stats-update", function ()
	-- for push backends, avoid sending out updates for each increment of
	-- the metric below.
	statsmanager.cork()
	measure_connections:clear()
	for _, session in pairs(sessions) do
		local host = session.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
		measure_connections:with_labels(host, type_, "ipv4"):add(is_ipv4)
		measure_connections:with_labels(host, type_, "ipv6"):add(is_ipv6)
	end
	statsmanager.uncork()
end);

--- Stream events handlers
local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};

function stream_callbacks.streamopened(session, attr)
	-- run _streamopened in async context
	session.thread:run({ stream = "opened", attr = attr });
end

function stream_callbacks._streamopened(session, attr)
	local send = session.send;
	if not attr.to then
		session:close{ condition = "improper-addressing",
			text = "A 'to' attribute is required on stream headers" };
		return;
	end
	local host = nameprep(attr.to);
	if not host then
		session:close{ condition = "improper-addressing",
			text = "A valid 'to' attribute is required on stream headers" };
		return;
	end
	if not session.host then
		session.host = host;
	elseif session.host ~= host then
		session:close{ condition = "not-authorized",
			text = "The 'to' attribute must remain the same across stream restarts" };
		return;
	end
	session.version = tonumber(attr.version) or 0;
	session.streamid = uuid_generate();
	(session.log or log)("debug", "Client sent opening <stream:stream> to %s", session.host);

	if not hosts[session.host] or not hosts[session.host].modules.c2s then
		-- We don't serve this host...
		session:close{ condition = "host-unknown", text = "This server does not serve "..tostring(session.host)};
		return;
	end

	session:open_stream(host, attr.from);

	-- Opening the stream can cause the stream to be closed
	if session.destroyed then return end

	(session.log or log)("debug", "Sent reply <stream:stream> to client");
	session.notopen = nil;

	-- If session.secure is *false* (not nil) then it means we /were/ encrypting
	-- since we now have a new stream header, session is secured
	if session.secure == false then
		session.secure = true;
		session.encrypted = true;

		local sock = session.conn:socket();
		if sock.info then
			local info = sock:info();
			(session.log or log)("info", "Stream encrypted (%s with %s)", info.protocol, info.cipher);
			session.compressed = info.compression;
			m_tls_params:with_labels(info.protocol, info.cipher):add(1)
		else
			(session.log or log)("info", "Stream encrypted");
		end
	end

	local features = st.stanza("stream:features");
	hosts[session.host].events.fire_event("stream-features", { origin = session, features = features });
	if features.tags[1] or session.full_jid then
		send(features);
	else
		if session.secure then
			-- Here SASL should be offered
			(session.log or log)("warn", "No stream features to offer on secure session. Check authentication settings.");
		else
			-- Normally STARTTLS would be offered
			(session.log or log)("warn", "No stream features to offer on insecure session. Check encryption and security settings.");
		end
		session:close{ condition = "undefined-condition", text = "No stream features to proceed with" };
	end
end

function stream_callbacks.streamclosed(session, attr)
	-- run _streamclosed in async context
	session.thread:run({ stream = "closed", attr = attr });
end

function stream_callbacks._streamclosed(session)
	session.log("debug", "Received </stream:stream>");
	session:close(false);
end

function stream_callbacks.error(session, error, data)
	if error == "no-stream" then
		session.log("debug", "Invalid opening stream header (%s)", (data:gsub("^([^\1]+)\1", "{%1}")));
		session:close("invalid-namespace");
	elseif error == "parse-error" then
		(session.log or log)("debug", "Client XML parse error: %s", data);
		session:close("not-well-formed");
	elseif error == "stream-error" then
		local condition, text = "undefined-condition";
		for child in data:childtags(nil, xmlns_xmpp_streams) do
			if child.name ~= "text" then
				condition = child.name;
			else
				text = child:get_text();
			end
			if condition ~= "undefined-condition" and text then
				break;
			end
		end
		text = condition .. (text and (" ("..text..")") or "");
		session.log("info", "Session closed by remote with error: %s", text);
		session:close(nil, text);
	end
end

function stream_callbacks.handlestanza(session, stanza)
	stanza = session.filter("stanzas/in", stanza);
	session.thread:run(stanza);
end

--- Session methods
local function session_close(session, reason)
	local log = session.log or log;
	if session.conn then
		if session.notopen then
			session:open_stream();
		end
		if reason then -- nil == no err, initiated by us, false == initiated by client
			local stream_error = st.stanza("stream:error");
			if type(reason) == "string" then -- assume stream error
				stream_error:tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' });
			elseif type(reason) == "table" then
				if reason.condition then
					stream_error:tag(reason.condition, stream_xmlns_attr):up();
					if reason.text then
						stream_error:tag("text", stream_xmlns_attr):text(reason.text):up();
					end
					if reason.extra then
						stream_error:add_child(reason.extra);
					end
				elseif reason.name then -- a stanza
					stream_error = reason;
				end
			end
			stream_error = tostring(stream_error);
			log("debug", "Disconnecting client, <stream:error> is: %s", stream_error);
			session.send(stream_error);
		end

		session.send("</stream:stream>");
		function session.send() return false; end

		local reason_text = (reason and (reason.name or reason.text or reason.condition)) or reason;
		session.log("debug", "c2s stream for %s closed: %s", session.full_jid or session.ip or "<unknown>", reason_text or "session closed");

		-- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote
		local conn = session.conn;
		if reason_text == nil and not session.notopen and session.type == "c2s" then
			-- Grace time to process data from authenticated cleanly-closed stream
			add_task(stream_close_timeout, function ()
				if not session.destroyed then
					session.log("warn", "Failed to receive a stream close response, closing connection anyway...");
					sm_destroy_session(session, reason_text);
					if conn then conn:close(); end
				end
			end);
		else
			sm_destroy_session(session, reason_text);
			if conn then conn:close(); end
		end
	else
		local reason_text = (reason and (reason.name or reason.text or reason.condition)) or reason;
		sm_destroy_session(session, reason_text);
	end
end

module:hook_global("user-deleted", function(event)
	local username, host = event.username, event.host;
	local user = hosts[host].sessions[username];
	if user and user.sessions then
		for _, session in pairs(user.sessions) do
			session:close{ condition = "not-authorized", text = "Account deleted" };
		end
	end
end, 200);

module:hook_global("user-password-changed", function(event)
	local username, host, resource = event.username, event.host, event.resource;
	local user = hosts[host].sessions[username];
	if user and user.sessions then
		for r, session in pairs(user.sessions) do
			if r ~= resource then
				session:close{ condition = "reset", text = "Password changed" };
			end
		end
	end
end, 200);

function runner_callbacks:ready()
	if self.data.conn then
		self.data.conn:resume();
	else
		(self.data.log or log)("debug", "Session has no connection to resume");
	end
end

function runner_callbacks:waiting()
	if self.data.conn then
		self.data.conn:pause();
	else
		(self.data.log or log)("debug", "Session has no connection to pause while waiting");
	end
end

function runner_callbacks:error(err)
	(self.data.log or log)("error", "Traceback[c2s]: %s", err);
end

--- Port listener
function listener.onconnect(conn)
	local session = sm_new_session(conn);

	session.log("info", "Client connected");

	-- Client is using legacy SSL (otherwise mod_tls sets this flag)
	if conn:ssl() then
		session.secure = true;
		session.encrypted = true;

		-- Check if TLS compression is used
		local sock = conn:socket();
		if sock.info then
			session.compressed = sock:info"compression";
		end
	end

	if opt_keepalives then
		conn:setoption("keepalive", opt_keepalives);
	end

	session.close = session_close;

	local stream = new_xmpp_stream(session, stream_callbacks, stanza_size_limit);
	session.stream = stream;
	session.notopen = true;

	function session.reset_stream()
		session.notopen = true;
		session.stream:reset();
	end

	session.thread = runner(function (stanza)
		if st.is_stanza(stanza) then
			core_process_stanza(session, stanza);
		elseif stanza.stream == "opened" then
			stream_callbacks._streamopened(session, stanza.attr);
		elseif stanza.stream == "closed" then
			stream_callbacks._streamclosed(session, stanza.attr);
		end
	end, runner_callbacks, session);

	local filter = session.filter;
	function session.data(data)
		-- Parse the data, which will store stanzas in session.pending_stanzas
		if data then
			data = filter("bytes/in", data);
			if data then
				local ok, err = stream:feed(data);
				if not ok then
					log("debug", "Received invalid XML (%s) %d bytes: %q", err, #data, data:sub(1, 300));
					if err == "stanza-too-large" then
						session:close({ condition = "policy-violation", text = "XML stanza is too big" });
					else
						session:close("not-well-formed");
					end
				end
			end
		end
	end

	if c2s_timeout then
		add_task(c2s_timeout, function ()
			if session.type == "c2s_unauthed" then
				(session.log or log)("debug", "Connection still not authenticated after c2s_timeout=%gs, closing it", c2s_timeout);
				session:close("connection-timeout");
			end
		end);
	end

	session.dispatch_stanza = stream_callbacks.handlestanza;

	sessions[conn] = session;
end

function listener.onincoming(conn, data)
	local session = sessions[conn];
	if session then
		session.data(data);
	end
end

function listener.ondisconnect(conn, err)
	local session = sessions[conn];
	if session then
		(session.log or log)("info", "Client disconnected: %s", err or "connection closed");
		sm_destroy_session(session, err);
		session.conn = nil;
		sessions[conn]  = nil;
	end
end

function listener.onreadtimeout(conn)
	local session = sessions[conn];
	if session then
		return (hosts[session.host] or prosody).events.fire_event("c2s-read-timeout", { session = session });
	end
end

function listener.ondrain(conn)
	local session = sessions[conn];
	if session then
		return (hosts[session.host] or prosody).events.fire_event("c2s-ondrain", { session = session });
	end
end

local function keepalive(event)
	local session = event.session;
	if not session.notopen then
		return event.session.send(' ');
	end
end

function listener.associate_session(conn, session)
	sessions[conn] = session;
end

function module.add_host(module)
	module:hook("c2s-read-timeout", keepalive, -1);
end

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

module:hook("server-stopping", function(event)
	local reason = event.reason;
	for _, session in pairs(sessions) do
		session:close{ condition = "system-shutdown", text = reason };
	end
end, -100);



module:provides("net", {
	name = "c2s";
	listener = listener;
	default_port = 5222;
	encryption = "starttls";
	multiplex = {
		protocol = "xmpp-client";
		pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>";
	};
});

module:provides("net", {
	name = "legacy_ssl";
	listener = listener;
	encryption = "ssl";
	multiplex = {
		pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>";
	};
});