diff options
| -rw-r--r-- | plugins/mod_websocket.lua | 301 |
1 files changed, 301 insertions, 0 deletions
diff --git a/plugins/mod_websocket.lua b/plugins/mod_websocket.lua new file mode 100644 index 00000000..313dbd41 --- /dev/null +++ b/plugins/mod_websocket.lua @@ -0,0 +1,301 @@ +-- 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. +-- + +module:set_global(); + +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 portmanager = require "core.portmanager"; +local sm_destroy_session = sessionmanager.destroy_session; +local log = module._log; + +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 consider_websocket_secure = module:get_option_boolean("consider_websocket_secure"); +local cross_domain = module:get_option("cross_domain_websocket"); +if cross_domain then + if cross_domain == true then + cross_domain = "*"; + elseif type(cross_domain) == "table" then + cross_domain = t_concat(cross_domain, ", "); + end + if type(cross_domain) ~= "string" then + cross_domain = nil; + end +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) + local attr = { + xmlns = xmlns_framing, + version = "1.0", + id = session.streamid or "", + from = session.host + }; + session.send(st.stanza("open", attr)); +end + +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 + log("debug", "Disconnecting client, <stream:error> is: %s", tostring(stream_error)); + session.send(stream_error); + end + + session.send(st.stanza("close", { xmlns = xmlns_framing })); + function session.send() return false; end + + 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 +function handle_request(event, path) + local request, response = event.request, event.response; + local conn = response.conn; + + if not request.headers.sec_websocket_key then + response.headers.content_type = "text/html"; + return [[<!DOCTYPE html><html><head><title>Websocket</title></head><body> + <p>It works! Now point your WebSocket client to this URL to connect to Prosody.</p> + </body></html>]]; + end + + local wants_xmpp = false; + (request.headers.sec_websocket_protocol or ""):gsub("([^,]*),?", function (proto) + if proto == "xmpp" then wants_xmpp = true; end + end); + + if not wants_xmpp then + return 501; + end + + local function websocket_close(code, message) + conn:write(build_close(code, message)); + conn:close(); + end + + local dataBuffer; + local function handle_frame(frame) + local opcode = frame.opcode; + local length = frame.length; + module:log("debug", "Websocket received frame: opcode=%0x, %i bytes", frame.opcode, #frame.data); + + -- Error cases + if frame.RSV1 or frame.RSV2 or frame.RSV3 then -- Reserved bits non zero + websocket_close(1002, "Reserved bits not zero"); + return false; + end + + if opcode == 0x8 then -- close frame + if length == 1 then + websocket_close(1002, "Close frame with payload, but too short for status code"); + return false; + elseif length >= 2 then + local status_code = parse_close(frame.data) + if status_code < 1000 then + websocket_close(1002, "Closed with invalid status code"); + return false; + elseif ((status_code > 1003 and status_code < 1007) or status_code > 1011) and status_code < 3000 then + websocket_close(1002, "Closed with reserved status code"); + return false; + end + end + end + + if opcode >= 0x8 then + if length > 125 then -- Control frame with too much payload + websocket_close(1002, "Payload too large"); + return false; + end + + if not frame.FIN then -- Fragmented control frame + websocket_close(1002, "Fragmented control frame"); + return false; + end + end + + if (opcode > 0x2 and opcode < 0x8) or (opcode > 0xA) then + websocket_close(1002, "Reserved opcode"); + return false; + end + + if opcode == 0x0 and not dataBuffer then + websocket_close(1002, "Unexpected continuation frame"); + return false; + end + + if (opcode == 0x1 or opcode == 0x2) and dataBuffer then + websocket_close(1002, "Continuation frame expected"); + return false; + end + + -- Valid cases + if opcode == 0x0 then -- Continuation frame + dataBuffer[#dataBuffer+1] = frame.data; + elseif opcode == 0x1 then -- Text frame + dataBuffer = {frame.data}; + elseif opcode == 0x2 then -- Binary frame + websocket_close(1003, "Only text frames are supported"); + return; + elseif opcode == 0x8 then -- Close request + websocket_close(1000, "Goodbye"); + return; + elseif opcode == 0x9 then -- Ping frame + frame.opcode = 0xA; + conn:write(build_frame(frame)); + return ""; + elseif opcode == 0xA then -- Pong frame + module:log("warn", "Received unexpected pong frame: " .. tostring(frame.data)); + return ""; + else + log("warn", "Received frame with unsupported opcode %i", opcode); + return ""; + end + + if frame.FIN then + local data = t_concat(dataBuffer, ""); + dataBuffer = nil; + return data; + end + return ""; + end + + conn:setlistener(c2s_listener); + c2s_listener.onconnect(conn); + + local session = sessions[conn]; + + session.secure = consider_websocket_secure or session.secure; + + session.open_stream = session_open_stream; + session.close = session_close; + + local frameBuffer = ""; + add_filter(session, "bytes/in", function(data) + local cache = {}; + frameBuffer = frameBuffer .. data; + local frame, length = parse_frame(frameBuffer); + + while frame do + frameBuffer = frameBuffer:sub(length + 1); + local result = handle_frame(frame); + if not result then return; end + cache[#cache+1] = filter_open_close(result); + frame, length = parse_frame(frameBuffer); + end + return t_concat(cache, ""); + end); + + add_filter(session, "stanzas/out", function(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); + + 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"; + response.headers.access_control_allow_origin = cross_domain; + + return ""; +end + +function module.add_host(module) + module:depends("http"); + module:provides("http", { + name = "websocket"; + default_path = "xmpp-websocket"; + route = { + ["GET"] = handle_request; + ["GET /"] = handle_request; + }; + }); +end |