plugins/mod_c2s.lua

   1 -- Prosody IM
   2 -- Copyright (C) 2008-2010 Matthew Wild
   3 -- Copyright (C) 2008-2010 Waqas Hussain
   4 --
   5 -- This project is MIT/X11 licensed. Please see the
   6 -- COPYING file in the source package for more information.
   7 --
   8
   9 module:set_global();
  10
  11 local add_task = require "util.timer".add_task;
  12 local new_xmpp_stream = require "util.xmppstream".new;
  13 local nameprep = require "util.encodings".stringprep.nameprep;
  14 local sessionmanager = require "core.sessionmanager";
  15 local st = require "util.stanza";
  16 local sm_new_session, sm_destroy_session = sessionmanager.new_session, sessionmanager.destroy_session;
  17 local uuid_generate = require "util.uuid".generate;
  18
  19 local xpcall, tostring, type = xpcall, tostring, type;
  20 local traceback = debug.traceback;
  21
  22 local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams";
  23
  24 local log = module._log;
  25
  26 local c2s_timeout = module:get_option_number("c2s_timeout");
  27 local stream_close_timeout = module:get_option_number("c2s_close_timeout", 5);
  28 local opt_keepalives = module:get_option_boolean("tcp_keepalives", false);
  29
  30 local sessions = module:shared("sessions");
  31 local core_process_stanza = prosody.core_process_stanza;
  32
  33 local stream_callbacks = { default_ns = "jabber:client", handlestanza = core_process_stanza };
  34 local listener = {};
  35
  36 --- Stream events handlers
  37 local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
  38 local default_stream_attr = { ["xmlns:stream"] = "http://etherx.jabber.org/streams", xmlns = stream_callbacks.default_ns, version = "1.0", id = "" };
  39
  40 function stream_callbacks.streamopened(session, attr)
  41         local send = session.send;
  42         session.host = nameprep(attr.to);
  43         if not session.host then
  44                 session:close{ condition = "improper-addressing",
  45                         text = "A valid 'to' attribute is required on stream headers" };
  46                 return;
  47         end
  48         session.version = tonumber(attr.version) or 0;
  49         session.streamid = uuid_generate();
  50         (session.log or session)("debug", "Client sent opening <stream:stream> to %s", session.host);
  51
  52         if not hosts[session.host] then
  53                 -- We don't serve this host...
  54                 session:close{ condition = "host-unknown", text = "This server does not serve "..tostring(session.host)};
  55                 return;
  56         end
  57
  58         send("<?xml version='1.0'?>"..st.stanza("stream:stream", {
  59                 xmlns = 'jabber:client', ["xmlns:stream"] = 'http://etherx.jabber.org/streams';
  60                 id = session.streamid, from = session.host, version = '1.0', ["xml:lang"] = 'en' }):top_tag());
  61
  62         (session.log or log)("debug", "Sent reply <stream:stream> to client");
  63         session.notopen = nil;
  64
  65         -- If session.secure is *false* (not nil) then it means we /were/ encrypting
  66         -- since we now have a new stream header, session is secured
  67         if session.secure == false then
  68                 session.secure = true;
  69         end
  70
  71         local features = st.stanza("stream:features");
  72         hosts[session.host].events.fire_event("stream-features", { origin = session, features = features });
  73         module:fire_event("stream-features", session, features);
  74
  75         send(features);
  76 end
  77
  78 function stream_callbacks.streamclosed(session)
  79         session.log("debug", "Received </stream:stream>");
  80         session:close(false);
  81 end
  82
  83 function stream_callbacks.error(session, error, data)
  84         if error == "no-stream" then
  85                 session.log("debug", "Invalid opening stream header");
  86                 session:close("invalid-namespace");
  87         elseif error == "parse-error" then
  88                 (session.log or log)("debug", "Client XML parse error: %s", tostring(data));
  89                 session:close("not-well-formed");
  90         elseif error == "stream-error" then
  91                 local condition, text = "undefined-condition";
  92                 for child in data:children() do
  93                         if child.attr.xmlns == xmlns_xmpp_streams then
  94                                 if child.name ~= "text" then
  95                                         condition = child.name;
  96                                 else
  97                                         text = child:get_text();
  98                                 end
  99                                 if condition ~= "undefined-condition" and text then
 100                                         break;
 101                                 end
 102                         end
 103                 end
 104                 text = condition .. (text and (" ("..text..")") or "");
 105                 session.log("info", "Session closed by remote with error: %s", text);
 106                 session:close(nil, text);
 107         end
 108 end
 109
 110 local function handleerr(err) log("error", "Traceback[c2s]: %s: %s", tostring(err), traceback()); end
 111 function stream_callbacks.handlestanza(session, stanza)
 112         stanza = session.filter("stanzas/in", stanza);
 113         if stanza then
 114                 return xpcall(function () return core_process_stanza(session, stanza) end, handleerr);
 115         end
 116 end
 117
 118 --- Session methods
 119 local function session_close(session, reason)
 120         local log = session.log or log;
 121         if session.conn then
 122                 if session.notopen then
 123                         session.send("<?xml version='1.0'?>");
 124                         session.send(st.stanza("stream:stream", default_stream_attr):top_tag());
 125                 end
 126                 if reason then -- nil == no err, initiated by us, false == initiated by client
 127                         if type(reason) == "string" then -- assume stream error
 128                                 log("debug", "Disconnecting client, <stream:error> is: %s", reason);
 129                                 session.send(st.stanza("stream:error"):tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' }));
 130                         elseif type(reason) == "table" then
 131                                 if reason.condition then
 132                                         local stanza = st.stanza("stream:error"):tag(reason.condition, stream_xmlns_attr):up();
 133                                         if reason.text then
 134                                                 stanza:tag("text", stream_xmlns_attr):text(reason.text):up();
 135                                         end
 136                                         if reason.extra then
 137                                                 stanza:add_child(reason.extra);
 138                                         end
 139                                         log("debug", "Disconnecting client, <stream:error> is: %s", tostring(stanza));
 140                                         session.send(stanza);
 141                                 elseif reason.name then -- a stanza
 142                                         log("debug", "Disconnecting client, <stream:error> is: %s", tostring(reason));
 143                                         session.send(reason);
 144                                 end
 145                         end
 146                 end
 147
 148                 session.send("</stream:stream>");
 149                 function session.send() return false; end
 150
 151                 local reason = (reason and (reason.text or reason.condition)) or reason;
 152                 session.log("info", "c2s stream for %s closed: %s", session.full_jid or ("<"..session.ip..">"), reason or "session closed");
 153
 154                 -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote
 155                 local conn = session.conn;
 156                 if reason == nil and not session.notopen and session.type == "c2s" then
 157                         -- Grace time to process data from authenticated cleanly-closed stream
 158                         add_task(stream_close_timeout, function ()
 159                                 if not session.destroyed then
 160                                         session.log("warn", "Failed to receive a stream close response, closing connection anyway...");
 161                                         sm_destroy_session(session, reason);
 162                                         conn:close();
 163                                 end
 164                         end);
 165                 else
 166                         sm_destroy_session(session, reason);
 167                         conn:close();
 168                 end
 169         end
 170 end
 171
 172 --- Port listener
 173 function listener.onconnect(conn)
 174         local session = sm_new_session(conn);
 175         sessions[conn] = session;
 176
 177         session.log("info", "Client connected");
 178
 179         -- Client is using legacy SSL (otherwise mod_tls sets this flag)
 180         if conn:ssl() then
 181                 session.secure = true;
 182         end
 183
 184         if opt_keepalives then
 185                 conn:setoption("keepalive", opt_keepalives);
 186         end
 187
 188         session.close = session_close;
 189
 190         local stream = new_xmpp_stream(session, stream_callbacks);
 191         session.stream = stream;
 192         session.notopen = true;
 193
 194         function session.reset_stream()
 195                 session.notopen = true;
 196                 session.stream:reset();
 197         end
 198
 199         local filter = session.filter;
 200         function session.data(data)
 201                 data = filter("bytes/in", data);
 202                 if data then
 203                         local ok, err = stream:feed(data);
 204                         if ok then return; end
 205                         log("debug", "Received invalid XML (%s) %d bytes: %s", tostring(err), #data, data:sub(1, 300):gsub("[\r\n]+", " "):gsub("[%z\1-\31]", "_"));
 206                         session:close("not-well-formed");
 207                 end
 208         end
 209
 210
 211         if c2s_timeout then
 212                 add_task(c2s_timeout, function ()
 213                         if session.type == "c2s_unauthed" then
 214                                 session:close("connection-timeout");
 215                         end
 216                 end);
 217         end
 218
 219         session.dispatch_stanza = stream_callbacks.handlestanza;
 220 end
 221
 222 function listener.onincoming(conn, data)
 223         local session = sessions[conn];
 224         if session then
 225                 session.data(data);
 226         end
 227 end
 228
 229 function listener.ondisconnect(conn, err)
 230         local session = sessions[conn];
 231         if session then
 232                 (session.log or log)("info", "Client disconnected: %s", err or "connection closed");
 233                 sm_destroy_session(session, err);
 234                 sessions[conn]  = nil;
 235         end
 236 end
 237
 238 function listener.associate_session(conn, session)
 239         sessions[conn] = session;
 240 end
 241
 242 module:add_item("net-provider", {
 243         name = "c2s";
 244         listener = listener;
 245         default_port = 5222;
 246         encryption = "starttls";
 247         multiplex = {
 248                 pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>";
 249         };
 250 });
 251
 252 module:add_item("net-provider", {
 253         name = "legacy_ssl";
 254         listener = listener;
 255         encryption = "ssl";
 256         multiplex = {
 257                 pattern = "^<.*:stream.*%sxmlns%s*=%s*(['\"])jabber:client%1.*>";
 258         };
 259 });
 260
 261