plugins/mod_websocket.lua

   1 -- Prosody IM
   2 -- Copyright (C) 2012-2014 Florian Zeitz
   3 --
   4 -- This project is MIT/X11 licensed. Please see the
   5 -- COPYING file in the source package for more information.
   6 --
   7 -- luacheck: ignore 431/log
   8
   9 module:set_global();
  10
  11 local add_task = require "util.timer".add_task;
  12 local add_filter = require "util.filters".add_filter;
  13 local sha1 = require "util.hashes".sha1;
  14 local base64 = require "util.encodings".base64.encode;
  15 local st = require "util.stanza";
  16 local parse_xml = require "util.xml".parse;
  17 local contains_token = require "util.http".contains_token;
  18 local portmanager = require "core.portmanager";
  19 local sm_destroy_session = require"core.sessionmanager".destroy_session;
  20 local log = module._log;
  21
  22 local websocket_frames = require"net.websocket.frames";
  23 local parse_frame = websocket_frames.parse;
  24 local build_frame = websocket_frames.build;
  25 local build_close = websocket_frames.build_close;
  26 local parse_close = websocket_frames.parse_close;
  27
  28 local t_concat = table.concat;
  29
  30 local stream_close_timeout = module:get_option_number("c2s_close_timeout", 5);
  31 local consider_websocket_secure = module:get_option_boolean("consider_websocket_secure");
  32 local cross_domain = module:get_option("cross_domain_websocket");
  33 if cross_domain ~= nil then
  34         module:log("info", "The 'cross_domain_websocket' option has been deprecated");
  35 end
  36 local xmlns_framing = "urn:ietf:params:xml:ns:xmpp-framing";
  37 local xmlns_streams = "http://etherx.jabber.org/streams";
  38 local xmlns_client = "jabber:client";
  39 local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
  40
  41 module:depends("c2s")
  42 local sessions = module:shared("c2s/sessions");
  43 local c2s_listener = portmanager.get_service("c2s").listener;
  44
  45 --- Session methods
  46 local function session_open_stream(session, from, to)
  47         local attr = {
  48                 xmlns = xmlns_framing,
  49                 ["xml:lang"] = "en",
  50                 version = "1.0",
  51                 id = session.streamid or "",
  52                 from = from or session.host, to = to,
  53         };
  54         if session.stream_attrs then
  55                 session:stream_attrs(from, to, attr)
  56         end
  57         session.send(st.stanza("open", attr));
  58 end
  59
  60 local function session_close(session, reason)
  61         local log = session.log or log;
  62         if session.conn then
  63                 if session.notopen then
  64                         session:open_stream();
  65                 end
  66                 if reason then -- nil == no err, initiated by us, false == initiated by client
  67                         local stream_error = st.stanza("stream:error");
  68                         if type(reason) == "string" then -- assume stream error
  69                                 stream_error:tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' });
  70                         elseif type(reason) == "table" then
  71                                 if reason.condition then
  72                                         stream_error:tag(reason.condition, stream_xmlns_attr):up();
  73                                         if reason.text then
  74                                                 stream_error:tag("text", stream_xmlns_attr):text(reason.text):up();
  75                                         end
  76                                         if reason.extra then
  77                                                 stream_error:add_child(reason.extra);
  78                                         end
  79                                 elseif reason.name then -- a stanza
  80                                         stream_error = reason;
  81                                 end
  82                         end
  83                         log("debug", "Disconnecting client, <stream:error> is: %s", stream_error);
  84                         session.send(stream_error);
  85                 end
  86
  87                 session.send(st.stanza("close", { xmlns = xmlns_framing }));
  88                 function session.send() return false; end
  89
  90                 -- luacheck: ignore 422/reason
  91                 -- FIXME reason should be handled in common place
  92                 local reason = (reason and (reason.name or reason.text or reason.condition)) or reason;
  93                 session.log("debug", "c2s stream for %s closed: %s", session.full_jid or ("<"..session.ip..">"), reason or "session closed");
  94
  95                 -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote
  96                 local conn = session.conn;
  97                 if reason == nil and not session.notopen and session.type == "c2s" then
  98                         -- Grace time to process data from authenticated cleanly-closed stream
  99                         add_task(stream_close_timeout, function ()
 100                                 if not session.destroyed then
 101                                         session.log("warn", "Failed to receive a stream close response, closing connection anyway...");
 102                                         sm_destroy_session(session, reason);
 103                                         conn:write(build_close(1000, "Stream closed"));
 104                                         conn:close();
 105                                 end
 106                         end);
 107                 else
 108                         sm_destroy_session(session, reason);
 109                         conn:write(build_close(1000, "Stream closed"));
 110                         conn:close();
 111                 end
 112         end
 113 end
 114
 115
 116 --- Filters
 117 local function filter_open_close(data)
 118         if not data:find(xmlns_framing, 1, true) then return data; end
 119
 120         local oc = parse_xml(data);
 121         if not oc then return data; end
 122         if oc.attr.xmlns ~= xmlns_framing then return data; end
 123         if oc.name == "close" then return "</stream:stream>"; end
 124         if oc.name == "open" then
 125                 oc.name = "stream:stream";
 126                 oc.attr.xmlns = nil;
 127                 oc.attr["xmlns:stream"] = xmlns_streams;
 128                 return oc:top_tag();
 129         end
 130
 131         return data;
 132 end
 133 function handle_request(event)
 134         local request, response = event.request, event.response;
 135         local conn = response.conn;
 136
 137         conn.starttls = false; -- Prevent mod_tls from believing starttls can be done
 138
 139         if not request.headers.sec_websocket_key then
 140                 response.headers.content_type = "text/html";
 141                 return [[<!DOCTYPE html><html><head><title>Websocket</title></head><body>
 142                         <p>It works! Now point your WebSocket client to this URL to connect to Prosody.</p>
 143                         </body></html>]];
 144         end
 145
 146         local wants_xmpp = contains_token(request.headers.sec_websocket_protocol or "", "xmpp");
 147
 148         if not wants_xmpp then
 149                 module:log("debug", "Client didn't want to talk XMPP, list of protocols was %s", request.headers.sec_websocket_protocol or "(empty)");
 150                 return 501;
 151         end
 152
 153         local function websocket_close(code, message)
 154                 conn:write(build_close(code, message));
 155                 conn:close();
 156         end
 157
 158         local dataBuffer;
 159         local function handle_frame(frame)
 160                 local opcode = frame.opcode;
 161                 local length = frame.length;
 162                 module:log("debug", "Websocket received frame: opcode=%0x, %i bytes", frame.opcode, #frame.data);
 163
 164                 -- Error cases
 165                 if frame.RSV1 or frame.RSV2 or frame.RSV3 then -- Reserved bits non zero
 166                         websocket_close(1002, "Reserved bits not zero");
 167                         return false;
 168                 end
 169
 170                 if opcode == 0x8 then -- close frame
 171                         if length == 1 then
 172                                 websocket_close(1002, "Close frame with payload, but too short for status code");
 173                                 return false;
 174                         elseif length >= 2 then
 175                                 local status_code = parse_close(frame.data)
 176                                 if status_code < 1000 then
 177                                         websocket_close(1002, "Closed with invalid status code");
 178                                         return false;
 179                                 elseif ((status_code > 1003 and status_code < 1007) or status_code > 1011) and status_code < 3000 then
 180                                         websocket_close(1002, "Closed with reserved status code");
 181                                         return false;
 182                                 end
 183                         end
 184                 end
 185
 186                 if opcode >= 0x8 then
 187                         if length > 125 then -- Control frame with too much payload
 188                                 websocket_close(1002, "Payload too large");
 189                                 return false;
 190                         end
 191
 192                         if not frame.FIN then -- Fragmented control frame
 193                                 websocket_close(1002, "Fragmented control frame");
 194                                 return false;
 195                         end
 196                 end
 197
 198                 if (opcode > 0x2 and opcode < 0x8) or (opcode > 0xA) then
 199                         websocket_close(1002, "Reserved opcode");
 200                         return false;
 201                 end
 202
 203                 if opcode == 0x0 and not dataBuffer then
 204                         websocket_close(1002, "Unexpected continuation frame");
 205                         return false;
 206                 end
 207
 208                 if (opcode == 0x1 or opcode == 0x2) and dataBuffer then
 209                         websocket_close(1002, "Continuation frame expected");
 210                         return false;
 211                 end
 212
 213                 -- Valid cases
 214                 if opcode == 0x0 then -- Continuation frame
 215                         dataBuffer[#dataBuffer+1] = frame.data;
 216                 elseif opcode == 0x1 then -- Text frame
 217                         dataBuffer = {frame.data};
 218                 elseif opcode == 0x2 then -- Binary frame
 219                         websocket_close(1003, "Only text frames are supported");
 220                         return;
 221                 elseif opcode == 0x8 then -- Close request
 222                         websocket_close(1000, "Goodbye");
 223                         return;
 224                 elseif opcode == 0x9 then -- Ping frame
 225                         frame.opcode = 0xA;
 226                         conn:write(build_frame(frame));
 227                         return "";
 228                 elseif opcode == 0xA then -- Pong frame, MAY be sent unsolicited, eg as keepalive
 229                         return "";
 230                 else
 231                         log("warn", "Received frame with unsupported opcode %i", opcode);
 232                         return "";
 233                 end
 234
 235                 if frame.FIN then
 236                         local data = t_concat(dataBuffer, "");
 237                         dataBuffer = nil;
 238                         return data;
 239                 end
 240                 return "";
 241         end
 242
 243         conn:setlistener(c2s_listener);
 244         c2s_listener.onconnect(conn);
 245
 246         local session = sessions[conn];
 247
 248         -- Use upstream IP if a HTTP proxy was used
 249         -- See mod_http and #540
 250         session.ip = request.ip;
 251
 252         session.secure = consider_websocket_secure or session.secure;
 253         session.websocket_request = request;
 254
 255         session.open_stream = session_open_stream;
 256         session.close = session_close;
 257
 258         local frameBuffer = "";
 259         add_filter(session, "bytes/in", function(data)
 260                 local cache = {};
 261                 frameBuffer = frameBuffer .. data;
 262                 local frame, length = parse_frame(frameBuffer);
 263
 264                 while frame do
 265                         frameBuffer = frameBuffer:sub(length + 1);
 266                         local result = handle_frame(frame);
 267                         if not result then return; end
 268                         cache[#cache+1] = filter_open_close(result);
 269                         frame, length = parse_frame(frameBuffer);
 270                 end
 271                 return t_concat(cache, "");
 272         end);
 273
 274         add_filter(session, "stanzas/out", function(stanza)
 275                 stanza = st.clone(stanza);
 276                 local attr = stanza.attr;
 277                 attr.xmlns = attr.xmlns or xmlns_client;
 278                 if stanza.name:find("^stream:") then
 279                         attr["xmlns:stream"] = attr["xmlns:stream"] or xmlns_streams;
 280                 end
 281                 return stanza;
 282         end, -1000);
 283
 284         add_filter(session, "bytes/out", function(data)
 285                 return build_frame({ FIN = true, opcode = 0x01, data = tostring(data)});
 286         end);
 287
 288         response.status_code = 101;
 289         response.headers.upgrade = "websocket";
 290         response.headers.connection = "Upgrade";
 291         response.headers.sec_webSocket_accept = base64(sha1(request.headers.sec_websocket_key .. "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
 292         response.headers.sec_webSocket_protocol = "xmpp";
 293
 294         session.log("debug", "Sending WebSocket handshake");
 295
 296         return "";
 297 end
 298
 299 local function keepalive(event)
 300         local session = event.session;
 301         if session.open_stream == session_open_stream then
 302                 return session.conn:write(build_frame({ opcode = 0x9, FIN = true }));
 303         end
 304 end
 305
 306 module:hook("c2s-read-timeout", keepalive, -0.9);
 307
 308 module:depends("http");
 309 module:provides("http", {
 310         name = "websocket";
 311         default_path = "xmpp-websocket";
 312         route = {
 313                 ["GET"] = handle_request;
 314                 ["GET /"] = handle_request;
 315         };
 316 });
 317
 318 function module.add_host(module)
 319         module:hook("c2s-read-timeout", keepalive, -0.9);
 320 end