# HG changeset patch # User Kim Alvefur # Date 1508313389 -7200 # Node ID cb2342cf3f3c61fc888befd008a4d4f85da4c36a # Parent 8d9aed6d1f87ef24e602054ea36976b3f4708033 mod_pep_plus: Snapshot from Prosody trunk 910d3c3f60a6 including dependencies diff -r 8d9aed6d1f87 -r cb2342cf3f3c mod_pep_plus/mod_pep_plus.lua --- a/mod_pep_plus/mod_pep_plus.lua Tue Oct 17 22:39:37 2017 +0200 +++ b/mod_pep_plus/mod_pep_plus.lua Wed Oct 18 09:56:29 2017 +0200 @@ -1,18 +1,19 @@ -local pubsub = require "util.pubsub"; +local pubsub = module:require "util_pubsub"; local jid_bare = require "util.jid".bare; local jid_split = require "util.jid".split; +local jid_join = require "util.jid".join; local set_new = require "util.set".new; local st = require "util.stanza"; local calculate_hash = require "util.caps".calculate_hash; local is_contact_subscribed = require "core.rostermanager".is_contact_subscribed; +local cache = require "util.cache"; +local set = require "util.set"; local xmlns_pubsub = "http://jabber.org/protocol/pubsub"; local xmlns_pubsub_event = "http://jabber.org/protocol/pubsub#event"; local xmlns_pubsub_owner = "http://jabber.org/protocol/pubsub#owner"; local lib_pubsub = module:require "pubsub"; -local handlers = lib_pubsub.handlers; -local pubsub_error_reply = lib_pubsub.pubsub_error_reply; local empty_set = set_new(); @@ -20,6 +21,11 @@ local recipients = {}; local hash_map = {}; +local host = module.host; + +local known_nodes_map = module:open_store("pep", "map"); +local known_nodes = module:open_store("pep"); + function module.save() return { services = services }; end @@ -28,25 +34,41 @@ services = data.services; end -local function subscription_presence(user_bare, recipient) +local function subscription_presence(username, recipient) + local user_bare = jid_join(username, host); local recipient_bare = jid_bare(recipient); if (recipient_bare == user_bare) then return true; end - local username, host = jid_split(user_bare); return is_contact_subscribed(username, host, recipient_bare); end -local function get_broadcaster(name) +local function simple_itemstore(username) + return function (config, node) + if config["persist_items"] then + module:log("debug", "Creating new persistent item store for user %s, node %q", username, node); + known_nodes_map:set(username, node, true); + local archive = module:open_store("pep_"..node, "archive"); + return lib_pubsub.archive_itemstore(archive, config, username, node, false); + else + module:log("debug", "Creating new ephemeral item store for user %s, node %q", username, node); + known_nodes_map:set(username, node, nil); + return cache.new(tonumber(config["max_items"])); + end + end +end + +local function get_broadcaster(username) + local user_bare = jid_join(username, host); local function simple_broadcast(kind, node, jids, item) if item then item = st.clone(item); item.attr.xmlns = nil; -- Clear the pubsub namespace end - local message = st.message({ from = name, type = "headline" }) + local message = st.message({ from = user_bare, type = "headline" }) :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 from %s: %s", jid, name, tostring(item)); + module:log("debug", "Sending notification to %s from %s: %s", jid, user_bare, tostring(item)); message.attr.to = jid; module:send(message); end @@ -54,8 +76,10 @@ return simple_broadcast; end -function get_pep_service(name) - local service = services[name]; +function get_pep_service(username) + module:log("debug", "get_pep_service(%q)", username); + local user_bare = jid_join(username, host); + local service = services[username]; if service then return service; end @@ -155,42 +179,51 @@ }; node_defaults = { - ["pubsub#max_items"] = "1"; + ["max_items"] = 1; + ["persist_items"] = true; }; autocreate_on_publish = true; autocreate_on_subscribe = true; - broadcaster = get_broadcaster(name); + itemstore = simple_itemstore(username); + broadcaster = get_broadcaster(username); get_affiliation = function (jid) - if jid_bare(jid) == name then + if jid_bare(jid) == user_bare then return "owner"; - elseif subscription_presence(name, jid) then + elseif subscription_presence(username, jid) then return "subscriber"; end end; normalize_jid = jid_bare; }); - services[name] = service; - module:add_item("pep-service", { service = service, jid = name }); + local nodes, err = known_nodes:get(username); + if nodes then + module:log("debug", "Restoring nodes for user %s", username); + for node in pairs(nodes) do + module:log("debug", "Restoring node %q", node); + service:create(node, true); + end + elseif err then + module:log("error", "Could not restore nodes for %s: %s", username, err); + else + module:log("debug", "No known nodes"); + end + services[username] = service; + module:add_item("pep-service", { service = service, jid = user_bare }); return service; end 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 - return origin.send(st.error_reply(stanza, "cancel", "bad-request")); + local service_name = origin.username; + if stanza.attr.to ~= nil then + service_name = jid_split(stanza.attr.to); end - local service_name = stanza.attr.to or origin.username.."@"..origin.host local service = get_pep_service(service_name); - 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 module:hook("iq/bare/"..xmlns_pubsub..":pubsub", handle_pubsub_iq); @@ -225,7 +258,7 @@ local function resend_last_item(jid, node, service) local ok, items = service:get_items(node, jid); if not ok then return; end - for i, id in ipairs(items) do + for _, id in ipairs(items) do service.config.broadcaster("items", node, { [jid] = true }, items[id]); end end @@ -268,30 +301,34 @@ module:hook("presence/bare", function(event) -- inbound presence to bare JID recieved local origin, stanza = event.origin, event.stanza; - local user = stanza.attr.to or (origin.username..'@'..origin.host); local t = stanza.attr.type; - local self = not stanza.attr.to; - local service = get_pep_service(user); + local is_self = not stanza.attr.to; + local username = jid_split(stanza.attr.to); + local user_bare = jid_bare(stanza.attr.to); + if is_self then + username = origin.username; + user_bare = jid_join(username, host); + end if not t then -- available presence - if self or subscription_presence(user, stanza.attr.from) then + if is_self or subscription_presence(username, stanza.attr.from) then local recipient = stanza.attr.from; - local current = recipients[user] and recipients[user][recipient]; + local current = recipients[username] and recipients[username][recipient]; local hash, query_node = get_caps_hash_from_presence(stanza, current); if current == hash or (current and current == hash_map[hash]) then return; end if not hash then - update_subscriptions(recipient, user); + update_subscriptions(recipient, username); else - recipients[user] = recipients[user] or {}; + recipients[username] = recipients[username] or {}; if hash_map[hash] then - update_subscriptions(recipient, user, hash_map[hash]); + update_subscriptions(recipient, username, hash_map[hash]); else - recipients[user][recipient] = hash; + recipients[username][recipient] = hash; local from_bare = origin.type == "c2s" and origin.username.."@"..origin.host; - if self or origin.type ~= "c2s" or (recipients[from_bare] and recipients[from_bare][origin.full_jid]) ~= hash then + if is_self or origin.type ~= "c2s" or (recipients[from_bare] and recipients[from_bare][origin.full_jid]) ~= hash then -- COMPAT from ~= stanza.attr.to because OneTeam can't deal with missing from attribute origin.send( - st.stanza("iq", {from=user, to=stanza.attr.from, id="disco", type="get"}) + st.stanza("iq", {from=user_bare, to=stanza.attr.from, id="disco", type="get"}) :tag("query", {xmlns = "http://jabber.org/protocol/disco#info", node = query_node}) ); end @@ -299,14 +336,14 @@ end end elseif t == "unavailable" then - update_subscriptions(stanza.attr.from, user); - elseif not self and t == "unsubscribe" then + update_subscriptions(stanza.attr.from, username); + elseif not is_self and t == "unsubscribe" then local from = jid_bare(stanza.attr.from); - local subscriptions = recipients[user]; + local subscriptions = recipients[username]; if subscriptions then for subscriber in pairs(subscriptions) do if jid_bare(subscriber) == from then - update_subscriptions(subscriber, user); + update_subscriptions(subscriber, username); end end end @@ -321,10 +358,15 @@ end -- Process disco response - local self = not stanza.attr.to; - local user = stanza.attr.to or (origin.username..'@'..origin.host); + local is_self = stanza.attr.to == nil; + local user_bare = jid_bare(stanza.attr.to); + local username = jid_split(stanza.attr.to); + if is_self then + username = origin.username; + user_bare = jid_join(username, host); + end local contact = stanza.attr.from; - local current = recipients[user] and recipients[user][contact]; + local current = recipients[username] and recipients[username][contact]; if type(current) ~= "string" then return; end -- check if waiting for recipient's response local ver = current; if not string.find(current, "#") then @@ -338,20 +380,26 @@ end end hash_map[ver] = notify; -- update hash map - if self then + if is_self then + -- Optimization: Fiddle with other local users for jid, item in pairs(origin.roster) do -- for all interested contacts - if item.subscription == "both" or item.subscription == "from" then - if not recipients[jid] then recipients[jid] = {}; end - update_subscriptions(contact, jid, notify); + if jid then + local contact_node, contact_host = jid_split(jid); + if contact_host == host and item.subscription == "both" or item.subscription == "from" then + update_subscriptions(user_bare, contact_node, notify); + end end end end - update_subscriptions(contact, user, notify); + update_subscriptions(contact, username, notify); end); module:hook("account-disco-info-node", function(event) local reply, stanza, origin = event.reply, event.stanza, event.origin; - local service_name = stanza.attr.to or origin.username.."@"..origin.host + local service_name = origin.username; + if stanza.attr.to ~= nil then + service_name = jid_split(stanza.attr.to); + end local service = get_pep_service(service_name); local node = event.node; local ok = service:get_items(node, jid_bare(stanza.attr.from) or true); @@ -361,33 +409,64 @@ end); module:hook("account-disco-info", function(event) - local reply = event.reply; + local origin, reply = event.origin, event.reply; + reply:tag('identity', {category='pubsub', type='pep'}):up(); - reply:tag('feature', {var='http://jabber.org/protocol/pubsub#publish'}):up(); + + local username = jid_split(reply.attr.from) or origin.username; + local service = get_pep_service(username); + + local suppored_features = lib_pubsub.get_feature_set(service) + set.new{ + -- Features not covered by the above + "access-presence", + "auto-subscribe", + "filtered-notifications", + "last-published", + "persistent-items", + "presence-notifications", + "presence-subscribe", + }; + + for feature in suppored_features do + reply:tag('feature', {var=xmlns_pubsub.."#"..feature}):up(); + end end); module:hook("account-disco-items-node", function(event) local reply, stanza, origin = event.reply, event.stanza, event.origin; local node = event.node; - local service_name = stanza.attr.to or origin.username.."@"..origin.host - local service = get_pep_service(service_name); + local is_self = stanza.attr.to == nil; + local user_bare = jid_bare(stanza.attr.to); + local username = jid_split(stanza.attr.to); + if is_self then + username = origin.username; + user_bare = jid_join(username, host); + end + local service = get_pep_service(username); local ok, ret = service:get_items(node, jid_bare(stanza.attr.from) or true); if not ok then return; end event.exists = true; for _, id in ipairs(ret) do - reply:tag("item", { jid = service_name, name = id }):up(); + reply:tag("item", { jid = user_bare, name = id }):up(); end end); module:hook("account-disco-items", function(event) local reply, stanza, origin = event.reply, event.stanza, event.origin; - local service_name = reply.attr.from or origin.username.."@"..origin.host - local service = get_pep_service(service_name); + local is_self = stanza.attr.to == nil; + local user_bare = jid_bare(stanza.attr.to); + local username = jid_split(stanza.attr.to); + if is_self then + username = origin.username; + user_bare = jid_join(username, host); + end + local service = get_pep_service(username); + local ok, ret = service:get_nodes(jid_bare(stanza.attr.from)); if not ok then return; end for node, node_obj in pairs(ret) do - reply:tag("item", { jid = service_name, node = node, name = node_obj.config.name }):up(); + reply:tag("item", { jid = user_bare, node = node, name = node_obj.config.name }):up(); end end); diff -r 8d9aed6d1f87 -r cb2342cf3f3c mod_pep_plus/pubsub.lib.lua --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/mod_pep_plus/pubsub.lib.lua Wed Oct 18 09:56:29 2017 +0200 @@ -0,0 +1,497 @@ +local t_unpack = table.unpack or unpack; -- luacheck: ignore 113 +local time_now = os.time; + +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; + +local xmlns_pubsub = "http://jabber.org/protocol/pubsub"; +local xmlns_pubsub_errors = "http://jabber.org/protocol/pubsub#errors"; +local xmlns_pubsub_owner = "http://jabber.org/protocol/pubsub#owner"; + +local _M = {}; + +local handlers = {}; +_M.handlers = handlers; + +local pubsub_errors = { + ["conflict"] = { "cancel", "conflict" }; + ["invalid-jid"] = { "modify", "bad-request", nil, "invalid-jid" }; + ["jid-required"] = { "modify", "bad-request", nil, "jid-required" }; + ["nodeid-required"] = { "modify", "bad-request", nil, "nodeid-required" }; + ["item-not-found"] = { "cancel", "item-not-found" }; + ["not-subscribed"] = { "modify", "unexpected-request", nil, "not-subscribed" }; + ["forbidden"] = { "auth", "forbidden" }; + ["not-allowed"] = { "cancel", "not-allowed" }; +}; +local function pubsub_error_reply(stanza, error) + local e = pubsub_errors[error]; + 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 + return reply; +end +_M.pubsub_error_reply = pubsub_error_reply; + +local node_config_form = dataform { + { + type = "hidden"; + name = "FORM_TYPE"; + value = "http://jabber.org/protocol/pubsub#node_config"; + }; + { + type = "text-single"; + name = "pubsub#max_items"; + label = "Max # of items to persist"; + }; + { + type = "boolean"; + name = "pubsub#persist_items"; + label = "Persist items to storage"; + }; +}; + +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" }; +}; +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 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, item_id); + if not ok then + origin.send(pubsub_error_reply(stanza, results)); + return true; + end + + local data = st.stanza("items", { node = node }); + for _, id in ipairs(results) do + data:add_child(results[id]); + end + local reply; + if data then + reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub }) + :add_child(data); + else + reply = pubsub_error_reply(stanza, "item-not-found"); + end + origin.send(reply); + return true; +end + +function handlers.get_subscriptions(origin, stanza, subscriptions, service) + local node = subscriptions.attr.node; + local ok, ret = service:get_subscriptions(node, stanza.attr.from, stanza.attr.from); + if not ok then + origin.send(pubsub_error_reply(stanza, ret)); + return true; + end + local reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub }) + :tag("subscriptions"); + for _, sub in ipairs(ret) do + reply:tag("subscription", { node = sub.node, jid = sub.jid, subscription = 'subscribed' }):up(); + end + origin.send(reply); + return true; +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"]; + }; + end + if node then + ok, ret = service:create(node, stanza.attr.from, config); + if ok then + reply = st.reply(stanza); + else + reply = pubsub_error_reply(stanza, ret); + end + else + repeat + node = uuid_generate(); + ok, ret = service:create(node, stanza.attr.from, config); + until ok or ret ~= "conflict"; + if ok then + reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub }) + :tag("create", { node = node }); + else + reply = pubsub_error_reply(stanza, ret); + end + end + origin.send(reply); + return true; +end + +function handlers.owner_set_delete(origin, stanza, delete, service) + local node = delete.attr.node; + + local reply; + if not node then + origin.send(pubsub_error_reply(stanza, "nodeid-required")); + return true; + end + local ok, ret = service:delete(node, stanza.attr.from); + if ok then + reply = st.reply(stanza); + else + reply = pubsub_error_reply(stanza, ret); + end + origin.send(reply); + return true; +end + +function handlers.set_subscribe(origin, stanza, subscribe, service) + local node, jid = subscribe.attr.node, subscribe.attr.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 + reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub }) + :tag("subscription", { + node = node, + jid = jid, + subscription = "subscribed" + }):up(); + if options_tag then + reply:add_child(options_tag); + end + else + reply = pubsub_error_reply(stanza, ret); + end + origin.send(reply); +end + +function handlers.set_unsubscribe(origin, stanza, unsubscribe, service) + local node, jid = unsubscribe.attr.node, unsubscribe.attr.jid; + if not (node and jid) then + origin.send(pubsub_error_reply(stanza, jid and "nodeid-required" or "invalid-jid")); + return true; + end + local ok, ret = service:remove_subscription(node, stanza.attr.from, jid); + local reply; + if ok then + reply = st.reply(stanza); + else + reply = pubsub_error_reply(stanza, ret); + end + origin.send(reply); + return true; +end + +function handlers.set_publish(origin, stanza, publish, service) + local node = publish.attr.node; + if not node then + origin.send(pubsub_error_reply(stanza, "nodeid-required")); + return true; + end + local item = publish:get_child("item"); + local id = (item and item.attr.id); + if not id then + id = uuid_generate(); + if item then + item.attr.id = id; + end + end + local ok, ret = service:publish(node, stanza.attr.from, id, item); + local reply; + if ok then + reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub }) + :tag("publish", { node = node }) + :tag("item", { id = id }); + else + reply = pubsub_error_reply(stanza, ret); + end + origin.send(reply); + return true; +end + +function handlers.set_retract(origin, stanza, retract, service) + local node, notify = retract.attr.node, retract.attr.notify; + notify = (notify == "1") or (notify == "true"); + local item = retract:get_child("item"); + local id = item and item.attr.id + if not (node and id) then + origin.send(pubsub_error_reply(stanza, node and "item-not-found" or "nodeid-required")); + return true; + end + local reply, notifier; + if notify then + notifier = st.stanza("retract", { id = id }); + end + local ok, ret = service:retract(node, stanza.attr.from, id, notifier); + if ok then + reply = st.reply(stanza); + else + reply = pubsub_error_reply(stanza, ret); + end + origin.send(reply); + return true; +end + +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; + if not node then + origin.send(pubsub_error_reply(stanza, "nodeid-required")); + return true; + end + local ok, ret = service:purge(node, stanza.attr.from, notify); + if ok then + reply = st.reply(stanza); + else + reply = pubsub_error_reply(stanza, ret); + end + origin.send(reply); + return true; +end + +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")); + return true; + end + + if not service:may(node, stanza.attr.from, "configure") 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 node_config = node_obj.config; + local pubsub_form_data = { + ["pubsub#max_items"] = tostring(node_config["max_items"]); + ["pubsub#persist_items"] = node_config["persist_items"] + } + local reply = st.reply(stanza) + :tag("pubsub", { xmlns = xmlns_pubsub_owner }) + :tag("configure", { node = node }) + :add_child(node_config_form:form(pubsub_form_data)); + origin.send(reply); + return true; +end + +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")); + return true; + end + if not service:may(node, stanza.attr.from, "configure") then + origin.send(pubsub_error_reply(stanza, "forbidden")); + return true; + end + 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)); + return true; + end + origin.send(st.reply(stanza)); + return true; +end + +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(pubsub_form_data)); + 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["pubsub#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 + return setmetatable(get_set, archive); +end +_M.archive_itemstore = archive_itemstore; + +return _M; diff -r 8d9aed6d1f87 -r cb2342cf3f3c mod_pep_plus/util_pubsub.lib.lua --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/mod_pep_plus/util_pubsub.lib.lua Wed Oct 18 09:56:29 2017 +0200 @@ -0,0 +1,450 @@ +local events = require "util.events"; +local cache = require "util.cache"; + +local service = {}; +local service_mt = { __index = service }; + +local default_config = { __index = { + itemstore = function (config, _) return cache.new(config["max_items"]) end; + broadcaster = function () end; + get_affiliation = function () end; + capabilities = {}; +} }; +local default_node_config = { __index = { + ["persist_items"] = false; + ["max_items"] = 20; +} }; + +local function new(config) + config = config or {}; + return setmetatable({ + config = setmetatable(config, default_config); + node_defaults = setmetatable(config.node_defaults or {}, default_node_config); + affiliations = {}; + subscriptions = {}; + nodes = {}; + data = {}; + events = events.new(); + }, service_mt); +end + +function service:jids_equal(jid1, jid2) + local normalize = self.config.normalize_jid; + return normalize(jid1) == normalize(jid2); +end + +function service:may(node, actor, action) + if actor == true then return true; end + + local node_obj = self.nodes[node]; + local node_aff = node_obj and node_obj.affiliations[actor]; + local service_aff = self.affiliations[actor] + or self.config.get_affiliation(actor, node, action) + or "none"; + + -- Check if node allows/forbids it + local node_capabilities = node_obj and node_obj.capabilities; + if node_capabilities then + local caps = node_capabilities[node_aff or service_aff]; + if caps then + local can = caps[action]; + if can ~= nil then + return can; + end + end + end + + -- Check service-wide capabilities instead + local service_capabilities = self.config.capabilities; + local caps = service_capabilities[node_aff or service_aff]; + if caps then + local can = caps[action]; + if can ~= nil then + return can; + end + end + + return false; +end + +function service:set_affiliation(node, actor, jid, affiliation) + -- Access checking + if not self:may(node, actor, "set_affiliation") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + node_obj.affiliations[jid] = affiliation; + local _, jid_sub = self:get_subscription(node, true, jid); + if not jid_sub and not self:may(node, jid, "be_unsubscribed") then + local ok, err = self:add_subscription(node, true, jid); + if not ok then + return ok, err; + end + elseif jid_sub and not self:may(node, jid, "be_subscribed") then + local ok, err = self:add_subscription(node, true, jid); + if not ok then + return ok, err; + end + end + return true; +end + +function service:add_subscription(node, actor, jid, options) + -- Access checking + local cap; + if actor == true or jid == actor or self:jids_equal(actor, jid) then + cap = "subscribe"; + else + cap = "subscribe_other"; + end + if not self:may(node, actor, cap) then + return false, "forbidden"; + end + if not self:may(node, jid, "be_subscribed") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + if not self.config.autocreate_on_subscribe then + return false, "item-not-found"; + else + local ok, err = self:create(node, true); + if not ok then + return ok, err; + end + node_obj = self.nodes[node]; + end + end + node_obj.subscribers[jid] = options or true; + local normal_jid = self.config.normalize_jid(jid); + local subs = self.subscriptions[normal_jid]; + if subs then + if not subs[jid] then + subs[jid] = { [node] = true }; + else + subs[jid][node] = true; + end + else + self.subscriptions[normal_jid] = { [jid] = { [node] = true } }; + end + self.events.fire_event("subscription-added", { node = node, jid = jid, normalized_jid = normal_jid, options = options }); + return true; +end + +function service:remove_subscription(node, actor, jid) + -- Access checking + local cap; + if actor == true or jid == actor or self:jids_equal(actor, jid) then + cap = "unsubscribe"; + else + cap = "unsubscribe_other"; + end + if not self:may(node, actor, cap) then + return false, "forbidden"; + end + if not self:may(node, jid, "be_unsubscribed") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + if not node_obj.subscribers[jid] then + return false, "not-subscribed"; + end + node_obj.subscribers[jid] = nil; + local normal_jid = self.config.normalize_jid(jid); + local subs = self.subscriptions[normal_jid]; + if subs then + local jid_subs = subs[jid]; + if jid_subs then + jid_subs[node] = nil; + if next(jid_subs) == nil then + subs[jid] = nil; + end + end + if next(subs) == nil then + self.subscriptions[normal_jid] = nil; + end + end + self.events.fire_event("subscription-removed", { node = node, jid = jid, normalized_jid = normal_jid }); + return true; +end + +function service:remove_all_subscriptions(actor, jid) + local normal_jid = self.config.normalize_jid(jid); + local subs = self.subscriptions[normal_jid] + subs = subs and subs[jid]; + if subs then + for node in pairs(subs) do + self:remove_subscription(node, true, jid); + end + end + return true; +end + +function service:get_subscription(node, actor, jid) + -- Access checking + local cap; + if actor == true or jid == actor or self:jids_equal(actor, jid) then + cap = "get_subscription"; + else + cap = "get_subscription_other"; + end + if not self:may(node, actor, cap) then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + return true, node_obj.subscribers[jid]; +end + +function service:create(node, actor, options) + -- Access checking + if not self:may(node, actor, "create") then + return false, "forbidden"; + end + -- + if self.nodes[node] then + return false, "conflict"; + end + + self.nodes[node] = { + name = node; + subscribers = {}; + config = setmetatable(options or {}, {__index=self.node_defaults}); + affiliations = {}; + }; + self.data[node] = self.config.itemstore(self.nodes[node].config, node); + self.events.fire_event("node-created", { node = node, actor = actor }); + local ok, err = self:set_affiliation(node, true, actor, "owner"); + if not ok then + self.nodes[node] = nil; + self.data[node] = nil; + end + return ok, err; +end + +function service:delete(node, actor) + -- Access checking + if not self:may(node, actor, "delete") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + self.nodes[node] = nil; + if self.data[node] and self.data[node].clear then + self.data[node]:clear(); + end + self.data[node] = nil; + self.events.fire_event("node-deleted", { node = node, actor = actor }); + self.config.broadcaster("delete", node, node_obj.subscribers); + return true; +end + +function service:publish(node, actor, id, item) + -- Access checking + if not self:may(node, actor, "publish") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + if not self.config.autocreate_on_publish then + return false, "item-not-found"; + end + local ok, err = self:create(node, true); + if not ok then + return ok, err; + end + node_obj = self.nodes[node]; + end + local node_data = self.data[node]; + local ok = node_data:set(id, item); + if not ok then + return nil, "internal-server-error"; + end + if type(ok) == "string" then id = ok; end + self.events.fire_event("item-published", { node = node, actor = actor, id = id, item = item }); + self.config.broadcaster("items", node, node_obj.subscribers, item, actor); + return true; +end + +function service:retract(node, actor, id, retract) + -- Access checking + if not self:may(node, actor, "retract") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if (not node_obj) or (not self.data[node]:get(id)) then + return false, "item-not-found"; + end + local ok = self.data[node]:set(id, nil); + if not ok then + return nil, "internal-server-error"; + end + self.events.fire_event("item-retracted", { node = node, actor = actor, id = id }); + if retract then + self.config.broadcaster("items", node, node_obj.subscribers, retract); + end + return true +end + +function service:purge(node, actor, notify) + -- Access checking + if not self:may(node, actor, "retract") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + if self.data[node] and self.data[node].clear then + self.data[node]:clear() + else + self.data[node] = self.config.itemstore(self.nodes[node].config, node); + end + self.events.fire_event("node-purged", { node = node, actor = actor }); + if notify then + self.config.broadcaster("purge", node, node_obj.subscribers); + end + return true +end + +function service:get_items(node, actor, id) + -- Access checking + if not self:may(node, actor, "get_items") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + if id then -- Restrict results to a single specific item + local with_id = self.data[node]:get(id); + if not with_id then + return false, "item-not-found"; + end + return true, { id, [id] = with_id }; + else + local data = {} + for key, value in self.data[node]:items() do + data[#data+1] = key; + data[key] = value; + end + return true, data; + end +end + +function service:get_nodes(actor) + -- Access checking + if not self:may(nil, actor, "get_nodes") then + return false, "forbidden"; + end + -- + return true, self.nodes; +end + +function service:get_subscriptions(node, actor, jid) + -- Access checking + local cap; + if actor == true or jid == actor or self:jids_equal(actor, jid) then + cap = "get_subscriptions"; + else + cap = "get_subscriptions_other"; + end + if not self:may(node, actor, cap) then + return false, "forbidden"; + end + -- + local node_obj; + if node then + node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + end + local normal_jid = self.config.normalize_jid(jid); + local subs = self.subscriptions[normal_jid]; + -- We return the subscription object from the node to save + -- a get_subscription() call for each node. + local ret = {}; + if subs then + for subscribed_jid, subscribed_nodes in pairs(subs) do + if node then -- Return only subscriptions to this node + if subscribed_nodes[node] then + ret[#ret+1] = { + node = node; + jid = subscribed_jid; + subscription = node_obj.subscribers[subscribed_jid]; + }; + end + else -- Return subscriptions to all nodes + local nodes = self.nodes; + for subscribed_node in pairs(subscribed_nodes) do + ret[#ret+1] = { + node = subscribed_node; + jid = subscribed_jid; + subscription = nodes[subscribed_node].subscribers[subscribed_jid]; + }; + end + end + end + end + return true, ret; +end + +-- Access models only affect 'none' affiliation caps, service/default access level... +function service:set_node_capabilities(node, actor, capabilities) + -- Access checking + if not self:may(node, actor, "configure") then + return false, "forbidden"; + end + -- + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + node_obj.capabilities = capabilities; + return true; +end + +function service:set_node_config(node, actor, new_config) + if not self:may(node, actor, "configure") then + return false, "forbidden"; + end + + local node_obj = self.nodes[node]; + if not node_obj then + return false, "item-not-found"; + end + + for k,v in pairs(new_config) do + node_obj.config[k] = v; + end + local new_data = self.config.itemstore(self.nodes[node].config, node); + for key, value in self.data[node]:items() do + new_data:set(key, value); + end + self.data[node] = new_data; + return true; +end + +return { + new = new; +};