X-Git-Url: https://git.enpas.org/?a=blobdiff_plain;f=plugins%2Fmod_bosh.lua;h=8b612286f7c0b7116be3ef830686f0c535acaf41;hb=e8ce67b7db3a77b0833cce7ad1ebb707bff234a6;hp=6550a44ae871a1f6d8265aebca8ad758cf64862d;hpb=4ed7b8f65197e08591b44c2fbd7781d1bee4cf32;p=prosody.git diff --git a/plugins/mod_bosh.lua b/plugins/mod_bosh.lua index 6550a44a..8b612286 100644 --- a/plugins/mod_bosh.lua +++ b/plugins/mod_bosh.lua @@ -6,12 +6,10 @@ -- COPYING file in the source package for more information. -- -module.host = "*" -- Global module +module:set_global(); -- Global module local hosts = _G.hosts; -local lxp = require "lxp"; local new_xmpp_stream = require "util.xmppstream".new; -local httpserver = require "net.httpserver"; local sm = require "core.sessionmanager"; local sm_destroy_session = sm.destroy_session; local new_uuid = require "util.uuid".generate; @@ -20,7 +18,6 @@ local core_process_stanza = core_process_stanza; local st = require "util.stanza"; local logger = require "util.logger"; local log = logger.init("mod_bosh"); -local timer = require "util.timer"; local xmlns_streams = "http://etherx.jabber.org/streams"; local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; @@ -29,16 +26,17 @@ local xmlns_bosh = "http://jabber.org/protocol/httpbind"; -- (hard-coded into a local stream_callbacks = { stream_ns = xmlns_bosh, stream_tag = "body", default_ns = "jabber:client" }; -local BOSH_DEFAULT_HOLD = tonumber(module:get_option("bosh_default_hold")) or 1; -local BOSH_DEFAULT_INACTIVITY = tonumber(module:get_option("bosh_max_inactivity")) or 60; -local BOSH_DEFAULT_POLLING = tonumber(module:get_option("bosh_max_polling")) or 5; -local BOSH_DEFAULT_REQUESTS = tonumber(module:get_option("bosh_max_requests")) or 2; +local BOSH_DEFAULT_HOLD = module:get_option_number("bosh_default_hold", 1); +local BOSH_DEFAULT_INACTIVITY = module:get_option_number("bosh_max_inactivity", 60); +local BOSH_DEFAULT_POLLING = module:get_option_number("bosh_max_polling", 5); +local BOSH_DEFAULT_REQUESTS = module:get_option_number("bosh_max_requests", 2); local consider_bosh_secure = module:get_option_boolean("consider_bosh_secure"); +local auto_cork = module:get_option_boolean("bosh_auto_cork", false); local default_headers = { ["Content-Type"] = "text/xml; charset=utf-8" }; -local cross_domain = module:get_option("cross_domain_bosh"); +local cross_domain = module:get_option("cross_domain_bosh", false); if cross_domain then default_headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"; default_headers["Access-Control-Allow-Headers"] = "Content-Type"; @@ -57,7 +55,7 @@ end local trusted_proxies = module:get_option_set("trusted_proxies", {"127.0.0.1"})._items; local function get_ip_from_request(request) - local ip = request.handler:ip(); + local ip = request.conn:ip(); local forwarded_for = request.headers["x-forwarded-for"]; if forwarded_for then forwarded_for = forwarded_for..", "..ip; @@ -79,11 +77,12 @@ local inactive_sessions = {}; -- Sessions which have no open requests -- Used to respond to idle sessions (those with waiting requests) local waiting_requests = {}; function on_destroy_request(request) + log("debug", "Request destroyed: %s", tostring(request)); waiting_requests[request] = nil; - local session = sessions[request.sid]; + local session = sessions[request.context.sid]; if session then local requests = session.requests; - for i,r in ipairs(requests) do + for i, r in ipairs(requests) do if r == request then t_remove(requests, i); break; @@ -91,39 +90,43 @@ function on_destroy_request(request) end -- If this session now has no requests open, mark it as inactive - if #requests == 0 and session.bosh_max_inactive and not inactive_sessions[session] then - inactive_sessions[session] = os_time(); - (session.log or log)("debug", "BOSH session marked as inactive at %d", inactive_sessions[session]); + local max_inactive = session.bosh_max_inactive; + if max_inactive and #requests == 0 then + inactive_sessions[session] = os_time() + max_inactive; + (session.log or log)("debug", "BOSH session marked as inactive (for %ds)", max_inactive); end end end -function handle_request(method, body, request) - if (not body) or request.method ~= "POST" then - if request.method == "OPTIONS" then - local headers = {}; - for k,v in pairs(default_headers) do headers[k] = v; end - headers["Content-Type"] = nil; - return { headers = headers, body = "" }; - else - return "You really don't look like a BOSH client to me... what do you want?"; - end - end - if not method then - log("debug", "Request %s suffered error %s", tostring(request.id), body); - return; - end - --log("debug", "Handling new request %s: %s\n----------", request.id, tostring(body)); - request.notopen = true; - request.log = log; - request.on_destroy = on_destroy_request; +function handle_OPTIONS(request) + local headers = {}; + for k,v in pairs(default_headers) do headers[k] = v; end + headers["Content-Type"] = nil; + return { headers = headers, body = "" }; +end + +function handle_POST(event) + log("debug", "Handling new request %s: %s\n----------", tostring(event.request), tostring(event.request.body)); + + local request, response = event.request, event.response; + response.on_destroy = on_destroy_request; + local body = request.body; + + local context = { request = request, response = response, notopen = true }; + local stream = new_xmpp_stream(context, stream_callbacks); + response.context = context; - local stream = new_xmpp_stream(request, stream_callbacks); -- stream:feed() calls the stream_callbacks, so all stanzas in -- the body are processed in this next line before it returns. + -- In particular, the streamopened() stream callback is where + -- much of the session logic happens, because it's where we first + -- get to see the 'sid' of this request. stream:feed(body); - local session = sessions[request.sid]; + -- Stanzas (if any) in the request have now been processed, and + -- we take care of the high-level BOSH logic here, including + -- giving a response or putting the request "on hold". + local session = sessions[context.sid]; if session then -- Session was marked as inactive, since we have -- a request open now, unmark it @@ -132,8 +135,11 @@ function handle_request(method, body, request) end local r = session.requests; - log("debug", "Session %s has %d out of %d requests open", request.sid, #r, session.bosh_hold); - log("debug", "and there are %d things in the send_buffer", #session.send_buffer); + log("debug", "Session %s has %d out of %d requests open", context.sid, #r, session.bosh_hold); + log("debug", "and there are %d things in the send_buffer:", #session.send_buffer); + for i, thing in ipairs(session.send_buffer) do + log("debug", " %s", tostring(thing)); + end if #r > session.bosh_hold then -- We are holding too many requests, send what's in the buffer, log("debug", "We are holding too many requests, so..."); @@ -153,12 +159,11 @@ function handle_request(method, body, request) session.send(resp); end - if not request.destroyed then + if not response.finished then -- We're keeping this request open, to respond later log("debug", "Have nothing to say, so leaving request unanswered for now"); if session.bosh_wait then - request.reply_before = os_time() + session.bosh_wait; - waiting_requests[request] = true; + waiting_requests[response] = os_time() + session.bosh_wait; end end @@ -167,7 +172,7 @@ function handle_request(method, body, request) session:close(); return nil; else - return true; -- Inform httpserver we shall reply later + return true; -- Inform http server we shall reply later end end end @@ -181,7 +186,7 @@ local function bosh_close_stream(session, reason) (session.log or log)("info", "BOSH client disconnected"); local close_reply = st.stanza("body", { xmlns = xmlns_bosh, type = "terminate", - ["xmlns:streams"] = xmlns_streams }); + ["xmlns:stream"] = xmlns_streams }); if reason then @@ -206,30 +211,32 @@ local function bosh_close_stream(session, reason) log("info", "Disconnecting client, is: %s", tostring(close_reply)); end - local session_close_response = { headers = default_headers, body = tostring(close_reply) }; - + local response_body = tostring(close_reply); for _, held_request in ipairs(session.requests) do - held_request:send(session_close_response); - held_request:destroy(); + held_request.headers = default_headers; + held_request:send(response_body); end sessions[session.sid] = nil; + inactive_sessions[session] = nil; sm_destroy_session(session); end -function stream_callbacks.streamopened(request, attr) +-- Handle the tag in the request payload. +function stream_callbacks.streamopened(context, attr) + local request, response = context.request, context.response; local sid = attr.sid; log("debug", "BOSH body open (sid: %s)", sid or ""); if not sid then -- New session request - request.notopen = nil; -- Signals that we accept this opening tag + context.notopen = nil; -- Signals that we accept this opening tag -- TODO: Sanity checks here (rid, to, known host, etc.) if not hosts[attr.to] then -- Unknown host log("debug", "BOSH client tried to connect to unknown host: %s", tostring(attr.to)); local close_reply = st.stanza("body", { xmlns = xmlns_bosh, type = "terminate", - ["xmlns:streams"] = xmlns_streams, condition = "host-unknown" }); - request:send(tostring(close_reply)); + ["xmlns:stream"] = xmlns_streams, condition = "host-unknown" }); + response:send(tostring(close_reply)); return; end @@ -248,8 +255,7 @@ function stream_callbacks.streamopened(request, attr) session.log("debug", "BOSH session created for request from %s", session.ip); log("info", "New BOSH session, assigned it sid '%s'", sid); - local r, send_buffer = session.requests, session.send_buffer; - local response = { headers = default_headers } + local r = session.requests; function session.send(s) -- We need to ensure that outgoing stanzas have the jabber:client xmlns if s.attr and not s.attr.xmlns then @@ -258,27 +264,16 @@ function stream_callbacks.streamopened(request, attr) end --log("debug", "Sending BOSH data: %s", tostring(s)); local oldest_request = r[1]; - if oldest_request then + if oldest_request and (not(auto_cork) or waiting_requests[oldest_request]) then log("debug", "We have an open request, so sending on that"); - response.body = t_concat({ + oldest_request.headers = default_headers; + oldest_request:send(t_concat({ "", tostring(s), "" - }); - oldest_request:send(response); - --log("debug", "Sent"); - if oldest_request.stayopen then - if #r>1 then - -- Move front request to back - t_insert(r, oldest_request); - t_remove(r, 1); - end - else - log("debug", "Destroying the request now..."); - oldest_request:destroy(); - end + })); elseif s ~= "" then log("debug", "Saved to send buffer because there are %d open requests", #r); -- Hmm, no requests are open :( @@ -294,7 +289,7 @@ function stream_callbacks.streamopened(request, attr) hosts[session.host].events.fire_event("stream-features", { origin = session, features = features }); fire_event("stream-features", session, features); --xmpp:version='1.0' xmlns:xmpp='urn:xmpp:xbosh' - local response = st.stanza("body", { xmlns = xmlns_bosh, + local body = st.stanza("body", { xmlns = xmlns_bosh, wait = attr.wait, inactivity = tostring(BOSH_DEFAULT_INACTIVITY), polling = tostring(BOSH_DEFAULT_POLLING), @@ -306,7 +301,8 @@ function stream_callbacks.streamopened(request, attr) ["xmlns:xmpp"] = "urn:xmpp:xbosh", ["xmlns:stream"] = "http://etherx.jabber.org/streams" }):add_child(features); - request:send{ headers = default_headers, body = tostring(response) }; + response.headers = default_headers; + response:send(tostring(body)); request.sid = sid; return; @@ -316,8 +312,9 @@ function stream_callbacks.streamopened(request, attr) if not session then -- Unknown sid log("info", "Client tried to use sid '%s' which we don't know about", sid); - request:send{ headers = default_headers, body = tostring(st.stanza("body", { xmlns = xmlns_bosh, type = "terminate", condition = "item-not-found" })) }; - request.notopen = nil; + response.headers = default_headers; + response:send(tostring(st.stanza("body", { xmlns = xmlns_bosh, type = "terminate", condition = "item-not-found" }))); + context.notopen = nil; return; end @@ -329,15 +326,25 @@ function stream_callbacks.streamopened(request, attr) elseif diff <= 0 then -- Repeated, ignore session.log("debug", "rid repeated (on request %s), ignoring: %s (diff %d)", request.id, session.rid, diff); - request.notopen = nil; - request.ignore = true; - request.sid = sid; - t_insert(session.requests, request); + context.notopen = nil; + context.ignore = true; + context.sid = sid; + t_insert(session.requests, response); return; end session.rid = rid; end + if attr.type == "terminate" then + -- Client wants to end this session, which we'll do + -- after processing any stanzas in this request + session.bosh_terminate = true; + end + + context.notopen = nil; -- Signals that we accept this opening tag + t_insert(session.requests, response); + context.sid = sid; + if session.notopen then local features = st.stanza("stream:features"); hosts[session.host].events.fire_event("stream-features", { origin = session, features = features }); @@ -345,22 +352,12 @@ function stream_callbacks.streamopened(request, attr) session.send(features); session.notopen = nil; end - - if attr.type == "terminate" then - -- Client wants to end this session, which we'll do - -- after processing any stanzas in this request - session.bosh_terminate = true; - end - - request.notopen = nil; -- Signals that we accept this opening tag - t_insert(session.requests, request); - request.sid = sid; end -function stream_callbacks.handlestanza(request, stanza) - if request.ignore then return; end +function stream_callbacks.handlestanza(context, stanza) + if context.ignore then return; end log("debug", "BOSH stanza received: %s\n", stanza:top_tag()); - local session = sessions[request.sid]; + local session = sessions[context.sid]; if session then if stanza.attr.xmlns == xmlns_bosh then stanza.attr.xmlns = nil; @@ -369,14 +366,17 @@ function stream_callbacks.handlestanza(request, stanza) end end -function stream_callbacks.error(request, error) +function stream_callbacks.error(context, error) log("debug", "Error parsing BOSH request payload; %s", error); - if not request.sid then - request:send({ headers = default_headers, status = "400 Bad Request" }); + if not context.sid then + local response = context.response; + response.headers = default_headers; + response.status_code = 400; + response:send(); return; end - local session = sessions[request.sid]; + local session = sessions[context.sid]; if error == "stream-error" then -- Remote stream error, we close normally session:close(); else @@ -389,30 +389,26 @@ function on_timer() -- log("debug", "Checking for requests soon to timeout..."); -- Identify requests timing out within the next few seconds local now = os_time() + 3; - for request in pairs(waiting_requests) do - if request.reply_before <= now then - log("debug", "%s was soon to timeout, sending empty response", request.id); + for request, reply_before in pairs(waiting_requests) do + if reply_before <= now then + log("debug", "%s was soon to timeout (at %d, now %d), sending empty response", tostring(request), reply_before, now); -- Send empty response to let the -- client know we're still here if request.conn then - sessions[request.sid].send(""); + sessions[request.context.sid].send(""); end end end now = now - 3; local n_dead_sessions = 0; - for session, inactive_since in pairs(inactive_sessions) do - if session.bosh_max_inactive then - if now - inactive_since > session.bosh_max_inactive then - (session.log or log)("debug", "BOSH client inactive too long, destroying session at %d", now); - sessions[session.sid] = nil; - inactive_sessions[session] = nil; - n_dead_sessions = n_dead_sessions + 1; - dead_sessions[n_dead_sessions] = session; - end - else + for session, close_after in pairs(inactive_sessions) do + if close_after < now then + (session.log or log)("debug", "BOSH client inactive too long, destroying session at %d", now); + sessions[session.sid] = nil; inactive_sessions[session] = nil; + n_dead_sessions = n_dead_sessions + 1; + dead_sessions[n_dead_sessions] = session; end end @@ -423,15 +419,30 @@ function on_timer() end return 1; end +module:add_timer(1, on_timer); -local function setup() - local ports = module:get_option("bosh_ports") or { 5280 }; - httpserver.new_from_config(ports, handle_request, { base = "http-bind" }); - timer.add_task(1, on_timer); -end -if prosody.start_time then -- already started - setup(); -else - prosody.events.add_handler("server-started", setup); +local GET_response = { + headers = { + content_type = "text/html"; + }; + body = [[ +

It works! Now point your BOSH client to this URL to connect to Prosody.

+

For more information see Prosody: Setting up BOSH.

+ ]]; +}; + +function module.add_host(module) + module:depends("http"); + module:provides("http", { + default_path = "/http-bind"; + route = { + ["GET"] = GET_response; + ["GET /"] = GET_response; + ["OPTIONS"] = handle_OPTIONS; + ["OPTIONS /"] = handle_OPTIONS; + ["POST"] = handle_POST; + ["POST /"] = handle_POST; + }; + }); end