mirror of
https://github.com/bjc/prosody.git
synced 2025-04-04 13:47:41 +03:00
4dcabd36ce
This event was added in a7c183bb4e64 and is required to make mod_smacks know that a session was intentionally closed and shouldn't be hibernated (see fcea4d9e7502). Because this was missing from mod_websocket's session.close(), mod_smacks would always attempt to hibernate websocket sessions even if they closed cleanly. That mod_websocket has its own copy of session.close() is something to fix another day (probably not in the stable branch). So for now this commit makes the minimal change to get things working again. Thanks to Damian and the Jitsi team for reporting.
375 lines
12 KiB
Lua
375 lines
12 KiB
Lua
-- Prosody IM
|
|
-- Copyright (C) 2012-2014 Florian Zeitz
|
|
--
|
|
-- This project is MIT/X11 licensed. Please see the
|
|
-- COPYING file in the source package for more information.
|
|
--
|
|
-- luacheck: ignore 431/log
|
|
|
|
module:set_global();
|
|
|
|
local add_task = require "util.timer".add_task;
|
|
local add_filter = require "util.filters".add_filter;
|
|
local sha1 = require "util.hashes".sha1;
|
|
local base64 = require "util.encodings".base64.encode;
|
|
local st = require "util.stanza";
|
|
local parse_xml = require "util.xml".parse;
|
|
local contains_token = require "util.http".contains_token;
|
|
local portmanager = require "core.portmanager";
|
|
local sm_destroy_session = require"core.sessionmanager".destroy_session;
|
|
local log = module._log;
|
|
local dbuffer = require "util.dbuffer";
|
|
|
|
local websocket_frames = require"net.websocket.frames";
|
|
local parse_frame = websocket_frames.parse;
|
|
local build_frame = websocket_frames.build;
|
|
local build_close = websocket_frames.build_close;
|
|
local parse_close = websocket_frames.parse_close;
|
|
|
|
local t_concat = table.concat;
|
|
|
|
local stanza_size_limit = module:get_option_number("c2s_stanza_size_limit", 1024 * 256);
|
|
local frame_buffer_limit = module:get_option_number("websocket_frame_buffer_limit", 2 * stanza_size_limit);
|
|
local frame_fragment_limit = module:get_option_number("websocket_frame_fragment_limit", 8);
|
|
local stream_close_timeout = module:get_option_number("c2s_close_timeout", 5);
|
|
local consider_websocket_secure = module:get_option_boolean("consider_websocket_secure");
|
|
local cross_domain = module:get_option("cross_domain_websocket");
|
|
if cross_domain ~= nil then
|
|
module:log("info", "The 'cross_domain_websocket' option has been deprecated");
|
|
end
|
|
local xmlns_framing = "urn:ietf:params:xml:ns:xmpp-framing";
|
|
local xmlns_streams = "http://etherx.jabber.org/streams";
|
|
local xmlns_client = "jabber:client";
|
|
local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
|
|
|
|
module:depends("c2s")
|
|
local sessions = module:shared("c2s/sessions");
|
|
local c2s_listener = portmanager.get_service("c2s").listener;
|
|
|
|
--- Session methods
|
|
local function session_open_stream(session, from, to)
|
|
local attr = {
|
|
xmlns = xmlns_framing,
|
|
["xml:lang"] = "en",
|
|
version = "1.0",
|
|
id = session.streamid or "",
|
|
from = from or session.host, to = to,
|
|
};
|
|
if session.stream_attrs then
|
|
session:stream_attrs(from, to, attr)
|
|
end
|
|
session.send(st.stanza("open", attr));
|
|
end
|
|
|
|
local function session_close(session, reason)
|
|
local log = session.log or log;
|
|
local close_event_payload = { session = session, reason = reason };
|
|
module:context(session.host):fire_event("pre-session-close", close_event_payload);
|
|
reason = close_event_payload.reason;
|
|
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 st.is_stanza(reason) then
|
|
stream_error = reason;
|
|
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
|
|
end
|
|
end
|
|
log("debug", "Disconnecting client, <stream:error> is: %s", stream_error);
|
|
session.send(stream_error);
|
|
end
|
|
|
|
session.send(st.stanza("close", { xmlns = xmlns_framing }));
|
|
function session.send() return false; end
|
|
|
|
-- luacheck: ignore 422/reason
|
|
-- FIXME reason should be handled in common place
|
|
local reason = (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..">"), reason 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 == 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);
|
|
conn:write(build_close(1000, "Stream closed"));
|
|
conn:close();
|
|
end
|
|
end);
|
|
else
|
|
sm_destroy_session(session, reason);
|
|
conn:write(build_close(1000, "Stream closed"));
|
|
conn:close();
|
|
end
|
|
end
|
|
end
|
|
|
|
|
|
--- Filters
|
|
local function filter_open_close(data)
|
|
if not data:find(xmlns_framing, 1, true) then return data; end
|
|
|
|
local oc = parse_xml(data);
|
|
if not oc then return data; end
|
|
if oc.attr.xmlns ~= xmlns_framing then return data; end
|
|
if oc.name == "close" then return "</stream:stream>"; end
|
|
if oc.name == "open" then
|
|
oc.name = "stream:stream";
|
|
oc.attr.xmlns = nil;
|
|
oc.attr["xmlns:stream"] = xmlns_streams;
|
|
return oc:top_tag();
|
|
end
|
|
|
|
return data;
|
|
end
|
|
|
|
local default_get_response_text = "It works! Now point your WebSocket client to this URL to connect to Prosody."
|
|
local websocket_get_response_text = module:get_option_string("websocket_get_response_text", default_get_response_text)
|
|
|
|
local default_get_response_body = [[<!DOCTYPE html><html><head><title>Websocket</title></head><body>
|
|
<p>]]..websocket_get_response_text..[[</p>
|
|
</body></html>]]
|
|
local websocket_get_response_body = module:get_option_string("websocket_get_response_body", default_get_response_body)
|
|
|
|
local function validate_frame(frame, max_length)
|
|
local opcode, length = frame.opcode, frame.length;
|
|
|
|
if max_length and length > max_length then
|
|
return false, 1009, "Payload too large";
|
|
end
|
|
|
|
-- Error cases
|
|
if frame.RSV1 or frame.RSV2 or frame.RSV3 then -- Reserved bits non zero
|
|
return false, 1002, "Reserved bits not zero";
|
|
end
|
|
|
|
if opcode == 0x8 and frame.data then -- close frame
|
|
if length == 1 then
|
|
return false, 1002, "Close frame with payload, but too short for status code";
|
|
elseif length >= 2 then
|
|
local status_code = parse_close(frame.data)
|
|
if status_code < 1000 then
|
|
return false, 1002, "Closed with invalid status code";
|
|
elseif ((status_code > 1003 and status_code < 1007) or status_code > 1011) and status_code < 3000 then
|
|
return false, 1002, "Closed with reserved status code";
|
|
end
|
|
end
|
|
end
|
|
|
|
if opcode >= 0x8 then
|
|
if length > 125 then -- Control frame with too much payload
|
|
return false, 1002, "Payload too large";
|
|
end
|
|
|
|
if not frame.FIN then -- Fragmented control frame
|
|
return false, 1002, "Fragmented control frame";
|
|
end
|
|
end
|
|
|
|
if (opcode > 0x2 and opcode < 0x8) or (opcode > 0xA) then
|
|
return false, 1002, "Reserved opcode";
|
|
end
|
|
|
|
-- Check opcode
|
|
if opcode == 0x2 then -- Binary frame
|
|
return false, 1003, "Only text frames are supported, RFC 7395 3.2";
|
|
elseif opcode == 0x8 then -- Close request
|
|
return false, 1000, "Goodbye";
|
|
end
|
|
|
|
-- Other (XMPP-specific) validity checks
|
|
if not frame.FIN then
|
|
return false, 1003, "Continuation frames are not supported, RFC 7395 3.3.3";
|
|
end
|
|
if opcode == 0x01 and frame.data and frame.data:byte(1, 1) ~= 60 then
|
|
return false, 1007, "Invalid payload start character, RFC 7395 3.3.3";
|
|
end
|
|
|
|
return true;
|
|
end
|
|
|
|
|
|
function handle_request(event)
|
|
local request, response = event.request, event.response;
|
|
local conn = response.conn;
|
|
|
|
conn.starttls = false; -- Prevent mod_tls from believing starttls can be done
|
|
|
|
if not request.headers.sec_websocket_key or request.method ~= "GET" then
|
|
return module:fire_event("http-message", {
|
|
response = event.response;
|
|
---
|
|
title = "Prosody WebSocket endpoint";
|
|
message = websocket_get_response_text;
|
|
warning = not (consider_websocket_secure or request.secure) and "This endpoint is not considered secure!" or nil;
|
|
}) or websocket_get_response_body;
|
|
end
|
|
|
|
local wants_xmpp = contains_token(request.headers.sec_websocket_protocol or "", "xmpp");
|
|
|
|
if not wants_xmpp then
|
|
module:log("debug", "Client didn't want to talk XMPP, list of protocols was %s", request.headers.sec_websocket_protocol or "(empty)");
|
|
return 501;
|
|
end
|
|
|
|
local function websocket_close(code, message)
|
|
conn:write(build_close(code, message));
|
|
conn:close();
|
|
end
|
|
|
|
local function websocket_handle_error(session, code, message)
|
|
if code == 1009 then -- stanza size limit exceeded
|
|
-- we close the session, rather than the connection,
|
|
-- otherwise a resuming client will simply resend the
|
|
-- offending stanza
|
|
session:close({ condition = "policy-violation", text = "stanza too large" });
|
|
else
|
|
websocket_close(code, message);
|
|
end
|
|
end
|
|
|
|
local function handle_frame(frame)
|
|
module:log("debug", "Websocket received frame: opcode=%0x, %i bytes", frame.opcode, #frame.data);
|
|
|
|
-- Check frame makes sense
|
|
local frame_ok, err_status, err_text = validate_frame(frame, stanza_size_limit);
|
|
if not frame_ok then
|
|
return frame_ok, err_status, err_text;
|
|
end
|
|
|
|
local opcode = frame.opcode;
|
|
if opcode == 0x9 then -- Ping frame
|
|
frame.opcode = 0xA;
|
|
frame.MASK = false; -- Clients send masked frames, servers don't, see #1484
|
|
conn:write(build_frame(frame));
|
|
return "";
|
|
elseif opcode == 0xA then -- Pong frame, MAY be sent unsolicited, eg as keepalive
|
|
return "";
|
|
elseif opcode ~= 0x1 then -- Not text frame (which is all we support)
|
|
log("warn", "Received frame with unsupported opcode %i", opcode);
|
|
return "";
|
|
end
|
|
|
|
return frame.data;
|
|
end
|
|
|
|
conn:setlistener(c2s_listener);
|
|
c2s_listener.onconnect(conn);
|
|
|
|
local session = sessions[conn];
|
|
|
|
-- Use upstream IP if a HTTP proxy was used
|
|
-- See mod_http and #540
|
|
session.ip = request.ip;
|
|
|
|
session.secure = consider_websocket_secure or request.secure or session.secure;
|
|
session.websocket_request = request;
|
|
|
|
session.open_stream = session_open_stream;
|
|
session.close = session_close;
|
|
|
|
local frameBuffer = dbuffer.new(frame_buffer_limit, frame_fragment_limit);
|
|
add_filter(session, "bytes/in", function(data)
|
|
if not frameBuffer:write(data) then
|
|
session.log("warn", "websocket frame buffer full - terminating session");
|
|
session:close({ condition = "resource-constraint", text = "frame buffer exceeded" });
|
|
return;
|
|
end
|
|
|
|
local cache = {};
|
|
local frame, length, partial = parse_frame(frameBuffer);
|
|
|
|
while frame do
|
|
frameBuffer:discard(length);
|
|
local result, err_status, err_text = handle_frame(frame);
|
|
if not result then
|
|
websocket_handle_error(session, err_status, err_text);
|
|
break;
|
|
end
|
|
cache[#cache+1] = filter_open_close(result);
|
|
frame, length, partial = parse_frame(frameBuffer);
|
|
end
|
|
|
|
if partial then
|
|
-- The header of the next frame is already in the buffer, run
|
|
-- some early validation here
|
|
local frame_ok, err_status, err_text = validate_frame(partial, stanza_size_limit);
|
|
if not frame_ok then
|
|
websocket_handle_error(session, err_status, err_text);
|
|
end
|
|
end
|
|
|
|
return t_concat(cache, "");
|
|
end);
|
|
|
|
add_filter(session, "stanzas/out", function(stanza)
|
|
stanza = st.clone(stanza);
|
|
local attr = stanza.attr;
|
|
attr.xmlns = attr.xmlns or xmlns_client;
|
|
if stanza.name:find("^stream:") then
|
|
attr["xmlns:stream"] = attr["xmlns:stream"] or xmlns_streams;
|
|
end
|
|
return stanza;
|
|
end, -1000);
|
|
|
|
add_filter(session, "bytes/out", function(data)
|
|
return build_frame({ FIN = true, opcode = 0x01, data = tostring(data)});
|
|
end);
|
|
|
|
response.status_code = 101;
|
|
response.headers.upgrade = "websocket";
|
|
response.headers.connection = "Upgrade";
|
|
response.headers.sec_webSocket_accept = base64(sha1(request.headers.sec_websocket_key .. "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
|
|
response.headers.sec_webSocket_protocol = "xmpp";
|
|
|
|
module:fire_event("websocket-session", { session = session, request = request });
|
|
|
|
session.log("debug", "Sending WebSocket handshake");
|
|
|
|
return "";
|
|
end
|
|
|
|
local function keepalive(event)
|
|
local session = event.session;
|
|
if session.open_stream == session_open_stream then
|
|
return session.conn:write(build_frame({ opcode = 0x9, FIN = true }));
|
|
end
|
|
end
|
|
|
|
function module.add_host(module)
|
|
module:hook("c2s-read-timeout", keepalive, -0.9);
|
|
|
|
module:depends("http");
|
|
module:provides("http", {
|
|
name = "websocket";
|
|
default_path = "xmpp-websocket";
|
|
cors = {
|
|
enabled = true;
|
|
};
|
|
route = {
|
|
["GET"] = handle_request;
|
|
["GET /"] = handle_request;
|
|
};
|
|
});
|
|
|
|
module:hook("c2s-read-timeout", keepalive, -0.9);
|
|
end
|
|
|
|
if require"core.modulemanager".get_modules_for_host("*"):contains(module.name) then
|
|
module:add_host();
|
|
end
|