diff options
Diffstat (limited to 'net')
| -rw-r--r-- | net/connlisteners.lua | 81 | ||||
| -rw-r--r-- | net/http.lua | 52 | ||||
| -rw-r--r-- | net/http/codes.lua | 66 | ||||
| -rw-r--r-- | net/http/parser.lua | 116 | ||||
| -rw-r--r-- | net/http/server.lua | 283 | ||||
| -rw-r--r-- | net/httpclient_listener.lua | 66 | ||||
| -rw-r--r-- | net/httpserver.lua | 8 | ||||
| -rw-r--r-- | net/multiplex_listener.lua | 50 | ||||
| -rw-r--r-- | net/server_event.lua | 19 | ||||
| -rw-r--r-- | net/server_select.lua | 1 | ||||
| -rw-r--r-- | net/xmppclient_listener.lua | 179 | ||||
| -rw-r--r-- | net/xmppcomponent_listener.lua | 218 | ||||
| -rw-r--r-- | net/xmppserver_listener.lua | 209 |
13 files changed, 520 insertions, 828 deletions
diff --git a/net/connlisteners.lua b/net/connlisteners.lua deleted file mode 100644 index 6a227c9d..00000000 --- a/net/connlisteners.lua +++ /dev/null @@ -1,81 +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 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; -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 - -return _M; 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..2e701027 --- /dev/null +++ b/net/http/codes.lua @@ -0,0 +1,66 @@ + +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"; + [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..c98c75af --- /dev/null +++ b/net/http/parser.lua @@ -0,0 +1,116 @@ + +local tonumber = tonumber; +local assert = assert; + +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 + path = path:gsub("^//+", "/"); -- TODO parse url more + 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 + 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..185ac9a0 --- /dev/null +++ b/net/http/server.lua @@ -0,0 +1,283 @@ + +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 _G = _G; +local legacy_httpserver = require "net.httpserver"; + +local _M = {}; + +local sessions = {}; +local handlers = {}; + +local listener = {}; + +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[event] + 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"))), priority }; + table.insert(handlers_array, handler); + end + end + end + if #handlers_array == 0 then return; end + 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 ~= b then -- If equal, compare next score value + return a_score[i] < b_score[i]; + end + end + return false; + end); + 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 all matching + for event in pairs(handlers) do + if is_wildcard_match(curr_event, event) then + handlers[event] = nil; + end + end + end + end; +}); + +local handle_request; +local _1, _2, _3; +local function _handle_request() return handle_request(_1, _2, _3); end +local function _traceback_handler(err) log("error", "Traceback[http]: %s: %s", tostring(err), debug.traceback()); end + +function listener.onconnect(conn) + local secure = conn:ssl() and true or nil; + local pending = {}; + local waiting = false; + local function process_next(last_response) + --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 503 Internal Server Error\r\n\r\nAn error occured during the processing of this request."); + 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) + 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; + }; + + if not request.headers.host then + response.status_code = 400; + response.headers.content_type = "text/html"; + response:send("<html><head>400 Bad Request</head><body>400 Bad Request: No Host header.</body></html>"); + else + -- TODO call handler + --response.headers.content_type = "text/plain"; + --response:send("host="..(request.headers.host or "").."\npath="..request.path.."\n"..(request.body or "")); + local host = request.headers.host; + if host then + host = host:match("[^:]*"):lower(); + 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 code, body = 200, ""; + local result_type = type(result); + if result_type == "number" then + response.status_code = result; + elseif result_type == "string" then + body = result; + elseif result_type == "table" then + body = result.body; + result.body = nil; + for k, v in pairs(result) do + response[k] = v; + end + end + response:send(body); + end + return; + end + end + + -- if handler not called, fallback to legacy httpserver handlers + _M.legacy_handler(request, response); + end +end +function _M.send_response(response, body) + local status_line = "HTTP/"..response.request.httpversion.." "..(response.status or codes[response.status_code]); + local headers = response.headers; + body = 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 headers.connection == "Keep-Alive" then + response:finish_cb(); + else + response.conn:close(); + end +end +function _M.legacy_handler(request, response) + log("debug", "Invoking legacy handler"); + local base = request.path:match("^/([^/?]+)"); + local legacy_server = legacy_httpserver and legacy_httpserver.new.http_servers[5280]; + local handler = legacy_server and legacy_server.handlers[base]; + if not handler then handler = legacy_httpserver and legacy_httpserver.set_default_handler.default_handler; end + if handler then + -- add legacy properties to request object + request.url = { path = request.path }; + request.handler = response.conn; + request.id = tostring{}:match("%x+$"); + local headers = {}; + for k,v in pairs(request.headers) do + headers[k:gsub("_", "-")] = v; + end + request.headers = headers; + function request:send(resp) + if self.destroyed then return; end + if resp.body or resp.headers then + if resp.headers then + for k,v in pairs(resp.headers) do response.headers[k] = v; end + end + response:send(resp.body) + else + response:send(resp) + end + self.sent = true; + self:destroy(); + end + function request:destroy() + if self.destroyed then return; end + if not self.sent then return self:send(""); end + self.destroyed = true; + if self.on_destroy then + log("debug", "Request has destroy callback"); + self:on_destroy(); + else + log("debug", "Request has no destroy callback"); + end + end + local r = handler(request.method, request.body, request); + if r ~= true then + request:send(r); + end + else + log("debug", "No handler found"); + response.status_code = 404; + response.headers.content_type = "text/html"; + response:send("<html><head><title>404 Not Found</title></head><body>404 Not Found: No such page.</body></html>"); + 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 + +_M.listener = listener; +_M.codes = codes; +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 44e8e24d..0f5a4186 100644 --- a/net/httpserver.lua +++ b/net/httpserver.lua @@ -10,8 +10,8 @@ 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 connlisteners_start = require "net.connlisteners".start; +--local connlisteners_get = require "net.connlisteners".get; local listener; local t_insert, t_concat = table.insert, table.concat; @@ -164,7 +164,7 @@ end function destroy_request(request) log("debug", "Destroying request %s", request.id); - listener = listener or connlisteners_get("httpserver"); + --listener = listener or connlisteners_get("httpserver"); if not request.destroyed then request.destroyed = true; if request.on_destroy then @@ -186,7 +186,7 @@ function new(params) http_server = { handlers = {} }; http_servers[params.port] = http_server; -- We weren't already listening on this port, so start now - connlisteners_start("httpserver", params); + --connlisteners_start("httpserver", params); end if params.base then http_server.handlers[params.base] = params.handler; 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 dbf5161f..8d6f5597 100644 --- a/net/server_event.lua +++ b/net/server_event.lua @@ -343,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 diff --git a/net/server_select.lua b/net/server_select.lua index 8802f620..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 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 dd7b2b91..00000000 --- a/net/xmppcomponent_listener.lua +++ /dev/null @@ -1,218 +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 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 048c6c7d..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", default_interface = "*" }; - --- 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.type == "s2sout_unauthed" 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 |