plugins/mod_bosh.lua

   1 -- Prosody IM
   2 -- Copyright (C) 2008-2010 Matthew Wild
   3 -- Copyright (C) 2008-2010 Waqas Hussain
   4 --
   5 -- This project is MIT/X11 licensed. Please see the
   6 -- COPYING file in the source package for more information.
   7 --
   8
   9 module:set_global(); -- Global module
  10
  11 local hosts = _G.hosts;
  12 local new_xmpp_stream = require "util.xmppstream".new;
  13 local sm = require "core.sessionmanager";
  14 local sm_destroy_session = sm.destroy_session;
  15 local new_uuid = require "util.uuid".generate;
  16 local fire_event = prosody.events.fire_event;
  17 local core_process_stanza = prosody.core_process_stanza;
  18 local st = require "util.stanza";
  19 local logger = require "util.logger";
  20 local log = logger.init("mod_bosh");
  21 local initialize_filters = require "util.filters".initialize;
  22 local math_min = math.min;
  23 local xpcall, tostring, type = xpcall, tostring, type;
  24 local traceback = debug.traceback;
  25
  26 local xmlns_streams = "http://etherx.jabber.org/streams";
  27 local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams";
  28 local xmlns_bosh = "http://jabber.org/protocol/httpbind"; -- (hard-coded into a literal in session.send)
  29
  30 local stream_callbacks = {
  31         stream_ns = xmlns_bosh, stream_tag = "body", default_ns = "jabber:client" };
  32
  33 local BOSH_DEFAULT_HOLD = module:get_option_number("bosh_default_hold", 1);
  34 local BOSH_DEFAULT_INACTIVITY = module:get_option_number("bosh_max_inactivity", 60);
  35 local BOSH_DEFAULT_POLLING = module:get_option_number("bosh_max_polling", 5);
  36 local BOSH_DEFAULT_REQUESTS = module:get_option_number("bosh_max_requests", 2);
  37 local bosh_max_wait = module:get_option_number("bosh_max_wait", 120);
  38
  39 local consider_bosh_secure = module:get_option_boolean("consider_bosh_secure");
  40
  41 local default_headers = { ["Content-Type"] = "text/xml; charset=utf-8" };
  42
  43 local cross_domain = module:get_option("cross_domain_bosh", false);
  44 if cross_domain then
  45         default_headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS";
  46         default_headers["Access-Control-Allow-Headers"] = "Content-Type";
  47         default_headers["Access-Control-Max-Age"] = "7200";
  48
  49         if cross_domain == true then
  50                 default_headers["Access-Control-Allow-Origin"] = "*";
  51         elseif type(cross_domain) == "table" then
  52                 cross_domain = table.concat(cross_domain, ", ");
  53         end
  54         if type(cross_domain) == "string" then
  55                 default_headers["Access-Control-Allow-Origin"] = cross_domain;
  56         end
  57 end
  58
  59 local trusted_proxies = module:get_option_set("trusted_proxies", {"127.0.0.1"})._items;
  60
  61 local function get_ip_from_request(request)
  62         local ip = request.conn:ip();
  63         local forwarded_for = request.headers.x_forwarded_for;
  64         if forwarded_for then
  65                 forwarded_for = forwarded_for..", "..ip;
  66                 for forwarded_ip in forwarded_for:gmatch("[^%s,]+") do
  67                         if not trusted_proxies[forwarded_ip] then
  68                                 ip = forwarded_ip;
  69                         end
  70                 end
  71         end
  72         return ip;
  73 end
  74
  75 local t_insert, t_remove, t_concat = table.insert, table.remove, table.concat;
  76 local os_time = os.time;
  77
  78 -- All sessions, and sessions that have no requests open
  79 local sessions, inactive_sessions = module:shared("sessions", "inactive_sessions");
  80
  81 -- Used to respond to idle sessions (those with waiting requests)
  82 local waiting_requests = {};
  83 function on_destroy_request(request)
  84         log("debug", "Request destroyed: %s", tostring(request));
  85         waiting_requests[request] = nil;
  86         local session = sessions[request.context.sid];
  87         if session then
  88                 local requests = session.requests;
  89                 for i, r in ipairs(requests) do
  90                         if r == request then
  91                                 t_remove(requests, i);
  92                                 break;
  93                         end
  94                 end
  95
  96                 -- If this session now has no requests open, mark it as inactive
  97                 local max_inactive = session.bosh_max_inactive;
  98                 if max_inactive and #requests == 0 then
  99                         inactive_sessions[session] = os_time() + max_inactive;
 100                         (session.log or log)("debug", "BOSH session marked as inactive (for %ds)", max_inactive);
 101                 end
 102         end
 103 end
 104
 105 function handle_OPTIONS(request)
 106         local headers = {};
 107         for k,v in pairs(default_headers) do headers[k] = v; end
 108         headers["Content-Type"] = nil;
 109         return { headers = headers, body = "" };
 110 end
 111
 112 function handle_POST(event)
 113         log("debug", "Handling new request %s: %s\n----------", tostring(event.request), tostring(event.request.body));
 114
 115         local request, response = event.request, event.response;
 116         response.on_destroy = on_destroy_request;
 117         local body = request.body;
 118
 119         local context = { request = request, response = response, notopen = true };
 120         local stream = new_xmpp_stream(context, stream_callbacks);
 121         response.context = context;
 122
 123         -- stream:feed() calls the stream_callbacks, so all stanzas in
 124         -- the body are processed in this next line before it returns.
 125         -- In particular, the streamopened() stream callback is where
 126         -- much of the session logic happens, because it's where we first
 127         -- get to see the 'sid' of this request.
 128         stream:feed(body);
 129
 130         -- Stanzas (if any) in the request have now been processed, and
 131         -- we take care of the high-level BOSH logic here, including
 132         -- giving a response or putting the request "on hold".
 133         local session = sessions[context.sid];
 134         if session then
 135                 -- Session was marked as inactive, since we have
 136                 -- a request open now, unmark it
 137                 if inactive_sessions[session] and #session.requests > 0 then
 138                         inactive_sessions[session] = nil;
 139                 end
 140
 141                 local r = session.requests;
 142                 log("debug", "Session %s has %d out of %d requests open", context.sid, #r, session.bosh_hold);
 143                 log("debug", "and there are %d things in the send_buffer:", #session.send_buffer);
 144                 for i, thing in ipairs(session.send_buffer) do
 145                         log("debug", "    %s", tostring(thing));
 146                 end
 147                 if #r > session.bosh_hold then
 148                         -- We are holding too many requests, send what's in the buffer,
 149                         log("debug", "We are holding too many requests, so...");
 150                         if #session.send_buffer > 0 then
 151                                 log("debug", "...sending what is in the buffer")
 152                                 session.send(t_concat(session.send_buffer));
 153                                 session.send_buffer = {};
 154                         else
 155                                 -- or an empty response
 156                                 log("debug", "...sending an empty response");
 157                                 session.send("");
 158                         end
 159                 elseif #session.send_buffer > 0 then
 160                         log("debug", "Session has data in the send buffer, will send now..");
 161                         local resp = t_concat(session.send_buffer);
 162                         session.send_buffer = {};
 163                         session.send(resp);
 164                 end
 165
 166                 if not response.finished then
 167                         -- We're keeping this request open, to respond later
 168                         log("debug", "Have nothing to say, so leaving request unanswered for now");
 169                         if session.bosh_wait then
 170                                 waiting_requests[response] = os_time() + session.bosh_wait;
 171                         end
 172                 end
 173
 174                 if session.bosh_terminate then
 175                         session.log("debug", "Closing session with %d requests open", #session.requests);
 176                         session:close();
 177                         return nil;
 178                 else
 179                         return true; -- Inform http server we shall reply later
 180                 end
 181         end
 182 end
 183
 184
 185 local function bosh_reset_stream(session) session.notopen = true; end
 186
 187 local stream_xmlns_attr = { xmlns = "urn:ietf:params:xml:ns:xmpp-streams" };
 188
 189 local function bosh_close_stream(session, reason)
 190         (session.log or log)("info", "BOSH client disconnected");
 191
 192         local close_reply = st.stanza("body", { xmlns = xmlns_bosh, type = "terminate",
 193                 ["xmlns:stream"] = xmlns_streams });
 194
 195
 196         if reason then
 197                 close_reply.attr.condition = "remote-stream-error";
 198                 if type(reason) == "string" then -- assume stream error
 199                         close_reply:tag("stream:error")
 200                                 :tag(reason, {xmlns = xmlns_xmpp_streams});
 201                 elseif type(reason) == "table" then
 202                         if reason.condition then
 203                                 close_reply:tag("stream:error")
 204                                         :tag(reason.condition, stream_xmlns_attr):up();
 205                                 if reason.text then
 206                                         close_reply:tag("text", stream_xmlns_attr):text(reason.text):up();
 207                                 end
 208                                 if reason.extra then
 209                                         close_reply:add_child(reason.extra);
 210                                 end
 211                         elseif reason.name then -- a stanza
 212                                 close_reply = reason;
 213                         end
 214                 end
 215                 log("info", "Disconnecting client, <stream:error> is: %s", tostring(close_reply));
 216         end
 217
 218         local response_body = tostring(close_reply);
 219         for _, held_request in ipairs(session.requests) do
 220                 held_request.headers = default_headers;
 221                 held_request:send(response_body);
 222         end
 223         sessions[session.sid]  = nil;
 224         inactive_sessions[session] = nil;
 225         sm_destroy_session(session);
 226 end
 227
 228 -- Handle the <body> tag in the request payload.
 229 function stream_callbacks.streamopened(context, attr)
 230         local request, response = context.request, context.response;
 231         local sid = attr.sid;
 232         log("debug", "BOSH body open (sid: %s)", sid or "<none>");
 233         if not sid then
 234                 -- New session request
 235                 context.notopen = nil; -- Signals that we accept this opening tag
 236
 237                 -- TODO: Sanity checks here (rid, to, known host, etc.)
 238                 if not hosts[attr.to] then
 239                         -- Unknown host
 240                         log("debug", "BOSH client tried to connect to unknown host: %s", tostring(attr.to));
 241                         local close_reply = st.stanza("body", { xmlns = xmlns_bosh, type = "terminate",
 242                                 ["xmlns:stream"] = xmlns_streams, condition = "host-unknown" });
 243                         response:send(tostring(close_reply));
 244                         return;
 245                 end
 246
 247                 -- New session
 248                 sid = new_uuid();
 249                 local session = {
 250                         type = "c2s_unauthed", conn = {}, sid = sid, rid = tonumber(attr.rid)-1, host = attr.to,
 251                         bosh_version = attr.ver, bosh_wait = math_min(attr.wait, bosh_max_wait), streamid = sid,
 252                         bosh_hold = BOSH_DEFAULT_HOLD, bosh_max_inactive = BOSH_DEFAULT_INACTIVITY,
 253                         requests = { }, send_buffer = {}, reset_stream = bosh_reset_stream,
 254                         close = bosh_close_stream, dispatch_stanza = core_process_stanza, notopen = true,
 255                         log = logger.init("bosh"..sid), secure = consider_bosh_secure or request.secure,
 256                         ip = get_ip_from_request(request);
 257                 };
 258                 sessions[sid] = session;
 259
 260                 local filter = initialize_filters(session);
 261
 262                 session.log("debug", "BOSH session created for request from %s", session.ip);
 263                 log("info", "New BOSH session, assigned it sid '%s'", sid);
 264
 265                 -- Send creation response
 266                 local creating_session = true;
 267
 268                 local r = session.requests;
 269                 function session.send(s)
 270                         -- We need to ensure that outgoing stanzas have the jabber:client xmlns
 271                         if s.attr and not s.attr.xmlns then
 272                                 s = st.clone(s);
 273                                 s.attr.xmlns = "jabber:client";
 274                         end
 275                         s = filter("stanzas/out", s);
 276                         --log("debug", "Sending BOSH data: %s", tostring(s));
 277                         t_insert(session.send_buffer, tostring(s));
 278
 279                         local oldest_request = r[1];
 280                         if oldest_request and not session.bosh_processing then
 281                                 log("debug", "We have an open request, so sending on that");
 282                                 oldest_request.headers = default_headers;
 283                                 local body_attr = { xmlns = "http://jabber.org/protocol/httpbind",
 284                                         ["xmlns:stream"] = "http://etherx.jabber.org/streams";
 285                                         type = session.bosh_terminate and "terminate" or nil;
 286                                         sid = sid;
 287                                 };
 288                                 if creating_session then
 289                                         creating_session = nil;
 290                                         body_attr.inactivity = tostring(BOSH_DEFAULT_INACTIVITY);
 291                                         body_attr.polling = tostring(BOSH_DEFAULT_POLLING);
 292                                         body_attr.requests = tostring(BOSH_DEFAULT_REQUESTS);
 293                                         body_attr.wait = tostring(session.bosh_wait);
 294                                         body_attr.hold = tostring(session.bosh_hold);
 295                                         body_attr.authid = sid;
 296                                         body_attr.secure = "true";
 297                                         body_attr.ver  = '1.6';
 298                                         body_attr.from = session.host;
 299                                         body_attr["xmlns:xmpp"] = "urn:xmpp:xbosh";
 300                                         body_attr["xmpp:version"] = "1.0";
 301                                 end
 302                                 oldest_request:send(st.stanza("body", body_attr):top_tag()..t_concat(session.send_buffer).."</body>");
 303                                 session.send_buffer = {};
 304                         end
 305                         return true;
 306                 end
 307                 request.sid = sid;
 308         end
 309
 310         local session = sessions[sid];
 311         if not session then
 312                 -- Unknown sid
 313                 log("info", "Client tried to use sid '%s' which we don't know about", sid);
 314                 response.headers = default_headers;
 315                 response:send(tostring(st.stanza("body", { xmlns = xmlns_bosh, type = "terminate", condition = "item-not-found" })));
 316                 context.notopen = nil;
 317                 return;
 318         end
 319
 320         if session.rid then
 321                 local rid = tonumber(attr.rid);
 322                 local diff = rid - session.rid;
 323                 if diff > 1 then
 324                         session.log("warn", "rid too large (means a request was lost). Last rid: %d New rid: %s", session.rid, attr.rid);
 325                 elseif diff <= 0 then
 326                         -- Repeated, ignore
 327                         session.log("debug", "rid repeated, ignoring: %s (diff %d)", session.rid, diff);
 328                         context.notopen = nil;
 329                         context.ignore = true;
 330                         context.sid = sid;
 331                         t_insert(session.requests, response);
 332                         return;
 333                 end
 334                 session.rid = rid;
 335         end
 336
 337         if attr.type == "terminate" then
 338                 -- Client wants to end this session, which we'll do
 339                 -- after processing any stanzas in this request
 340                 session.bosh_terminate = true;
 341         end
 342
 343         context.notopen = nil; -- Signals that we accept this opening tag
 344         t_insert(session.requests, response);
 345         context.sid = sid;
 346         session.bosh_processing = true; -- Used to suppress replies until processing of this request is done
 347
 348         if session.notopen then
 349                 local features = st.stanza("stream:features");
 350                 hosts[session.host].events.fire_event("stream-features", { origin = session, features = features });
 351                 fire_event("stream-features", session, features);
 352                 session.send(tostring(features));
 353                 session.notopen = nil;
 354         end
 355 end
 356
 357 local function handleerr(err) log("error", "Traceback[bosh]: %s", traceback(tostring(err), 2)); end
 358 function stream_callbacks.handlestanza(context, stanza)
 359         if context.ignore then return; end
 360         log("debug", "BOSH stanza received: %s\n", stanza:top_tag());
 361         local session = sessions[context.sid];
 362         if session then
 363                 if stanza.attr.xmlns == xmlns_bosh then
 364                         stanza.attr.xmlns = nil;
 365                 end
 366                 stanza = session.filter("stanzas/in", stanza);
 367                 if stanza then
 368                         return xpcall(function () return core_process_stanza(session, stanza) end, handleerr);
 369                 end
 370         end
 371 end
 372
 373 function stream_callbacks.streamclosed(request)
 374         local session = sessions[request.sid];
 375         if session then
 376                 session.bosh_processing = false;
 377                 if #session.send_buffer > 0 then
 378                         session.send("");
 379                 end
 380         end
 381 end
 382
 383 function stream_callbacks.error(context, error)
 384         log("debug", "Error parsing BOSH request payload; %s", error);
 385         if not context.sid then
 386                 local response = context.response;
 387                 response.headers = default_headers;
 388                 response.status_code = 400;
 389                 response:send();
 390                 return;
 391         end
 392
 393         local session = sessions[context.sid];
 394         if error == "stream-error" then -- Remote stream error, we close normally
 395                 session:close();
 396         else
 397                 session:close({ condition = "bad-format", text = "Error processing stream" });
 398         end
 399 end
 400
 401 local dead_sessions = {};
 402 function on_timer()
 403         -- log("debug", "Checking for requests soon to timeout...");
 404         -- Identify requests timing out within the next few seconds
 405         local now = os_time() + 3;
 406         for request, reply_before in pairs(waiting_requests) do
 407                 if reply_before <= now then
 408                         log("debug", "%s was soon to timeout (at %d, now %d), sending empty response", tostring(request), reply_before, now);
 409                         -- Send empty response to let the
 410                         -- client know we're still here
 411                         if request.conn then
 412                                 sessions[request.context.sid].send("");
 413                         end
 414                 end
 415         end
 416
 417         now = now - 3;
 418         local n_dead_sessions = 0;
 419         for session, close_after in pairs(inactive_sessions) do
 420                 if close_after < now then
 421                         (session.log or log)("debug", "BOSH client inactive too long, destroying session at %d", now);
 422                         sessions[session.sid]  = nil;
 423                         inactive_sessions[session] = nil;
 424                         n_dead_sessions = n_dead_sessions + 1;
 425                         dead_sessions[n_dead_sessions] = session;
 426                 end
 427         end
 428
 429         for i=1,n_dead_sessions do
 430                 local session = dead_sessions[i];
 431                 dead_sessions[i] = nil;
 432                 sm_destroy_session(session, "BOSH client silent for over "..session.bosh_max_inactive.." seconds");
 433         end
 434         return 1;
 435 end
 436 module:add_timer(1, on_timer);
 437
 438
 439 local GET_response = {
 440         headers = {
 441                 content_type = "text/html";
 442         };
 443         body = [[<html><body>
 444         <p>It works! Now point your BOSH client to this URL to connect to Prosody.</p>
 445         <p>For more information see <a href="http://prosody.im/doc/setting_up_bosh">Prosody: Setting up BOSH</a>.</p>
 446         </body></html>]];
 447 };
 448
 449 function module.add_host(module)
 450         module:depends("http");
 451         module:provides("http", {
 452                 default_path = "/http-bind";
 453                 route = {
 454                         ["GET"] = GET_response;
 455                         ["GET /"] = GET_response;
 456                         ["OPTIONS"] = handle_OPTIONS;
 457                         ["OPTIONS /"] = handle_OPTIONS;
 458                         ["POST"] = handle_POST;
 459                         ["POST /"] = handle_POST;
 460                 };
 461         });
 462 end