# HG changeset patch # User Matthew Wild # Date 1327276633 0 # Node ID db27a4c18b6a900585a722db4acf2fb634724ef8 # Parent 50aca1e0bfbd54e05604f4323d7156d3cfc5b019 mod_c2s, sessionmanager, xmppclient_listener: Move all c2s network and stream logic into a new module, mod_c2s diff -r 50aca1e0bfbd -r db27a4c18b6a core/sessionmanager.lua --- a/core/sessionmanager.lua Sun Jan 22 23:55:48 2012 +0000 +++ b/core/sessionmanager.lua Sun Jan 22 23:57:13 2012 +0000 @@ -6,11 +6,8 @@ -- COPYING file in the source package for more information. -- - - local tonumber, tostring, setmetatable = tonumber, tostring, setmetatable; local ipairs, pairs, print, next= ipairs, pairs, print, next; -local format = string.format; local hosts = hosts; local full_sessions = full_sessions; @@ -19,10 +16,8 @@ local logger = require "util.logger"; local log = logger.init("sessionmanager"); local error = error; -local uuid_generate = require "util.uuid".generate; local rm_load_roster = require "core.rostermanager".load_roster; local config_get = require "core.configmanager".get; -local nameprep = require "util.encodings".stringprep.nameprep; local resourceprep = require "util.encodings".stringprep.resourceprep; local nodeprep = require "util.encodings".stringprep.nodeprep; @@ -216,50 +211,6 @@ return true; end -function streamopened(session, attr) - local send = session.send; - session.host = attr.to; - if not session.host then - session:close{ condition = "improper-addressing", - text = "A 'to' attribute is required on stream headers" }; - return; - end - session.host = nameprep(session.host); - session.version = tonumber(attr.version) or 0; - session.streamid = uuid_generate(); - (session.log or session)("debug", "Client sent opening to %s", session.host); - - if not hosts[session.host] then - -- We don't serve this host... - session:close{ condition = "host-unknown", text = "This server does not serve "..tostring(session.host)}; - return; - end - - send(""); - send(format("", session.streamid, session.host)); - - (session.log or log)("debug", "Sent reply to client"); - session.notopen = nil; - - -- If session.secure is *false* (not nil) then it means we /were/ encrypting - -- since we now have a new stream header, session is secured - if session.secure == false then - session.secure = true; - end - - local features = st.stanza("stream:features"); - hosts[session.host].events.fire_event("stream-features", { origin = session, features = features }); - fire_event("stream-features", session, features); - - send(features); - -end - -function streamclosed(session) - session.log("debug", "Received "); - session:close(); -end - function send_to_available_resources(user, host, stanza) local jid = user.."@"..host; local count = 0; diff -r 50aca1e0bfbd -r db27a4c18b6a net/xmppclient_listener.lua --- a/net/xmppclient_listener.lua Sun Jan 22 23:55:48 2012 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -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(""); - 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, 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, is: %s", tostring(stanza)); - session.send(stanza); - elseif reason.name then -- a stanza - log("info", "Disconnecting client, is: %s", tostring(reason)); - session.send(reason); - end - end - end - session.send(""); - 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 -r 50aca1e0bfbd -r db27a4c18b6a plugins/mod_c2s.lua --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/plugins/mod_c2s.lua Sun Jan 22 23:57:13 2012 +0000 @@ -0,0 +1,224 @@ +-- 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. +-- + +module:set_global(); + +local new_xmpp_stream = require "util.xmppstream".new; +local nameprep = require "util.encodings".stringprep.nameprep; +local portmanager = require "core.portmanager"; +local sessionmanager = require "core.sessionmanager"; +local st = require "util.stanza"; +local sm_new_session, sm_destroy_session = sessionmanager.new_session, sessionmanager.destroy_session; +local uuid_generate = require "util.uuid".generate; + +local xpcall, tostring, type = xpcall, tostring, type; +local format = string.format; +local traceback = debug.traceback; + +local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; + +local log = module._log; + +local opt_keepalives = module:get_option_boolean("tcp_keepalives", false); + +local sessions = module:shared("sessions"); + +local stream_callbacks = { default_ns = "jabber:client", handlestanza = core_process_stanza }; +local listener = { default_port = 5222, default_mode = "*a" }; + +--- Stream events handlers +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 = "" }; + +function stream_callbacks.streamopened(session, attr) + local send = session.send; + session.host = attr.to; + if not session.host then + session:close{ condition = "improper-addressing", + text = "A 'to' attribute is required on stream headers" }; + return; + end + session.host = nameprep(session.host); + session.version = tonumber(attr.version) or 0; + session.streamid = uuid_generate(); + (session.log or session)("debug", "Client sent opening to %s", session.host); + + if not hosts[session.host] then + -- We don't serve this host... + session:close{ condition = "host-unknown", text = "This server does not serve "..tostring(session.host)}; + return; + end + + send(""); + send(format("", session.streamid, session.host)); + + (session.log or log)("debug", "Sent reply to client"); + session.notopen = nil; + + -- If session.secure is *false* (not nil) then it means we /were/ encrypting + -- since we now have a new stream header, session is secured + if session.secure == false then + session.secure = true; + end + + local features = st.stanza("stream:features"); + hosts[session.host].events.fire_event("stream-features", { origin = session, features = features }); + module:fire_event("stream-features", session, features); + + send(features); +end + +function stream_callbacks.streamclosed(session) + session.log("debug", "Received "); + session:close(); +end + +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 + +--- Session methods +local function session_close(session, reason) + local log = session.log or log; + if session.conn then + if session.notopen then + session.send(""); + 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, 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, is: %s", tostring(stanza)); + session.send(stanza); + elseif reason.name then -- a stanza + log("info", "Disconnecting client, is: %s", tostring(reason)); + session.send(reason); + end + end + end + session.send(""); + session.conn:close(); + listener.ondisconnect(session.conn, (reason and (reason.text or reason.condition)) or reason or "session closed"); + end +end + +--- Port listener +function listener.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 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 + + session.dispatch_stanza = stream_callbacks.handlestanza; +end + +function listener.onincoming(conn, data) + local session = sessions[conn]; + if session then + session.data(data); + end +end + +function listener.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 listener.associate_session(conn, session) + sessions[conn] = session; +end + +portmanager.register_service("c2s", { + listener = listener; + default_port = 5222; + encryption = "starttls"; +}); + +portmanager.register_service("legacy_ssl", { + listener = listener; + encryption = "ssl"; +}); + +