diff options
author | Matthew Wild <mwild1@gmail.com> | 2012-04-29 02:10:55 +0100 |
---|---|---|
committer | Matthew Wild <mwild1@gmail.com> | 2012-04-29 02:10:55 +0100 |
commit | bd5f9b8cbfd168a21141607218446222ab9d8642 (patch) | |
tree | e78baabff21e9b6fefaeef4a709098ca30b35922 /net | |
parent | 9b89441dd008a8b3fda12d9217ff6f9ba240df9a (diff) | |
parent | a2055895d362fed58ff314820bb7fddd53e40436 (diff) | |
download | prosody-bd5f9b8cbfd168a21141607218446222ab9d8642.tar.gz prosody-bd5f9b8cbfd168a21141607218446222ab9d8642.zip |
Merge 0.9->trunk
Diffstat (limited to 'net')
-rw-r--r-- | net/connlisteners.lua | 84 | ||||
-rw-r--r-- | net/dns.lua | 1 | ||||
-rw-r--r-- | net/http.lua | 52 | ||||
-rw-r--r-- | net/http/codes.lua | 67 | ||||
-rw-r--r-- | net/http/parser.lua | 139 | ||||
-rw-r--r-- | net/http/server.lua | 273 | ||||
-rw-r--r-- | net/httpclient_listener.lua | 66 | ||||
-rw-r--r-- | net/httpserver.lua | 240 | ||||
-rw-r--r-- | net/httpserver_listener.lua | 46 | ||||
-rw-r--r-- | net/multiplex_listener.lua | 50 | ||||
-rw-r--r-- | net/server_event.lua | 28 | ||||
-rw-r--r-- | net/server_select.lua | 23 | ||||
-rw-r--r-- | net/xmppclient_listener.lua | 179 | ||||
-rw-r--r-- | net/xmppcomponent_listener.lua | 220 | ||||
-rw-r--r-- | net/xmppserver_listener.lua | 209 |
15 files changed, 566 insertions, 1111 deletions
diff --git a/net/connlisteners.lua b/net/connlisteners.lua index 6a227c9d..99ddc720 100644 --- a/net/connlisteners.lua +++ b/net/connlisteners.lua @@ -1,81 +1,15 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- +-- COMPAT w/pre-0.9 +local log = require "util.logger".init("net.connlisteners"); +local traceback = debug.traceback; +module "httpserver" - -local listeners_dir = (CFG_SOURCEDIR or ".").."/net/"; -local server = require "net.server"; -local log = require "util.logger".init("connlisteners"); -local tostring = tostring; -local type = type -local ipairs = ipairs - -local dofile, xpcall, error = - dofile, xpcall, error - -local debug_traceback = debug.traceback; - -module "connlisteners" - -local listeners = {}; - -function register(name, listener) - if listeners[name] and listeners[name] ~= listener then - log("debug", "Listener %s is already registered, not registering any more", name); - return false; - end - listeners[name] = listener; - log("debug", "Registered connection listener %s", name); - return true; +function fail() + log("error", "Attempt to use legacy connlisteners API. For more info see http://prosody.im/doc/developers/network"); + log("error", "Legacy connlisteners API usage, %s", traceback("", 2)); end -function deregister(name) - listeners[name] = nil; -end - -function get(name) - local h = listeners[name]; - if not h then - local ok, ret = xpcall(function() dofile(listeners_dir..name:gsub("[^%w%-]", "_").."_listener.lua") end, debug_traceback); - if not ok then - log("error", "Error while loading listener '%s': %s", tostring(name), tostring(ret)); - return nil, ret; - end - h = listeners[name]; - end - return h; -end - -function start(name, udata) - local h, err = get(name); - if not h then - error("No such connection module: "..name.. (err and (" ("..err..")") or ""), 0); - end - - local interfaces = (udata and udata.interface) or h.default_interface or "*"; - if type(interfaces) == "string" then interfaces = {interfaces}; end - local port = (udata and udata.port) or h.default_port or error("Can't start listener "..name.." because no port was specified, and it has no default port", 0); - local mode = (udata and udata.mode) or h.default_mode or 1; - local ssl = (udata and udata.ssl) or nil; - local autossl = udata and udata.type == "ssl"; - - if autossl and not ssl then - return nil, "no ssl context"; - end - - ok, err = true, {}; - for _, interface in ipairs(interfaces) do - local handler - handler, err[interface] = server.addserver(interface, port, h, mode, autossl and ssl or nil); - ok = ok and handler; - end - - return ok, err; -end +register, deregister = fail, fail; +get, start = fail, fail, epic_fail; return _M; diff --git a/net/dns.lua b/net/dns.lua index 8f428476..5b32a56c 100644 --- a/net/dns.lua +++ b/net/dns.lua @@ -358,6 +358,7 @@ function resolver:name() -- - - - - - - - - - - - - - - - - - - - - - name local remember, pointers = nil, 0; local len = self:byte(); local n = {}; + if len == 0 then return "." end -- Root label while len > 0 do if len >= 0xc0 then -- name is "compressed" pointers = pointers + 1; diff --git a/net/http.lua b/net/http.lua index 6287f408..59f2c080 100644 --- a/net/http.lua +++ b/net/http.lua @@ -13,9 +13,6 @@ local httpstream_new = require "util.httpstream".new; local server = require "net.server" -local connlisteners_get = require "net.connlisteners".get; -local listener = connlisteners_get("httpclient") or error("No httpclient listener!"); - local t_insert, t_concat = table.insert, table.concat; local pairs, ipairs = pairs, ipairs; local tonumber, tostring, xpcall, select, debug_traceback, char, format = @@ -25,6 +22,52 @@ local log = require "util.logger".init("http"); module "http" +local requests = {}; -- Open requests + +local listener = { default_port = 80, default_mode = "*a" }; + +function listener.onconnect(conn) + local req = requests[conn]; + -- Send the request + local request_line = { req.method or "GET", " ", req.path, " HTTP/1.1\r\n" }; + if req.query then + t_insert(request_line, 4, "?"..req.query); + end + + conn:write(t_concat(request_line)); + local t = { [2] = ": ", [4] = "\r\n" }; + for k, v in pairs(req.headers) do + t[1], t[3] = k, v; + conn:write(t_concat(t)); + end + conn:write("\r\n"); + + if req.body then + conn:write(req.body); + end +end + +function listener.onincoming(conn, data) + local request = requests[conn]; + + if not request then + log("warn", "Received response from connection %s with no request attached!", tostring(conn)); + return; + end + + if data and request.reader then + request:reader(data); + end +end + +function listener.ondisconnect(conn, err) + local request = requests[conn]; + if request and request.conn then + request:reader(nil); + end + requests[conn] = nil; +end + function urlencode(s) return s and (s:gsub("%W", function (c) return format("%%%02x", c:byte()); end)); end function urldecode(s) return s and (s:gsub("%%(%x%x)", function (c) return char(tonumber(c,16)); end)); end @@ -152,8 +195,7 @@ function request(u, ex, callback) req.reader = request_reader; req.state = "status"; - listener.register_request(req.handler, req); - + requests[req.handler] = req; return req; end diff --git a/net/http/codes.lua b/net/http/codes.lua new file mode 100644 index 00000000..0cadd079 --- /dev/null +++ b/net/http/codes.lua @@ -0,0 +1,67 @@ + +local response_codes = { + -- Source: http://www.iana.org/assignments/http-status-codes + -- s/^\(\d*\)\s*\(.*\S\)\s*\[RFC.*\]\s*$/^I["\1"] = "\2"; + [100] = "Continue"; + [101] = "Switching Protocols"; + [102] = "Processing"; + + [200] = "OK"; + [201] = "Created"; + [202] = "Accepted"; + [203] = "Non-Authoritative Information"; + [204] = "No Content"; + [205] = "Reset Content"; + [206] = "Partial Content"; + [207] = "Multi-Status"; + [208] = "Already Reported"; + [226] = "IM Used"; + + [300] = "Multiple Choices"; + [301] = "Moved Permanently"; + [302] = "Found"; + [303] = "See Other"; + [304] = "Not Modified"; + [305] = "Use Proxy"; + -- The 306 status code was used in a previous version of [RFC2616], is no longer used, and the code is reserved. + [307] = "Temporary Redirect"; + + [400] = "Bad Request"; + [401] = "Unauthorized"; + [402] = "Payment Required"; + [403] = "Forbidden"; + [404] = "Not Found"; + [405] = "Method Not Allowed"; + [406] = "Not Acceptable"; + [407] = "Proxy Authentication Required"; + [408] = "Request Timeout"; + [409] = "Conflict"; + [410] = "Gone"; + [411] = "Length Required"; + [412] = "Precondition Failed"; + [413] = "Request Entity Too Large"; + [414] = "Request-URI Too Long"; + [415] = "Unsupported Media Type"; + [416] = "Requested Range Not Satisfiable"; + [417] = "Expectation Failed"; + [418] = "I'm a teapot"; + [422] = "Unprocessable Entity"; + [423] = "Locked"; + [424] = "Failed Dependency"; + -- The 425 status code is reserved for the WebDAV advanced collections expired proposal [RFC2817] + [426] = "Upgrade Required"; + + [500] = "Internal Server Error"; + [501] = "Not Implemented"; + [502] = "Bad Gateway"; + [503] = "Service Unavailable"; + [504] = "Gateway Timeout"; + [505] = "HTTP Version Not Supported"; + [506] = "Variant Also Negotiates"; -- Experimental + [507] = "Insufficient Storage"; + [508] = "Loop Detected"; + [510] = "Not Extended"; +}; + +for k,v in pairs(response_codes) do response_codes[k] = k.." "..v; end +return setmetatable(response_codes, { __index = function(t, k) return k.." Unassigned"; end }) diff --git a/net/http/parser.lua b/net/http/parser.lua new file mode 100644 index 00000000..3d9d1a87 --- /dev/null +++ b/net/http/parser.lua @@ -0,0 +1,139 @@ + +local tonumber = tonumber; +local assert = assert; + +local function preprocess_path(path) + if path:sub(1,1) ~= "/" then + path = "/"..path; + end + local level = 0; + for component in path:gmatch("([^/]+)/") do + if component == ".." then + level = level - 1; + elseif component ~= "." then + level = level + 1; + end + if level < 0 then + return nil; + end + end + return path; +end + +local httpstream = {}; + +function httpstream.new(success_cb, error_cb, parser_type, options_cb) + local client = true; + if not parser_type or parser_type == "server" then client = false; else assert(parser_type == "client", "Invalid parser type"); end + local buf = ""; + local chunked; + local state = nil; + local packet; + local len; + local have_body; + local error; + return { + feed = function(self, data) + if error then return nil, "parse has failed"; end + if not data then -- EOF + if state and client and not len then -- reading client body until EOF + packet.body = buf; + success_cb(packet); + elseif buf ~= "" then -- unexpected EOF + error = true; return error_cb(); + end + return; + end + buf = buf..data; + while #buf > 0 do + if state == nil then -- read request + local index = buf:find("\r\n\r\n", nil, true); + if not index then return; end -- not enough data + local method, path, httpversion, status_code, reason_phrase; + local first_line; + local headers = {}; + for line in buf:sub(1,index+1):gmatch("([^\r\n]+)\r\n") do -- parse request + if first_line then + local key, val = line:match("^([^%s:]+): *(.*)$"); + if not key then error = true; return error_cb("invalid-header-line"); end -- TODO handle multi-line and invalid headers + key = key:lower(); + headers[key] = headers[key] and headers[key]..","..val or val; + else + first_line = line; + if client then + httpversion, status_code, reason_phrase = line:match("^HTTP/(1%.[01]) (%d%d%d) (.*)$"); + if not status_code then error = true; return error_cb("invalid-status-line"); end + have_body = not + ( (options_cb and options_cb().method == "HEAD") + or (status_code == 204 or status_code == 304 or status_code == 301) + or (status_code >= 100 and status_code < 200) ); + chunked = have_body and headers["transfer-encoding"] == "chunked"; + else + method, path, httpversion = line:match("^(%w+) (%S+) HTTP/(1%.[01])$"); + if not method then error = true; return error_cb("invalid-status-line"); end + end + end + end + len = tonumber(headers["content-length"]); -- TODO check for invalid len + if client then + -- FIXME handle '100 Continue' response (by skipping it) + if not have_body then len = 0; end + packet = { + code = status_code; + httpversion = httpversion; + headers = headers; + body = have_body and "" or nil; + -- COMPAT the properties below are deprecated + responseversion = httpversion; + responseheaders = headers; + }; + else + -- path normalization + if path:match("^https?://") then + headers.host, path = path:match("^https?://([^/]*)(.*)"); + end + path = preprocess_path(path); + + len = len or 0; + packet = { + method = method; + path = path; + httpversion = httpversion; + headers = headers; + body = nil; + }; + end + buf = buf:sub(index + 4); + state = true; + end + if state then -- read body + if client then + if chunked then + local index = buf:find("\r\n", nil, true); + if not index then return; end -- not enough data + local chunk_size = buf:match("^%x+"); + if not chunk_size then error = true; return error_cb("invalid-chunk-size"); end + chunk_size = tonumber(chunk_size, 16); + index = index + 2; + if chunk_size == 0 then + state = nil; success_cb(packet); + elseif #buf - index + 1 >= chunk_size then -- we have a chunk + packet.body = packet.body..buf:sub(index, index + chunk_size - 1); + buf = buf:sub(index + chunk_size); + end + error("trailers"); -- FIXME MUST read trailers + elseif len and #buf >= len then + packet.body, buf = buf:sub(1, len), buf:sub(len + 1); + state = nil; success_cb(packet); + end + elseif #buf >= len then + packet.body, buf = buf:sub(1, len), buf:sub(len + 1); + state = nil; success_cb(packet); + end + end + end + end; + }; +end + +return httpstream; diff --git a/net/http/server.lua b/net/http/server.lua new file mode 100644 index 00000000..69908e4e --- /dev/null +++ b/net/http/server.lua @@ -0,0 +1,273 @@ + +local t_insert, t_remove, t_concat = table.insert, table.remove, table.concat; +local parser_new = require "net.http.parser".new; +local events = require "util.events".new(); +local addserver = require "net.server".addserver; +local log = require "util.logger".init("http.server"); +local os_date = os.date; +local pairs = pairs; +local s_upper = string.upper; +local setmetatable = setmetatable; +local xpcall = xpcall; +local debug = debug; +local tostring = tostring; +local codes = require "net.http.codes"; + +local _M = {}; + +local sessions = {}; +local listener = {}; +local hosts = {}; +local default_host; + +local function is_wildcard_event(event) + return event:sub(-2, -1) == "/*"; +end +local function is_wildcard_match(wildcard_event, event) + return wildcard_event:sub(1, -2) == event:sub(1, #wildcard_event-1); +end + +local event_map = events._event_map; +setmetatable(events._handlers, { + __index = function (handlers, curr_event) + if is_wildcard_event(curr_event) then return; end -- Wildcard events cannot be fired + -- Find all handlers that could match this event, sort them + -- and then put the array into handlers[curr_event] (and return it) + local matching_handlers_set = {}; + local handlers_array = {}; + for event, handlers_set in pairs(event_map) do + if event == curr_event or + is_wildcard_event(event) and is_wildcard_match(event, curr_event) then + for handler, priority in pairs(handlers_set) do + matching_handlers_set[handler] = { (select(2, event:gsub("/", "%1"))), is_wildcard_event(event) and 0 or 1, priority }; + table.insert(handlers_array, handler); + end + end + end + if #handlers_array > 0 then + table.sort(handlers_array, function(b, a) + local a_score, b_score = matching_handlers_set[a], matching_handlers_set[b]; + for i = 1, #a_score do + if a_score[i] ~= b_score[i] then -- If equal, compare next score value + return a_score[i] < b_score[i]; + end + end + return false; + end); + else + handlers_array = false; + end + rawset(handlers, curr_event, handlers_array); + return handlers_array; + end; + __newindex = function (handlers, curr_event, handlers_array) + if handlers_array == nil + and is_wildcard_event(curr_event) then + -- Invalidate the indexes of all matching events + for event in pairs(handlers) do + if is_wildcard_match(curr_event, event) then + handlers[event] = nil; + end + end + end + rawset(handlers, curr_event, handlers_array); + end; +}); + +local handle_request; +local _1, _2, _3; +local function _handle_request() return handle_request(_1, _2, _3); end + +local last_err; +local function _traceback_handler(err) last_err = err; log("error", "Traceback[http]: %s: %s", tostring(err), debug.traceback()); end +events.add_handler("http-error", function (error) + return "Error processing request: "..codes[error.code]..". Check your error log for more information."; +end, -1); + +function listener.onconnect(conn) + local secure = conn:ssl() and true or nil; + local pending = {}; + local waiting = false; + local function process_next() + --if waiting then log("debug", "can't process_next, waiting"); return; end + if sessions[conn] and #pending > 0 then + local request = t_remove(pending); + --log("debug", "process_next: %s", request.path); + waiting = true; + --handle_request(conn, request, process_next); + _1, _2, _3 = conn, request, process_next; + if not xpcall(_handle_request, _traceback_handler) then + conn:write("HTTP/1.0 500 Internal Server Error\r\n\r\n"..events.fire_event("http-error", { code = 500, private_message = last_err })); + conn:close(); + end + else + --log("debug", "ready for more"); + waiting = false; + end + end + local function success_cb(request) + --log("debug", "success_cb: %s", request.path); + request.secure = secure; + t_insert(pending, request); + if not waiting then + process_next(); + end + end + local function error_cb(err) + log("debug", "error_cb: %s", err or "<nil>"); + -- FIXME don't close immediately, wait until we process current stuff + -- FIXME if err, send off a bad-request response + sessions[conn] = nil; + conn:close(); + end + sessions[conn] = parser_new(success_cb, error_cb); +end + +function listener.ondisconnect(conn) + local open_response = conn._http_open_response; + if open_response and open_response.on_destroy then + open_response.finished = true; + open_response:on_destroy(); + end + sessions[conn] = nil; +end + +function listener.onincoming(conn, data) + sessions[conn]:feed(data); +end + +local headerfix = setmetatable({}, { + __index = function(t, k) + local v = "\r\n"..k:gsub("_", "-"):gsub("%f[%w].", s_upper)..": "; + t[k] = v; + return v; + end +}); + +function _M.hijack_response(response, listener) + error("TODO"); +end +function handle_request(conn, request, finish_cb) + --log("debug", "handler: %s", request.path); + local headers = {}; + for k,v in pairs(request.headers) do headers[k:gsub("-", "_")] = v; end + request.headers = headers; + request.conn = conn; + + local date_header = os_date('!%a, %d %b %Y %H:%M:%S GMT'); -- FIXME use + local conn_header = request.headers.connection; + local keep_alive = conn_header == "Keep-Alive" or (request.httpversion == "1.1" and conn_header ~= "close"); + + local response = { + request = request; + status_code = 200; + headers = { date = date_header, connection = (keep_alive and "Keep-Alive" or "close") }; + conn = conn; + send = _M.send_response; + finish_cb = finish_cb; + }; + conn._http_open_response = response; + + local host = (request.headers.host or ""):match("[^:]+"); + + -- Some sanity checking + local err_code, err; + if not request.path then + err_code, err = 400, "Invalid path"; + elseif not hosts[host] then + if hosts[default_host] then + host = default_host; + elseif host then + err_code, err = 404, "Unknown host: "..host; + else + err_code, err = 400, "Missing or invalid 'Host' header"; + end + end + + if err then + response.status_code = err_code; + response:send(events.fire_event("http-error", { code = err_code, message = err })); + return; + end + + local event = request.method.." "..host..request.path:match("[^?]*"); + local payload = { request = request, response = response }; + --log("debug", "Firing event: %s", event); + local result = events.fire_event(event, payload); + if result ~= nil then + if result ~= true then + local body; + local result_type = type(result); + if result_type == "number" then + response.status_code = result; + if result >= 400 then + body = events.fire_event("http-error", { code = result }); + end + elseif result_type == "string" then + body = result; + elseif result_type == "table" then + for k, v in pairs(result) do + response[k] = v; + end + end + response:send(body); + end + return; + end + + -- if handler not called, return 404 + response.status_code = 404; + response:send(events.fire_event("http-error", { code = 404 })); +end +function _M.send_response(response, body) + if response.finished then return; end + response.finished = true; + response.conn._http_open_response = nil; + + local status_line = "HTTP/"..response.request.httpversion.." "..(response.status or codes[response.status_code]); + local headers = response.headers; + body = body or response.body or ""; + headers.content_length = #body; + + local output = { status_line }; + for k,v in pairs(headers) do + t_insert(output, headerfix[k]..v); + end + t_insert(output, "\r\n\r\n"); + t_insert(output, body); + + response.conn:write(t_concat(output)); + if response.on_destroy then + response:on_destroy(); + response.on_destroy = nil; + end + if headers.connection == "Keep-Alive" then + response:finish_cb(); + else + response.conn:close(); + end +end +function _M.add_handler(event, handler, priority) + events.add_handler(event, handler, priority); +end +function _M.remove_handler(event, handler) + events.remove_handler(event, handler); +end + +function _M.listen_on(port, interface, ssl) + addserver(interface or "*", port, listener, "*a", ssl); +end +function _M.add_host(host) + hosts[host] = true; +end +function _M.remove_host(host) + hosts[host] = nil; +end +function _M.set_default_host(host) + default_host = host; +end + +_M.listener = listener; +_M.codes = codes; +_M._events = events; +return _M; diff --git a/net/httpclient_listener.lua b/net/httpclient_listener.lua deleted file mode 100644 index c4e3c153..00000000 --- a/net/httpclient_listener.lua +++ /dev/null @@ -1,66 +0,0 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- - -local log = require "util.logger".init("httpclient_listener"); -local t_concat, t_insert = table.concat, table.insert; - -local connlisteners_register = require "net.connlisteners".register; - -local requests = {}; -- Open requests -local buffers = {}; -- Buffers of partial lines - -local httpclient = { default_port = 80, default_mode = "*a" }; - -function httpclient.onconnect(conn) - local req = requests[conn]; - -- Send the request - local request_line = { req.method or "GET", " ", req.path, " HTTP/1.1\r\n" }; - if req.query then - t_insert(request_line, 4, "?"..req.query); - end - - conn:write(t_concat(request_line)); - local t = { [2] = ": ", [4] = "\r\n" }; - for k, v in pairs(req.headers) do - t[1], t[3] = k, v; - conn:write(t_concat(t)); - end - conn:write("\r\n"); - - if req.body then - conn:write(req.body); - end -end - -function httpclient.onincoming(conn, data) - local request = requests[conn]; - - if not request then - log("warn", "Received response from connection %s with no request attached!", tostring(conn)); - return; - end - - if data and request.reader then - request:reader(data); - end -end - -function httpclient.ondisconnect(conn, err) - local request = requests[conn]; - if request and request.conn then - request:reader(nil); - end - requests[conn] = nil; -end - -function httpclient.register_request(conn, req) - log("debug", "Attaching request %s to connection %s", tostring(req.id or req), tostring(conn)); - requests[conn] = req; -end - -connlisteners_register("httpclient", httpclient); diff --git a/net/httpserver.lua b/net/httpserver.lua index 74f61c56..7d574788 100644 --- a/net/httpserver.lua +++ b/net/httpserver.lua @@ -1,239 +1,15 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- - - -local server = require "net.server" -local url_parse = require "socket.url".parse; -local httpstream_new = require "util.httpstream".new; - -local connlisteners_start = require "net.connlisteners".start; -local connlisteners_get = require "net.connlisteners".get; -local listener; - -local t_insert, t_concat = table.insert, table.concat; -local tonumber, tostring, pairs, ipairs, type = tonumber, tostring, pairs, ipairs, type; -local xpcall = xpcall; -local debug_traceback = debug.traceback; - -local urlencode = function (s) return s and (s:gsub("%W", function (c) return ("%%%02x"):format(c:byte()); end)); end - -local log = require "util.logger".init("httpserver"); - -local http_servers = {}; +-- COMPAT w/pre-0.9 +local log = require "util.logger".init("net.httpserver"); +local traceback = debug.traceback; module "httpserver" -local default_handler; - -local function send_response(request, response) - -- Write status line - local resp; - if response.body or response.headers then - local body = response.body and tostring(response.body); - log("debug", "Sending response to %s", request.id); - resp = { "HTTP/1.0 "..(response.status or "200 OK").."\r\n" }; - local h = response.headers; - if h then - for k, v in pairs(h) do - t_insert(resp, k..": "..v.."\r\n"); - end - end - if body and not (h and h["Content-Length"]) then - t_insert(resp, "Content-Length: "..#body.."\r\n"); - end - t_insert(resp, "\r\n"); - - if body and request.method ~= "HEAD" then - t_insert(resp, body); - end - request.write(t_concat(resp)); - else - -- Response we have is just a string (the body) - log("debug", "Sending 200 response to %s", request.id or "<none>"); - - local resp = "HTTP/1.0 200 OK\r\n" - .. "Connection: close\r\n" - .. "Content-Type: text/html\r\n" - .. "Content-Length: "..#response.."\r\n" - .. "\r\n" - .. response; - - request.write(resp); - end - if not request.stayopen then - request:destroy(); - end -end - -local function call_callback(request, err) - if request.handled then return; end - request.handled = true; - local callback = request.callback; - if not callback and request.path then - local path = request.url.path; - local base = path:match("^/([^/?]+)"); - if not base then - base = path:match("^http://[^/?]+/([^/?]+)"); - end - - callback = (request.server and request.server.handlers[base]) or default_handler; - end - if callback then - local _callback = callback; - function callback(method, body, request) - local ok, result = xpcall(function() return _callback(method, body, request) end, debug_traceback); - if ok then return result; end - log("error", "Error in HTTP server handler: %s", result); - -- TODO: When we support pipelining, request.destroyed - -- won't be the right flag - we just want to see if there - -- has been a response to this request yet. - if not request.destroyed then - return { - status = "500 Internal Server Error"; - headers = { ["Content-Type"] = "text/plain" }; - body = "There was an error processing your request. See the error log for more details."; - }; - end - end - if err then - log("debug", "Request error: "..err); - if not callback(nil, err, request) then - destroy_request(request); - end - return; - end - - local response = callback(request.method, request.body and t_concat(request.body), request); - if response then - if response == true and not request.destroyed then - -- Keep connection open, we will reply later - log("debug", "Request %s left open, on_destroy is %s", request.id, tostring(request.on_destroy)); - elseif response ~= true then - -- Assume response - send_response(request, response); - destroy_request(request); - end - else - log("debug", "Request handler provided no response, destroying request..."); - -- No response, close connection - destroy_request(request); - end - end -end - -local function request_reader(request, data, startpos) - if not request.parser then - local function success_cb(r) - for k,v in pairs(r) do request[k] = v; end - request.url = url_parse(request.path); - request.url.path = request.url.path and request.url.path:gsub("%%(%x%x)", function(x) return x.char(tonumber(x, 16)) end); - request.body = { request.body }; - call_callback(request); - end - local function error_cb(r) - call_callback(request, r or "connection-closed"); - destroy_request(request); - end - request.parser = httpstream_new(success_cb, error_cb); - end - request.parser:feed(data); -end - --- The default handler for requests -default_handler = function (method, body, request) - log("debug", method.." request for "..tostring(request.path) .. " on port "..request.handler:serverport()); - return { status = "404 Not Found", - headers = { ["Content-Type"] = "text/html" }, - body = "<html><head><title>Page Not Found</title></head><body>Not here :(</body></html>" }; -end - - -function new_request(handler) - return { handler = handler, conn = handler, - write = function (...) return handler:write(...); end, state = "request", - server = http_servers[handler:serverport()], - send = send_response, - destroy = destroy_request, - id = tostring{}:match("%x+$") - }; -end - -function destroy_request(request) - log("debug", "Destroying request %s", request.id); - listener = listener or connlisteners_get("httpserver"); - if not request.destroyed then - request.destroyed = true; - if request.on_destroy then - log("debug", "Request has destroy callback"); - request.on_destroy(request); - else - log("debug", "Request has no destroy callback"); - end - request.handler:close() - if request.conn then - listener.ondisconnect(request.conn, "closed"); - end - end -end - -function new(params) - local http_server = http_servers[params.port]; - if not http_server then - http_server = { handlers = {} }; - http_servers[params.port] = http_server; - -- We weren't already listening on this port, so start now - connlisteners_start("httpserver", params); - end - if params.base then - http_server.handlers[params.base] = params.handler; - end -end - -function set_default_handler(handler) - default_handler = handler; -end - -function new_from_config(ports, handle_request, default_options) - if type(handle_request) == "string" then -- COMPAT with old plugins - log("warn", "Old syntax of httpserver.new_from_config being used to register %s", handle_request); - handle_request, default_options = default_options, { base = handle_request }; - end - ports = ports or {5280}; - for _, options in ipairs(ports) do - local port = default_options.port or 5280; - local base = default_options.base; - local ssl = default_options.ssl or false; - local interface = default_options.interface; - if type(options) == "number" then - port = options; - elseif type(options) == "table" then - port = options.port or port; - base = options.path or base; - ssl = options.ssl or ssl; - interface = options.interface or interface; - elseif type(options) == "string" then - base = options; - end - - if ssl then - ssl.mode = "server"; - ssl.protocol = "sslv23"; - ssl.options = "no_sslv2"; - end - - new{ port = port, interface = interface, - base = base, handler = handle_request, - ssl = ssl, type = (ssl and "ssl") or "tcp" }; - end +function fail() + log("error", "Attempt to use legacy HTTP API. For more info see http://prosody.im/doc/developers/legacy_http"); + log("error", "Legacy HTTP API usage, %s", traceback("", 2)); end -_M.request_reader = request_reader; -_M.send_response = send_response; -_M.urlencode = urlencode; +new, new_from_config = fail, fail; +set_default_handler = fail; return _M; diff --git a/net/httpserver_listener.lua b/net/httpserver_listener.lua deleted file mode 100644 index dd14b43c..00000000 --- a/net/httpserver_listener.lua +++ /dev/null @@ -1,46 +0,0 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- - - - -local connlisteners_register = require "net.connlisteners".register; -local new_request = require "net.httpserver".new_request; -local request_reader = require "net.httpserver".request_reader; - -local requests = {}; -- Open requests - -local httpserver = { default_port = 80, default_mode = "*a" }; - -function httpserver.onincoming(conn, data) - local request = requests[conn]; - - if not request then - request = new_request(conn); - requests[conn] = request; - - -- If using HTTPS, request is secure - if conn:ssl() then - request.secure = true; - end - end - - if data and data ~= "" then - request_reader(request, data); - end -end - -function httpserver.ondisconnect(conn, err) - local request = requests[conn]; - if request and not request.destroyed then - request.conn = nil; - request_reader(request, nil); - end - requests[conn] = nil; -end - -connlisteners_register("httpserver", httpserver); diff --git a/net/multiplex_listener.lua b/net/multiplex_listener.lua deleted file mode 100644 index b515ccce..00000000 --- a/net/multiplex_listener.lua +++ /dev/null @@ -1,50 +0,0 @@ - -local connlisteners_register = require "net.connlisteners".register; -local connlisteners_get = require "net.connlisteners".get; - -local httpserver_listener = connlisteners_get("httpserver"); -local xmppserver_listener = connlisteners_get("xmppserver"); -local xmppclient_listener = connlisteners_get("xmppclient"); -local xmppcomponent_listener = connlisteners_get("xmppcomponent"); - -local server = { default_mode = "*a" }; - -local buffer = {}; - -function server.onincoming(conn, data) - if not data then return; end - local buf = buffer[conn]; - buffer[conn] = nil; - buf = buf and buf..data or data; - if buf:match("^[a-zA-Z]") then - local listener = httpserver_listener; - conn:setlistener(listener); - local onconnect = listener.onconnect; - if onconnect then onconnect(conn) end - listener.onincoming(conn, buf); - elseif buf:match(">") then - local listener; - local xmlns = buf:match("%sxmlns%s*=%s*['\"]([^'\"]*)"); - if xmlns == "jabber:server" then - listener = xmppserver_listener; - elseif xmlns == "jabber:component:accept" then - listener = xmppcomponent_listener; - else - listener = xmppclient_listener; - end - conn:setlistener(listener); - local onconnect = listener.onconnect; - if onconnect then onconnect(conn) end - listener.onincoming(conn, buf); - elseif #buf > 1024 then - conn:close(); - else - buffer[conn] = buf; - end -end - -function server.ondisconnect(conn, err) - buffer[conn] = nil; -- warn if no buffer? -end - -connlisteners_register("multiplex", server); diff --git a/net/server_event.lua b/net/server_event.lua index f4745c34..8d6f5597 100644 --- a/net/server_event.lua +++ b/net/server_event.lua @@ -295,7 +295,10 @@ do end function interface_mt:resume() - return self:_lock(self.nointerface, false, self.nowriting); + self:_lock(self.nointerface, false, self.nowriting); + if not self.eventread then + self.eventread = addevent( base, self.conn, EV_READ, self.readcallback, cfg.READ_TIMEOUT ); -- register callback + end end function interface_mt:counter(c) @@ -340,24 +343,11 @@ do return nil, "writebuffer not empty, waiting" end else - debug( "try to close server with id:", self.id, "args:", now ) + debug( "try to close server with id:", tostring(self.id), "args:", tostring(now) ) self.fatalerror = "server to close" self:_lock( true ) - local count = 0 - for _, item in ipairs( interfacelist( ) ) do - if ( item.type ~= "server" ) and ( item._server == self ) then -- client/server match - if item:close( now ) then -- writebuffer was empty - count = count + 1 - end - end - end - local timeout = 0 -- dont wait for unfinished writebuffers of clients... - if not now then - timeout = cfg.WRITE_TIMEOUT -- ...or wait for it - end - self:_close( timeout ) -- add new event to remove the server interface - debug( "seconds remained until server is closed:", timeout ) - return count -- returns finished clients with empty writebuffer + self:_close( 0 ) -- add new event to remove the server interface + return true end end @@ -642,6 +632,10 @@ do return -1 end end + if interface.noreading then + interface.eventread = nil; + return -1; + end return EV_READ, cfg.READ_TIMEOUT end end diff --git a/net/server_select.lua b/net/server_select.lua index 412d03c0..d6cfc1f8 100644 --- a/net/server_select.lua +++ b/net/server_select.lua @@ -202,6 +202,7 @@ wrapserver = function( listeners, socket, ip, serverport, pattern, sslctx, maxco socket:close( ) _sendlistlen = removesocket( _sendlist, socket, _sendlistlen ) _readlistlen = removesocket( _readlist, socket, _readlistlen ) + _server[ip..":"..serverport] = nil; _socketlist[ socket ] = nil handler = nil socket = nil @@ -596,25 +597,23 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport handler.sendbuffer = handshake handshake( socket ) -- do handshake end - handler.readbuffer = _readbuffer - handler.sendbuffer = _sendbuffer - - if sslctx then - out_put "server.lua: auto-starting ssl negotiation..." - handler.autostart_ssl = true; - handler:starttls(sslctx); - end - - else - handler.readbuffer = _readbuffer - handler.sendbuffer = _sendbuffer end + + handler.readbuffer = _readbuffer + handler.sendbuffer = _sendbuffer send = socket.send receive = socket.receive shutdown = ( ssl and id ) or socket.shutdown _socketlist[ socket ] = handler _readlistlen = addsocket(_readlist, socket, _readlistlen) + + if sslctx and luasec then + out_put "server.lua: auto-starting ssl negotiation..." + handler.autostart_ssl = true; + handler:starttls(sslctx); + end + return handler, socket end diff --git a/net/xmppclient_listener.lua b/net/xmppclient_listener.lua deleted file mode 100644 index 4cc90cbf..00000000 --- a/net/xmppclient_listener.lua +++ /dev/null @@ -1,179 +0,0 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- - - - -local logger = require "logger"; -local log = logger.init("xmppclient_listener"); -local new_xmpp_stream = require "util.xmppstream".new; - -local connlisteners_register = require "net.connlisteners".register; - -local sessionmanager = require "core.sessionmanager"; -local sm_new_session, sm_destroy_session = sessionmanager.new_session, sessionmanager.destroy_session; -local sm_streamopened = sessionmanager.streamopened; -local sm_streamclosed = sessionmanager.streamclosed; -local st = require "util.stanza"; -local xpcall = xpcall; -local tostring = tostring; -local type = type; -local traceback = debug.traceback; - -local config = require "core.configmanager"; -local opt_keepalives = config.get("*", "core", "tcp_keepalives"); - -local stream_callbacks = { default_ns = "jabber:client", - streamopened = sm_streamopened, streamclosed = sm_streamclosed, handlestanza = core_process_stanza }; - -local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; - -function stream_callbacks.error(session, error, data) - if error == "no-stream" then - session.log("debug", "Invalid opening stream header"); - session:close("invalid-namespace"); - elseif error == "parse-error" then - (session.log or log)("debug", "Client XML parse error: %s", tostring(data)); - session:close("not-well-formed"); - elseif error == "stream-error" then - local condition, text = "undefined-condition"; - for child in data:children() do - if child.attr.xmlns == xmlns_xmpp_streams then - if child.name ~= "text" then - condition = child.name; - else - text = child:get_text(); - end - if condition ~= "undefined-condition" and text then - break; - end - end - end - text = condition .. (text and (" ("..text..")") or ""); - session.log("info", "Session closed by remote with error: %s", text); - session:close(nil, text); - end -end - -local function handleerr(err) log("error", "Traceback[c2s]: %s: %s", tostring(err), traceback()); end -function stream_callbacks.handlestanza(session, stanza) - stanza = session.filter("stanzas/in", stanza); - if stanza then - return xpcall(function () return core_process_stanza(session, stanza) end, handleerr); - end -end - -local sessions = {}; -local xmppclient = { default_port = 5222, default_mode = "*a" }; - --- These are session methods -- - -local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'}; -local default_stream_attr = { ["xmlns:stream"] = "http://etherx.jabber.org/streams", xmlns = stream_callbacks.default_ns, version = "1.0", id = "" }; -local function session_close(session, reason) - local log = session.log or log; - if session.conn then - if session.notopen then - session.send("<?xml version='1.0'?>"); - session.send(st.stanza("stream:stream", default_stream_attr):top_tag()); - end - if reason then - if type(reason) == "string" then -- assume stream error - log("info", "Disconnecting client, <stream:error> is: %s", reason); - session.send(st.stanza("stream:error"):tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' })); - elseif type(reason) == "table" then - if reason.condition then - local stanza = st.stanza("stream:error"):tag(reason.condition, stream_xmlns_attr):up(); - if reason.text then - stanza:tag("text", stream_xmlns_attr):text(reason.text):up(); - end - if reason.extra then - stanza:add_child(reason.extra); - end - log("info", "Disconnecting client, <stream:error> is: %s", tostring(stanza)); - session.send(stanza); - elseif reason.name then -- a stanza - log("info", "Disconnecting client, <stream:error> is: %s", tostring(reason)); - session.send(reason); - end - end - end - session.send("</stream:stream>"); - session.conn:close(); - xmppclient.ondisconnect(session.conn, (reason and (reason.text or reason.condition)) or reason or "session closed"); - end -end - - --- End of session methods -- - -function xmppclient.onconnect(conn) - local session = sm_new_session(conn); - sessions[conn] = session; - - session.log("info", "Client connected"); - - -- Client is using legacy SSL (otherwise mod_tls sets this flag) - if conn:ssl() then - session.secure = true; - end - - if opt_keepalives ~= nil then - conn:setoption("keepalive", opt_keepalives); - end - - session.close = session_close; - - local stream = new_xmpp_stream(session, stream_callbacks); - session.stream = stream; - - session.notopen = true; - - function session.reset_stream() - session.notopen = true; - session.stream:reset(); - end - - local filter = session.filter; - function session.data(data) - data = filter("bytes/in", data); - if data then - local ok, err = stream:feed(data); - if ok then return; end - log("debug", "Received invalid XML (%s) %d bytes: %s", tostring(err), #data, data:sub(1, 300):gsub("[\r\n]+", " "):gsub("[%z\1-\31]", "_")); - session:close("not-well-formed"); - end - end - - local handlestanza = stream_callbacks.handlestanza; - function session.dispatch_stanza(session, stanza) - return handlestanza(session, stanza); - end -end - -function xmppclient.onincoming(conn, data) - local session = sessions[conn]; - if session then - session.data(data); - end -end - -function xmppclient.ondisconnect(conn, err) - local session = sessions[conn]; - if session then - (session.log or log)("info", "Client disconnected: %s", err); - sm_destroy_session(session, err); - sessions[conn] = nil; - session = nil; - end -end - -function xmppclient.associate_session(conn, session) - sessions[conn] = session; -end - -connlisteners_register("xmppclient", xmppclient); diff --git a/net/xmppcomponent_listener.lua b/net/xmppcomponent_listener.lua deleted file mode 100644 index 90293559..00000000 --- a/net/xmppcomponent_listener.lua +++ /dev/null @@ -1,220 +0,0 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- - - -local hosts = _G.hosts; - -local t_concat = table.concat; -local tostring = tostring; -local type = type; -local pairs = pairs; - -local lxp = require "lxp"; -local logger = require "util.logger"; -local config = require "core.configmanager"; -local connlisteners = require "net.connlisteners"; -local uuid_gen = require "util.uuid".generate; -local jid_split = require "util.jid".split; -local sha1 = require "util.hashes".sha1; -local st = require "util.stanza"; -local new_xmpp_stream = require "util.xmppstream".new; - -local sessions = {}; - -local log = logger.init("componentlistener"); - -local component_listener = { default_port = 5347; default_mode = "*a"; default_interface = config.get("*", "core", "component_interface") or "127.0.0.1" }; - -local xmlns_component = 'jabber:component:accept'; - ---- Callbacks/data for xmppstream to handle streams for us --- - -local stream_callbacks = { default_ns = xmlns_component }; - -local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; - -function stream_callbacks.error(session, error, data, data2) - if session.destroyed then return; end - log("warn", "Error processing component stream: "..tostring(error)); - if error == "no-stream" then - session:close("invalid-namespace"); - elseif error == "parse-error" then - session.log("warn", "External component %s XML parse error: %s", tostring(session.host), tostring(data)); - session:close("not-well-formed"); - elseif error == "stream-error" then - local condition, text = "undefined-condition"; - for child in data:children() do - if child.attr.xmlns == xmlns_xmpp_streams then - if child.name ~= "text" then - condition = child.name; - else - text = child:get_text(); - end - if condition ~= "undefined-condition" and text then - break; - end - end - end - text = condition .. (text and (" ("..text..")") or ""); - session.log("info", "Session closed by remote with error: %s", text); - session:close(nil, text); - end -end - -function stream_callbacks.streamopened(session, attr) - if config.get(attr.to, "core", "component_module") ~= "component" then - -- Trying to act as a component domain which - -- hasn't been configured - session:close{ condition = "host-unknown", text = tostring(attr.to).." does not match any configured external components" }; - return; - end - - -- Note that we don't create the internal component - -- until after the external component auths successfully - - session.host = attr.to; - session.streamid = uuid_gen(); - session.notopen = nil; - - session.send(st.stanza("stream:stream", { xmlns=xmlns_component, - ["xmlns:stream"]='http://etherx.jabber.org/streams', id=session.streamid, from=session.host }):top_tag()); - -end - -function stream_callbacks.streamclosed(session) - session.log("debug", "Received </stream:stream>"); - session:close(); -end - -local core_process_stanza = core_process_stanza; - -function stream_callbacks.handlestanza(session, stanza) - -- Namespaces are icky. - if not stanza.attr.xmlns and stanza.name == "handshake" then - stanza.attr.xmlns = xmlns_component; - end - if not stanza.attr.xmlns or stanza.attr.xmlns == "jabber:client" then - local from = stanza.attr.from; - if from then - if session.component_validate_from then - local _, domain = jid_split(stanza.attr.from); - if domain ~= session.host then - -- Return error - session.log("warn", "Component sent stanza with missing or invalid 'from' address"); - session:close{ - condition = "invalid-from"; - text = "Component tried to send from address <"..tostring(from) - .."> which is not in domain <"..tostring(session.host)..">"; - }; - return; - end - end - else - stanza.attr.from = session.host; - end - if not stanza.attr.to then - session.log("warn", "Rejecting stanza with no 'to' address"); - session.send(st.error_reply(stanza, "modify", "bad-request", "Components MUST specify a 'to' address on stanzas")); - return; - end - end - return core_process_stanza(session, stanza); -end - ---- Closing a component connection -local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'}; -local default_stream_attr = { ["xmlns:stream"] = "http://etherx.jabber.org/streams", xmlns = stream_callbacks.default_ns, version = "1.0", id = "" }; -local function session_close(session, reason) - if session.destroyed then return; end - local log = session.log or log; - if session.conn then - if session.notopen then - session.send("<?xml version='1.0'?>"); - session.send(st.stanza("stream:stream", default_stream_attr):top_tag()); - end - if reason then - if type(reason) == "string" then -- assume stream error - log("info", "Disconnecting component, <stream:error> is: %s", reason); - session.send(st.stanza("stream:error"):tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' })); - elseif type(reason) == "table" then - if reason.condition then - local stanza = st.stanza("stream:error"):tag(reason.condition, stream_xmlns_attr):up(); - if reason.text then - stanza:tag("text", stream_xmlns_attr):text(reason.text):up(); - end - if reason.extra then - stanza:add_child(reason.extra); - end - log("info", "Disconnecting component, <stream:error> is: %s", tostring(stanza)); - session.send(stanza); - elseif reason.name then -- a stanza - log("info", "Disconnecting component, <stream:error> is: %s", tostring(reason)); - session.send(reason); - end - end - end - session.send("</stream:stream>"); - session.conn:close(); - component_listener.ondisconnect(session.conn, "stream error"); - end -end - ---- Component connlistener -function component_listener.onconnect(conn) - local _send = conn.write; - local session = { type = "component", conn = conn, send = function (data) return _send(conn, tostring(data)); end }; - - -- Logging functions -- - local conn_name = "jcp"..tostring(conn):match("[a-f0-9]+$"); - session.log = logger.init(conn_name); - session.close = session_close; - - session.log("info", "Incoming Jabber component connection"); - - local stream = new_xmpp_stream(session, stream_callbacks); - session.stream = stream; - - session.notopen = true; - - function session.reset_stream() - session.notopen = true; - session.stream:reset(); - end - - function session.data(conn, data) - local ok, err = stream:feed(data); - if ok then return; end - log("debug", "Received invalid XML (%s) %d bytes: %s", tostring(err), #data, data:sub(1, 300):gsub("[\r\n]+", " "):gsub("[%z\1-\31]", "_")); - session:close("not-well-formed"); - end - - session.dispatch_stanza = stream_callbacks.handlestanza; - - sessions[conn] = session; -end -function component_listener.onincoming(conn, data) - local session = sessions[conn]; - session.data(conn, data); -end -function component_listener.ondisconnect(conn, err) - local session = sessions[conn]; - if session then - (session.log or log)("info", "component disconnected: %s (%s)", tostring(session.host), tostring(err)); - if session.on_destroy then session:on_destroy(err); end - sessions[conn] = nil; - for k in pairs(session) do - if k ~= "log" and k ~= "close" then - session[k] = nil; - end - end - session.destroyed = true; - session = nil; - end -end - -connlisteners.register('xmppcomponent', component_listener); diff --git a/net/xmppserver_listener.lua b/net/xmppserver_listener.lua deleted file mode 100644 index 3af0b962..00000000 --- a/net/xmppserver_listener.lua +++ /dev/null @@ -1,209 +0,0 @@ --- Prosody IM --- Copyright (C) 2008-2010 Matthew Wild --- Copyright (C) 2008-2010 Waqas Hussain --- --- This project is MIT/X11 licensed. Please see the --- COPYING file in the source package for more information. --- - - -local tostring = tostring; -local type = type; -local xpcall = xpcall; -local s_format = string.format; -local traceback = debug.traceback; - -local logger = require "logger"; -local log = logger.init("xmppserver_listener"); -local st = require "util.stanza"; -local connlisteners_register = require "net.connlisteners".register; -local new_xmpp_stream = require "util.xmppstream".new; -local s2s_new_incoming = require "core.s2smanager".new_incoming; -local s2s_streamopened = require "core.s2smanager".streamopened; -local s2s_streamclosed = require "core.s2smanager".streamclosed; -local s2s_destroy_session = require "core.s2smanager".destroy_session; -local s2s_attempt_connect = require "core.s2smanager".attempt_connection; -local stream_callbacks = { default_ns = "jabber:server", - streamopened = s2s_streamopened, streamclosed = s2s_streamclosed, handlestanza = core_process_stanza }; - -local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; - -function stream_callbacks.error(session, error, data) - if error == "no-stream" then - session:close("invalid-namespace"); - elseif error == "parse-error" then - session.log("debug", "Server-to-server XML parse error: %s", tostring(error)); - session:close("not-well-formed"); - elseif error == "stream-error" then - local condition, text = "undefined-condition"; - for child in data:children() do - if child.attr.xmlns == xmlns_xmpp_streams then - if child.name ~= "text" then - condition = child.name; - else - text = child:get_text(); - end - if condition ~= "undefined-condition" and text then - break; - end - end - end - text = condition .. (text and (" ("..text..")") or ""); - session.log("info", "Session closed by remote with error: %s", text); - session:close(nil, text); - end -end - -local function handleerr(err) log("error", "Traceback[s2s]: %s: %s", tostring(err), traceback()); end -function stream_callbacks.handlestanza(session, stanza) - if stanza.attr.xmlns == "jabber:client" then --COMPAT: Prosody pre-0.6.2 may send jabber:client - stanza.attr.xmlns = nil; - end - stanza = session.filter("stanzas/in", stanza); - if stanza then - return xpcall(function () return core_process_stanza(session, stanza) end, handleerr); - end -end - -local sessions = {}; -local xmppserver = { default_port = 5269, default_mode = "*a" }; - --- These are session methods -- - -local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'}; -local default_stream_attr = { ["xmlns:stream"] = "http://etherx.jabber.org/streams", xmlns = stream_callbacks.default_ns, version = "1.0", id = "" }; -local function session_close(session, reason, remote_reason) - local log = session.log or log; - if session.conn then - if session.notopen then - session.sends2s("<?xml version='1.0'?>"); - session.sends2s(st.stanza("stream:stream", default_stream_attr):top_tag()); - end - if reason then - if type(reason) == "string" then -- assume stream error - log("info", "Disconnecting %s[%s], <stream:error> is: %s", session.host or "(unknown host)", session.type, reason); - session.sends2s(st.stanza("stream:error"):tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' })); - elseif type(reason) == "table" then - if reason.condition then - local stanza = st.stanza("stream:error"):tag(reason.condition, stream_xmlns_attr):up(); - if reason.text then - stanza:tag("text", stream_xmlns_attr):text(reason.text):up(); - end - if reason.extra then - stanza:add_child(reason.extra); - end - log("info", "Disconnecting %s[%s], <stream:error> is: %s", session.host or "(unknown host)", session.type, tostring(stanza)); - session.sends2s(stanza); - elseif reason.name then -- a stanza - log("info", "Disconnecting %s->%s[%s], <stream:error> is: %s", session.from_host or "(unknown host)", session.to_host or "(unknown host)", session.type, tostring(reason)); - session.sends2s(reason); - end - end - end - session.sends2s("</stream:stream>"); - if session.notopen or not session.conn:close() then - session.conn:close(true); -- Force FIXME: timer? - end - session.conn:close(); - xmppserver.ondisconnect(session.conn, remote_reason or (reason and (reason.text or reason.condition)) or reason or "stream closed"); - end -end - - --- End of session methods -- - -local function initialize_session(session) - local stream = new_xmpp_stream(session, stream_callbacks); - session.stream = stream; - - session.notopen = true; - - function session.reset_stream() - session.notopen = true; - session.stream:reset(); - end - - local filter = session.filter; - function session.data(data) - data = filter("bytes/in", data); - if data then - local ok, err = stream:feed(data); - if ok then return; end - (session.log or log)("warn", "Received invalid XML: %s", data); - (session.log or log)("warn", "Problem was: %s", err); - session:close("not-well-formed"); - end - end - - session.close = session_close; - local handlestanza = stream_callbacks.handlestanza; - function session.dispatch_stanza(session, stanza) - return handlestanza(session, stanza); - end -end - -function xmppserver.onconnect(conn) - if not sessions[conn] then -- May be an existing outgoing session - local session = s2s_new_incoming(conn); - sessions[conn] = session; - - -- Logging functions -- - local conn_name = "s2sin"..tostring(conn):match("[a-f0-9]+$"); - session.log = logger.init(conn_name); - - session.log("info", "Incoming s2s connection"); - - initialize_session(session); - end -end - -function xmppserver.onincoming(conn, data) - local session = sessions[conn]; - if session then - session.data(data); - end -end - -function xmppserver.onstatus(conn, status) - if status == "ssl-handshake-complete" then - local session = sessions[conn]; - if session and session.direction == "outgoing" then - local to_host, from_host = session.to_host, session.from_host; - session.log("debug", "Sending stream header..."); - session.sends2s(s_format([[<stream:stream xmlns='jabber:server' xmlns:db='jabber:server:dialback' xmlns:stream='http://etherx.jabber.org/streams' from='%s' to='%s' version='1.0'>]], from_host, to_host)); - end - end -end - -function xmppserver.ondisconnect(conn, err) - local session = sessions[conn]; - if session then - if err and err ~= "closed" and session.srv_hosts then - (session.log or log)("debug", "s2s connection attempt failed: %s", err); - if s2s_attempt_connect(session, err) then - (session.log or log)("debug", "...so we're going to try another target"); - return; -- Session lives for now - end - end - (session.log or log)("info", "s2s disconnected: %s->%s (%s)", tostring(session.from_host), tostring(session.to_host), tostring(err or "closed")); - s2s_destroy_session(session, err); - sessions[conn] = nil; - session = nil; - end -end - -function xmppserver.register_outgoing(conn, session) - session.direction = "outgoing"; - sessions[conn] = session; - - initialize_session(session); -end - -connlisteners_register("xmppserver", xmppserver); - - --- We need to perform some initialisation when a connection is created --- We also need to perform that same initialisation at other points (SASL, TLS, ...) - --- ...and we need to handle data --- ...and record all sessions associated with connections |