From 4c6992a00e582dc2c9c6a7ad721cd7e831202a1e Mon Sep 17 00:00:00 2001 From: Matthew Wild Date: Mon, 1 Jun 2020 15:42:19 +0100 Subject: mod_admin_socket, util.adminstream: New module to manage a local unix domain socket for admin functionality --- plugins/mod_admin_socket.lua | 69 +++++++++++ util/adminstream.lua | 285 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 354 insertions(+) create mode 100644 plugins/mod_admin_socket.lua create mode 100644 util/adminstream.lua diff --git a/plugins/mod_admin_socket.lua b/plugins/mod_admin_socket.lua new file mode 100644 index 00000000..c094aad2 --- /dev/null +++ b/plugins/mod_admin_socket.lua @@ -0,0 +1,69 @@ +module:set_global(); + +local have_unix, unix = pcall(require, "socket.unix"); + +if not have_unix or type(unix) ~= "table" then + module:log_status("error", "LuaSocket unix socket support not available or incompatible, ensure it is up to date"); + return; +end + +local server = require "net.server"; + +local adminstream = require "util.adminstream"; + +local socket_path = module:get_option_string("admin_socket", prosody.paths.data.."/prosody.sock"); + +local sessions = module:shared("sessions"); + +local function fire_admin_event(session, stanza) + local event_data = { + origin = session, stanza = stanza; + }; + local event_name; + if stanza.attr.xmlns then + event_name = "admin/"..stanza.attr.xmlns..":"..stanza.name; + else + event_name = "admin/"..stanza.name; + end + module:log("debug", "Firing %s", event_name); + return module:fire_event(event_name, event_data); +end + +module:hook("server-stopping", function () + for _, session in pairs(sessions) do + session:close("system-shutdown"); + end + os.remove(socket_path); +end); + +--- Unix domain socket management + +local conn, sock; + +local listeners = adminstream.server(sessions, fire_admin_event).listeners; + +local function accept_connection() + module:log("debug", "accepting..."); + local client = sock:accept(); + if not client then return; end + server.wrapclient(client, "unix", 0, listeners, "*a"); +end + +function module.load() + sock = unix.stream(); + sock:settimeout(0); + os.remove(socket_path); + assert(sock:bind(socket_path)); + assert(sock:listen()); + conn = server.watchfd(sock:getfd(), accept_connection); +end + +function module.unload() + if conn then + conn:close(); + end + if sock then + sock:close(); + end + os.remove(socket_path); +end diff --git a/util/adminstream.lua b/util/adminstream.lua new file mode 100644 index 00000000..186cb0e9 --- /dev/null +++ b/util/adminstream.lua @@ -0,0 +1,285 @@ +local st = require "util.stanza"; +local new_xmpp_stream = require "util.xmppstream".new; +local sessionlib = require "util.session"; +local gettime = require "util.time".now; +local runner = require "util.async".runner; +local add_task = require "util.timer".add_task; +local events = require "util.events"; + +local stream_close_timeout = 5; + +local log = require "util.logger".init("adminstream"); + +local xmlns_xmpp_streams = "urn:ietf:params:xml:ns:xmpp-streams"; + +local stream_callbacks = { default_ns = "xmpp:prosody.im/admin" }; + +function stream_callbacks.streamopened(session, attr) + -- run _streamopened in async context + session.thread:run({ stream = "opened", attr = attr }); +end + +function stream_callbacks._streamopened(session, attr) --luacheck: ignore 212/attr + if session.type ~= "client" then + session:open_stream(); + end + session.notopen = nil; +end + +function stream_callbacks.streamclosed(session, attr) + -- run _streamclosed in async context + session.thread:run({ stream = "closed", attr = attr }); +end + +function stream_callbacks._streamclosed(session) + session.log("debug", "Received "); + session:close(false); +end + +function stream_callbacks.error(session, error, data) + if error == "no-stream" then + session.log("debug", "Invalid opening stream header (%s)", (data:gsub("^([^\1]+)\1", "{%1}"))); + session:close("invalid-namespace"); + elseif error == "parse-error" then + session.log("debug", "Client XML parse error: %s", data); + session:close("not-well-formed"); + elseif error == "stream-error" then + local condition, text = "undefined-condition"; + for child in data:childtags(nil, xmlns_xmpp_streams) do + if child.name ~= "text" then + condition = child.name; + else + text = child:get_text(); + end + if condition ~= "undefined-condition" and text then + break; + 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.handlestanza(session, stanza) + session.thread:run(stanza); +end + +local runner_callbacks = {}; + +function runner_callbacks:error(err) + self.data.log("error", "Traceback[c2s]: %s", err); +end + +local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'}; + +local function destroy_session(session, reason) + if session.destroyed then return; end + session.destroyed = true; + session.log("debug", "Destroying session: %s", reason or "unknown reason"); +end + +local function session_close(session, reason) + local log = session.log or log; + if session.conn then + if session.notopen then + session:open_stream(); + end + if reason then -- nil == no err, initiated by us, false == initiated by client + local stream_error = st.stanza("stream:error"); + if type(reason) == "string" then -- assume stream error + stream_error:tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' }); + elseif type(reason) == "table" then + if reason.condition then + stream_error:tag(reason.condition, stream_xmlns_attr):up(); + if reason.text then + stream_error:tag("text", stream_xmlns_attr):text(reason.text):up(); + end + if reason.extra then + stream_error:add_child(reason.extra); + end + elseif reason.name then -- a stanza + stream_error = reason; + end + end + stream_error = tostring(stream_error); + log("debug", "Disconnecting client, is: %s", stream_error); + session.send(stream_error); + end + + session.send(""); + function session.send() return false; end + + local reason_text = (reason and (reason.name or reason.text or reason.condition)) or reason; + session.log("debug", "c2s stream for %s closed: %s", session.full_jid or session.ip or "", reason_text or "session closed"); + + -- Authenticated incoming stream may still be sending us stanzas, so wait for from remote + local conn = session.conn; + if reason_text == nil and not session.notopen and session.type == "c2s" then + -- Grace time to process data from authenticated cleanly-closed stream + add_task(stream_close_timeout, function () + if not session.destroyed then + session.log("warn", "Failed to receive a stream close response, closing connection anyway..."); + destroy_session(session); + conn:close(); + end + end); + else + destroy_session(session, reason_text); + conn:close(); + end + else + local reason_text = (reason and (reason.name or reason.text or reason.condition)) or reason; + destroy_session(session, reason_text); + end +end + +--- Public methods + +local function new_server(sessions, stanza_handler) + local listeners = {}; + + function listeners.onconnect(conn) + log("debug", "New connection"); + local session = sessionlib.new("admin"); + sessionlib.set_id(session); + sessionlib.set_logger(session); + sessionlib.set_conn(session, conn); + + session.conntime = gettime(); + session.type = "admin"; + + local stream = new_xmpp_stream(session, stream_callbacks); + session.stream = stream; + session.notopen = true; + + session.thread = runner(function (stanza) + if st.is_stanza(stanza) then + stanza_handler(session, stanza); + elseif stanza.stream == "opened" then + stream_callbacks._streamopened(session, stanza.attr); + elseif stanza.stream == "closed" then + stream_callbacks._streamclosed(session, stanza.attr); + end + end, runner_callbacks, session); + + function session.data(data) + -- Parse the data, which will store stanzas in session.pending_stanzas + if data then + local ok, err = stream:feed(data); + if not ok then + session.log("debug", "Received invalid XML (%s) %d bytes: %q", err, #data, data:sub(1, 300)); + session:close("not-well-formed"); + end + end + end + + session.close = session_close; + + session.send = function (t) + session.log("debug", "Sending[%s]: %s", session.type, t.top_tag and t:top_tag() or t:match("^[^>]*>?")); + return session.rawsend(tostring(t)); + end + + function session.rawsend(t) + local ret, err = conn:write(t); + if not ret then + session.log("debug", "Error writing to connection: %s", err); + return false, err; + end + return true; + end + + sessions[conn] = session; + end + + function listeners.onincoming(conn, data) + local session = sessions[conn]; + if session then + session.data(data); + end + end + + function listeners.ondisconnect(conn, err) + local session = sessions[conn]; + if session then + session.log("info", "Admin client disconnected: %s", err or "connection closed"); + session.conn = nil; + sessions[conn] = nil; + end + end + return { + listeners = listeners; + }; +end + +local function new_client() + local client = { + type = "client"; + events = events.new(); + log = log; + }; + + local listeners = {}; + + function listeners.onconnect(conn) + log("debug", "Connected"); + client.conn = conn; + + local stream = new_xmpp_stream(client, stream_callbacks); + client.stream = stream; + client.notopen = true; + + client.thread = runner(function (stanza) + if st.is_stanza(stanza) then + client.events.fire_event("received", stanza); + elseif stanza.stream == "opened" then + stream_callbacks._streamopened(client, stanza.attr); + client.events.fire_event("connected"); + elseif stanza.stream == "closed" then + client.events.fire_event("disconnected"); + stream_callbacks._streamclosed(client, stanza.attr); + end + end, runner_callbacks, client); + + client.close = session_close; + + function client.send(t) + client.log("debug", "Sending: %s", t.top_tag and t:top_tag() or t:match("^[^>]*>?")); + return client.rawsend(tostring(t)); + end + + function client.rawsend(t) + local ret, err = conn:write(t); + if not ret then + client.log("debug", "Error writing to connection: %s", err); + return false, err; + end + return true; + end + client.log("debug", "Opening stream..."); + client:open_stream(); + end + + function listeners.onincoming(conn, data) --luacheck: ignore 212/conn + local ok, err = client.stream:feed(data); + if not ok then + client.log("debug", "Received invalid XML (%s) %d bytes: %q", err, #data, data:sub(1, 300)); + client:close("not-well-formed"); + end + end + + function listeners.ondisconnect(conn, err) --luacheck: ignore 212/conn + client.log("info", "Admin client disconnected: %s", err or "connection closed"); + client.conn = nil; + end + + client.listeners = listeners; + + return client; +end + +return { + server = new_server; + client = new_client; +}; -- cgit v1.2.3