diff options
Diffstat (limited to 'plugins/mod_pubsub')
| -rw-r--r-- | plugins/mod_pubsub/mod_pubsub.lua | 114 | ||||
| -rw-r--r-- | plugins/mod_pubsub/pubsub.lib.lua | 328 |
2 files changed, 371 insertions, 71 deletions
diff --git a/plugins/mod_pubsub/mod_pubsub.lua b/plugins/mod_pubsub/mod_pubsub.lua index 8e7bfc53..82c787aa 100644 --- a/plugins/mod_pubsub/mod_pubsub.lua +++ b/plugins/mod_pubsub/mod_pubsub.lua @@ -2,6 +2,7 @@ local pubsub = require "util.pubsub"; local st = require "util.stanza"; local jid_bare = require "util.jid".bare; local usermanager = require "core.usermanager"; +local new_id = require "util.id".medium; local xmlns_pubsub = "http://jabber.org/protocol/pubsub"; local xmlns_pubsub_event = "http://jabber.org/protocol/pubsub#event"; @@ -12,32 +13,35 @@ local autocreate_on_subscribe = module:get_option_boolean("autocreate_on_subscri local pubsub_disco_name = module:get_option_string("name", "Prosody PubSub Service"); local expose_publisher = module:get_option_boolean("expose_publisher", false) +local enable_persistence = module:get_option_boolean("experimental_pubsub_item_persistence", false); + local service; local lib_pubsub = module:require "pubsub"; -local handlers = lib_pubsub.handlers; -local pubsub_error_reply = lib_pubsub.pubsub_error_reply; module:depends("disco"); module:add_identity("pubsub", "service", pubsub_disco_name); module:add_feature("http://jabber.org/protocol/pubsub"); function handle_pubsub_iq(event) - local origin, stanza = event.origin, event.stanza; - local pubsub = stanza.tags[1]; - local action = pubsub.tags[1]; - if not action then - origin.send(st.error_reply(stanza, "cancel", "bad-request")); - return true; - end - local handler = handlers[stanza.attr.type.."_"..action.name]; - if handler then - handler(origin, stanza, action, service); - return true; - end + return lib_pubsub.handle_pubsub_iq(event, service); +end + +local node_store = module:open_store(module.name.."_nodes"); + +local function create_simple_itemstore(node_config, node_name) + local archive = module:open_store("pubsub_"..node_name, "archive"); + return lib_pubsub.archive_itemstore(archive, node_config, nil, node_name); +end + +if enable_persistence then + module:log("warn", "Item persistence is an experimental feature. Note that ownership information is lost on restart.") +else + create_simple_itemstore = nil; end -function simple_broadcast(kind, node, jids, item, actor) + +function simple_broadcast(kind, node, jids, item, actor, node_obj) if item then item = st.clone(item); item.attr.xmlns = nil; -- Clear the pubsub namespace @@ -45,52 +49,55 @@ function simple_broadcast(kind, node, jids, item, actor) item.attr.publisher = actor end end - local message = st.message({ from = module.host, type = "headline" }) + + local id = new_id(); + local msg_type = node_obj and node_obj.config.message_type or "headline"; + local message = st.message({ from = module.host, type = msg_type, id = id }) :tag("event", { xmlns = xmlns_pubsub_event }) :tag(kind, { node = node }) - :add_child(item); - for jid in pairs(jids) do - module:log("debug", "Sending notification to %s", jid); - message.attr.to = jid; - module:send(message); - end -end -module:hook("iq/host/"..xmlns_pubsub..":pubsub", handle_pubsub_iq); -module:hook("iq/host/"..xmlns_pubsub_owner..":pubsub", handle_pubsub_iq); + if item then + message:add_child(item); + end -local feature_map = { - create = { "create-nodes", "instant-nodes", "item-ids" }; - retract = { "delete-items", "retract-items" }; - purge = { "purge-nodes" }; - publish = { "publish", autocreate_on_publish and "auto-create" }; - delete = { "delete-nodes" }; - get_items = { "retrieve-items" }; - add_subscription = { "subscribe" }; - get_subscriptions = { "retrieve-subscriptions" }; - set_configure = { "config-node" }; - get_default = { "retrieve-default" }; -}; - -local function add_disco_features_from_service(service) - for method, features in pairs(feature_map) do - if service[method] then - for _, feature in ipairs(features) do - if feature then - module:add_feature(xmlns_pubsub.."#"..feature); + -- Compose a sensible textual representation of at least Atom payloads + if node_obj and item and node_obj.config.include_body and item.tags[1] then + local payload = item.tags[1]; + if payload.attr.xmlns == "http://www.w3.org/2005/Atom" then + message:reset(); + local title = payload:get_child_text("title"); + local summary = payload:get_child_text("summary"); + if not summary and title then + local author = payload:find("author/name#"); + summary = title; + if author then + summary = author .. " posted " .. summary; end end + if summary then + message:body(summary); + end end end - for affiliation in pairs(service.config.capabilities) do - if affiliation ~= "none" and affiliation ~= "owner" then - module:add_feature(xmlns_pubsub.."#"..affiliation.."-affiliation"); - end + + module:broadcast(jids, message, pairs); +end + +function is_item_stanza(item) + return st.is_stanza(item) and item.attr.xmlns == xmlns_pubsub and item.name == "item"; +end + +module:hook("iq/host/"..xmlns_pubsub..":pubsub", handle_pubsub_iq); +module:hook("iq/host/"..xmlns_pubsub_owner..":pubsub", handle_pubsub_iq); + +local function add_disco_features_from_service(service) --luacheck: ignore 431/service + for feature in lib_pubsub.get_feature_set(service) do + module:add_feature(xmlns_pubsub.."#"..feature); end end module:hook("host-disco-info-node", function (event) - local stanza, origin, reply, node = event.stanza, event.origin, event.reply, event.node; + local stanza, reply, node = event.stanza, event.reply, event.node; local ok, ret = service:get_nodes(stanza.attr.from); if not ok or not ret[node] then return; @@ -100,7 +107,7 @@ module:hook("host-disco-info-node", function (event) end); module:hook("host-disco-items-node", function (event) - local stanza, origin, reply, node = event.stanza, event.origin, event.reply, event.node; + local stanza, reply, node = event.stanza, event.reply, event.node; local ok, ret = service:get_items(node, stanza.attr.from); if not ok then return; @@ -114,8 +121,8 @@ end); module:hook("host-disco-items", function (event) - local stanza, origin, reply = event.stanza, event.origin, event.reply; - local ok, ret = service:get_nodes(event.stanza.attr.from); + local stanza, reply = event.stanza, event.reply; + local ok, ret = service:get_nodes(stanza.attr.from); if not ok then return; end @@ -225,7 +232,10 @@ function module.load() autocreate_on_publish = autocreate_on_publish; autocreate_on_subscribe = autocreate_on_subscribe; + nodestore = node_store; + itemstore = create_simple_itemstore; broadcaster = simple_broadcast; + itemcheck = is_item_stanza; get_affiliation = get_affiliation; normalize_jid = jid_bare; diff --git a/plugins/mod_pubsub/pubsub.lib.lua b/plugins/mod_pubsub/pubsub.lib.lua index 1497c21c..2ec6e8de 100644 --- a/plugins/mod_pubsub/pubsub.lib.lua +++ b/plugins/mod_pubsub/pubsub.lib.lua @@ -1,4 +1,10 @@ +local t_unpack = table.unpack or unpack; -- luacheck: ignore 113 +local time_now = os.time; + +local jid_prep = require "util.jid".prep; +local set = require "util.set"; local st = require "util.stanza"; +local it = require "util.iterators"; local uuid_generate = require "util.uuid".generate; local dataform = require"util.dataforms".new; @@ -23,7 +29,7 @@ local pubsub_errors = { }; local function pubsub_error_reply(stanza, error) local e = pubsub_errors[error]; - local reply = st.error_reply(stanza, unpack(e, 1, 3)); + local reply = st.error_reply(stanza, t_unpack(e, 1, 3)); if e[4] then reply:tag(e[4], { xmlns = xmlns_pubsub_errors }):up(); end @@ -31,7 +37,7 @@ local function pubsub_error_reply(stanza, error) end _M.pubsub_error_reply = pubsub_error_reply; -local node_config_form = require"util.dataforms".new { +local node_config_form = dataform { { type = "hidden"; name = "FORM_TYPE"; @@ -42,18 +48,113 @@ local node_config_form = require"util.dataforms".new { name = "pubsub#max_items"; label = "Max # of items to persist"; }; + { + type = "boolean"; + name = "pubsub#persist_items"; + label = "Persist items to storage"; + }; + { + type = "boolean"; + name = "pubsub#include_body"; + label = "Receive message body in addition to payload?"; + }; + { + type = "list-single"; + name = "pubsub#notification_type"; + label = "Specify the delivery style for notifications"; + options = { + { label = "Messages of type normal", value = "normal" }, + { label = "Messages of type headline", value = "headline", default = true }, + }; + }; +}; + +local options_form = dataform { + { + type = "hidden"; + name = "FORM_TYPE"; + value = "http://jabber.org/protocol/pubsub#subscribe_options"; + } + -- No options yet. File a feature request ;) +}; + +local service_method_feature_map = { + add_subscription = { "subscribe" }; + create = { "create-nodes", "instant-nodes", "item-ids", "create-and-configure" }; + delete = { "delete-nodes" }; + get_items = { "retrieve-items" }; + get_subscriptions = { "retrieve-subscriptions" }; + node_defaults = { "retrieve-default" }; + publish = { "publish" }; + purge = { "purge-nodes" }; + retract = { "delete-items", "retract-items" }; + set_node_config = { "config-node" }; + set_affiliation = { "modify-affiliations" }; }; +local service_config_feature_map = { + autocreate_on_publish = { "auto-create" }; +}; + +function _M.get_feature_set(service) + local supported_features = set.new(); + + for method, features in pairs(service_method_feature_map) do + if service[method] then + for _, feature in ipairs(features) do + if feature then + supported_features:add(feature); + end + end + end + end + + for option, features in pairs(service_config_feature_map) do + if service.config[option] then + for _, feature in ipairs(features) do + if feature then + supported_features:add(feature); + end + end + end + end + + for affiliation in pairs(service.config.capabilities) do + if affiliation ~= "none" and affiliation ~= "owner" then + supported_features:add(affiliation.."-affiliation"); + end + end + + return supported_features; +end + +function _M.handle_pubsub_iq(event, service) + local origin, stanza = event.origin, event.stanza; + local pubsub_tag = stanza.tags[1]; + local action = pubsub_tag.tags[1]; + if not action then + return origin.send(st.error_reply(stanza, "cancel", "bad-request")); + end + local prefix = ""; + if pubsub_tag.attr.xmlns == xmlns_pubsub_owner then + prefix = "owner_"; + end + local handler = handlers[prefix..stanza.attr.type.."_"..action.name]; + if handler then + handler(origin, stanza, action, service); + return true; + end +end function handlers.get_items(origin, stanza, items, service) local node = items.attr.node; local item = items:get_child("item"); - local id = item and item.attr.id; + local item_id = item and item.attr.id; if not node then origin.send(pubsub_error_reply(stanza, "nodeid-required")); return true; end - local ok, results = service:get_items(node, stanza.attr.from, id); + local ok, results = service:get_items(node, stanza.attr.from, item_id); if not ok then origin.send(pubsub_error_reply(stanza, results)); return true; @@ -95,8 +196,28 @@ end function handlers.set_create(origin, stanza, create, service) local node = create.attr.node; local ok, ret, reply; + local config; + local configure = stanza.tags[1]:get_child("configure"); + if configure then + local config_form = configure:get_child("x", "jabber:x:data"); + if not config_form then + origin.send(st.error_reply(stanza, "modify", "bad-request", "Missing dataform")); + return true; + end + local form_data, err = node_config_form:data(config_form); + if not form_data then + origin.send(st.error_reply(stanza, "modify", "bad-request", err)); + return true; + end + config = { + ["max_items"] = tonumber(form_data["pubsub#max_items"]); + ["persist_items"] = form_data["pubsub#persist_items"]; + ["notification_type"] = form_data["pubsub#notification_type"]; + ["include_body"] = form_data["pubsub#include_body"]; + }; + end if node then - ok, ret = service:create(node, stanza.attr.from); + ok, ret = service:create(node, stanza.attr.from, config); if ok then reply = st.reply(stanza); else @@ -105,7 +226,7 @@ function handlers.set_create(origin, stanza, create, service) else repeat node = uuid_generate(); - ok, ret = service:create(node, stanza.attr.from); + ok, ret = service:create(node, stanza.attr.from, config); until ok or ret ~= "conflict"; if ok then reply = st.reply(stanza) @@ -119,10 +240,10 @@ function handlers.set_create(origin, stanza, create, service) return true; end -function handlers.set_delete(origin, stanza, delete, service) +function handlers.owner_set_delete(origin, stanza, delete, service) local node = delete.attr.node; - local reply, notifier; + local reply; if not node then origin.send(pubsub_error_reply(stanza, "nodeid-required")); return true; @@ -139,17 +260,15 @@ end function handlers.set_subscribe(origin, stanza, subscribe, service) local node, jid = subscribe.attr.node, subscribe.attr.jid; + jid = jid_prep(jid); if not (node and jid) then origin.send(pubsub_error_reply(stanza, jid and "nodeid-required" or "invalid-jid")); return true; end - --[[ local options_tag, options = stanza.tags[1]:get_child("options"), nil; if options_tag then options = options_form:data(options_tag.tags[1]); end - --]] - local options_tag, options; -- FIXME local ok, ret = service:add_subscription(node, stanza.attr.from, jid, options); local reply; if ok then @@ -171,6 +290,7 @@ end function handlers.set_unsubscribe(origin, stanza, unsubscribe, service) local node, jid = unsubscribe.attr.node, unsubscribe.attr.jid; + jid = jid_prep(jid); if not (node and jid) then origin.send(pubsub_error_reply(stanza, jid and "nodeid-required" or "invalid-jid")); return true; @@ -203,6 +323,9 @@ function handlers.set_publish(origin, stanza, publish, service) local ok, ret = service:publish(node, stanza.attr.from, id, item); local reply; if ok then + if type(ok) == "string" then + id = ok; + end reply = st.reply(stanza) :tag("pubsub", { xmlns = xmlns_pubsub }) :tag("publish", { node = node }) @@ -237,7 +360,7 @@ function handlers.set_retract(origin, stanza, retract, service) return true; end -function handlers.set_purge(origin, stanza, purge, service) +function handlers.owner_set_purge(origin, stanza, purge, service) local node, notify = purge.attr.node, purge.attr.notify; notify = (notify == "1") or (notify == "true"); local reply; @@ -255,7 +378,7 @@ function handlers.set_purge(origin, stanza, purge, service) return true; end -function handlers.get_configure(origin, stanza, config, service) +function handlers.owner_get_configure(origin, stanza, config, service) local node = config.attr.node; if not node then origin.send(pubsub_error_reply(stanza, "nodeid-required")); @@ -273,15 +396,22 @@ function handlers.get_configure(origin, stanza, config, service) return true; end + local node_config = node_obj.config; + local pubsub_form_data = { + ["pubsub#max_items"] = tostring(node_config["max_items"]); + ["pubsub#persist_items"] = node_config["persist_items"]; + ["pubsub#notification_type"] = node_config["notification_type"]; + ["pubsub#include_body"] = node_config["include_body"]; + } local reply = st.reply(stanza) :tag("pubsub", { xmlns = xmlns_pubsub_owner }) :tag("configure", { node = node }) - :add_child(node_config_form:form(node_obj.config)); + :add_child(node_config_form:form(pubsub_form_data)); origin.send(reply); return true; end -function handlers.set_configure(origin, stanza, config, service) +function handlers.owner_set_configure(origin, stanza, config, service) local node = config.attr.node; if not node then origin.send(pubsub_error_reply(stanza, "nodeid-required")); @@ -291,11 +421,20 @@ function handlers.set_configure(origin, stanza, config, service) origin.send(pubsub_error_reply(stanza, "forbidden")); return true; end - local new_config, err = node_config_form:data(config.tags[1]); - if not new_config then + local config_form = config:get_child("x", "jabber:x:data"); + if not config_form then + origin.send(st.error_reply(stanza, "modify", "bad-request", "Missing dataform")); + return true; + end + local form_data, err = node_config_form:data(config_form); + if not form_data then origin.send(st.error_reply(stanza, "modify", "bad-request", err)); return true; end + local new_config = { + ["max_items"] = tonumber(form_data["pubsub#max_items"]); + ["persist_items"] = form_data["pubsub#persist_items"]; + }; local ok, err = service:set_node_config(node, stanza.attr.from, new_config); if not ok then origin.send(pubsub_error_reply(stanza, err)); @@ -305,13 +444,164 @@ function handlers.set_configure(origin, stanza, config, service) return true; end -function handlers.get_default(origin, stanza, default, service) +function handlers.owner_get_default(origin, stanza, default, service) -- luacheck: ignore 212/default + local pubsub_form_data = { + ["pubsub#max_items"] = tostring(service.node_defaults["max_items"]); + ["pubsub#persist_items"] = service.node_defaults["persist_items"] + } local reply = st.reply(stanza) :tag("pubsub", { xmlns = xmlns_pubsub_owner }) :tag("default") - :add_child(node_config_form:form(service.node_defaults)); + :add_child(node_config_form:form(pubsub_form_data)); + origin.send(reply); + return true; +end + +function handlers.owner_get_affiliations(origin, stanza, affiliations, service) + local node = affiliations.attr.node; + if not node then + origin.send(pubsub_error_reply(stanza, "nodeid-required")); + return true; + end + if not service:may(node, stanza.attr.from, "set_affiliation") then + origin.send(pubsub_error_reply(stanza, "forbidden")); + return true; + end + + local node_obj = service.nodes[node]; + if not node_obj then + origin.send(pubsub_error_reply(stanza, "item-not-found")); + return true; + end + + local reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub_owner }) + :tag("affiliations", { node = node }); + + for jid, affiliation in pairs(node_obj.affiliations) do + reply:tag("affiliation", { jid = jid, affiliation = affiliation }):up(); + end + origin.send(reply); return true; end +function handlers.owner_set_affiliations(origin, stanza, affiliations, service) + local node = affiliations.attr.node; + if not node then + origin.send(pubsub_error_reply(stanza, "nodeid-required")); + return true; + end + if not service:may(node, stanza.attr.from, "set_affiliation") then + origin.send(pubsub_error_reply(stanza, "forbidden")); + return true; + end + + local node_obj = service.nodes[node]; + if not node_obj then + origin.send(pubsub_error_reply(stanza, "item-not-found")); + return true; + end + + for affiliation_tag in affiliations:childtags("affiliation") do + local jid = affiliation_tag.attr.jid; + local affiliation = affiliation_tag.attr.affiliation; + + jid = jid_prep(jid); + if affiliation == "none" then affiliation = nil; end + + local ok, err = service:set_affiliation(node, stanza.attr.from, jid, affiliation); + if not ok then + -- FIXME Incomplete error handling, + -- see XEP 60 8.9.2.4 Multiple Simultaneous Modifications + origin.send(pubsub_error_reply(stanza, err)); + return true; + end + end + + local reply = st.reply(stanza); + origin.send(reply); + return true; +end + +local function create_encapsulating_item(id, payload) + local item = st.stanza("item", { id = id, xmlns = xmlns_pubsub }); + item:add_child(payload); + return item; +end + +local function archive_itemstore(archive, config, user, node) + module:log("debug", "Creation of itemstore for node %s with config %s", node, config); + local get_set = {}; + function get_set:items() -- luacheck: ignore 212/self + local data, err = archive:find(user, { + limit = tonumber(config["max_items"]); + reverse = true; + }); + if not data then + module:log("error", "Unable to get items: %s", err); + return true; + end + module:log("debug", "Listed items %s", data); + return it.reverse(function() + local id, payload, when, publisher = data(); + if id == nil then + return; + end + local item = create_encapsulating_item(id, payload, publisher); + return id, item; + end); + end + function get_set:get(key) -- luacheck: ignore 212/self + local data, err = archive:find(user, { + key = key; + -- Get the last item with that key, if the archive doesn't deduplicate + reverse = true, + limit = 1; + }); + if not data then + module:log("error", "Unable to get item: %s", err); + return nil, err; + end + local id, payload, when, publisher = data(); + module:log("debug", "Get item %s (published at %s by %s)", id, when, publisher); + if id == nil then + return nil; + end + return create_encapsulating_item(id, payload, publisher); + end + function get_set:set(key, value) -- luacheck: ignore 212/self + local data, err; + if value ~= nil then + local publisher = value.attr.publisher; + local payload = value.tags[1]; + data, err = archive:append(user, key, payload, time_now(), publisher); + else + data, err = archive:delete(user, { key = key; }); + end + if not data then + module:log("error", "Unable to set item: %s", err); + return nil, err; + end + return data; + end + function get_set:clear() -- luacheck: ignore 212/self + return archive:delete(user); + end + function get_set:resize(size) -- luacheck: ignore 212/self + return archive:delete(user, { + truncate = size; + }); + end + function get_set:tail() + -- This should conveniently return the last item + local item = self:get(nil); + if item then + return item.attr.id, item; + end + end + return setmetatable(get_set, archive); +end +_M.archive_itemstore = archive_itemstore; + return _M; |