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