plugins/mod_muc_mam.lua

   1 -- XEP-0313: Message Archive Management for Prosody MUC
   2 -- Copyright (C) 2011-2017 Kim Alvefur
   3 --
   4 -- This file is MIT/X11 licensed.
   5
   6 if module:get_host_type() ~= "component" then
   7         module:log_status("error", "mod_%s should be loaded only on a MUC component, not normal hosts", module.name);
   8         return;
   9 end
  10
  11 local xmlns_mam     = "urn:xmpp:mam:2";
  12 local xmlns_delay   = "urn:xmpp:delay";
  13 local xmlns_forward = "urn:xmpp:forward:0";
  14 local xmlns_st_id   = "urn:xmpp:sid:0";
  15 local xmlns_muc_user = "http://jabber.org/protocol/muc#user";
  16 local muc_form_enable = "muc#roomconfig_enablearchiving"
  17
  18 local st = require "util.stanza";
  19 local rsm = require "util.rsm";
  20 local jid_bare = require "util.jid".bare;
  21 local jid_split = require "util.jid".split;
  22 local jid_prep = require "util.jid".prep;
  23 local dataform = require "util.dataforms".new;
  24 local get_form_type = require "util.dataforms".get_type;
  25
  26 local mod_muc = module:depends"muc";
  27 local get_room_from_jid = mod_muc.get_room_from_jid;
  28
  29 local is_stanza = st.is_stanza;
  30 local tostring = tostring;
  31 local time_now = os.time;
  32 local m_min = math.min;
  33 local timestamp, timestamp_parse, datestamp = import( "util.datetime", "datetime", "parse", "date");
  34 local default_max_items, max_max_items = 20, module:get_option_number("max_archive_query_results", 50);
  35
  36 local cleanup_after = module:get_option_string("muc_log_expires_after", "1w");
  37 local cleanup_interval = module:get_option_number("muc_log_cleanup_interval", 4 * 60 * 60);
  38
  39 local default_history_length = 20;
  40 local max_history_length = module:get_option_number("max_history_messages", math.huge);
  41
  42 local function get_historylength(room)
  43         return math.min(room._data.history_length or default_history_length, max_history_length);
  44 end
  45
  46 function schedule_cleanup()
  47         -- replaced by non-noop later if cleanup is enabled
  48 end
  49
  50 local log_all_rooms = module:get_option_boolean("muc_log_all_rooms", false);
  51 local log_by_default = module:get_option_boolean("muc_log_by_default", true);
  52
  53 local archive_store = "muc_log";
  54 local archive = module:open_store(archive_store, "archive");
  55
  56 local archive_item_limit = module:get_option_number("storage_archive_item_limit", archive.caps and archive.caps.quota or 1000);
  57
  58 if archive.name == "null" or not archive.find then
  59         if not archive.find then
  60                 module:log("error", "Attempt to open archive storage returned a driver without archive API support");
  61                 module:log("error", "mod_%s does not support archiving",
  62                         archive._provided_by or archive.name and "storage_"..archive.name.."(?)" or "<unknown>");
  63         else
  64                 module:log("error", "Attempt to open archive storage returned null driver");
  65         end
  66         module:log("info", "See https://prosody.im/doc/storage and https://prosody.im/doc/archiving for more information");
  67         return false;
  68 end
  69
  70 local function archiving_enabled(room)
  71         if log_all_rooms then
  72                 module:log("debug", "Archiving all rooms");
  73                 return true;
  74         end
  75         local enabled = room._data.archiving;
  76         if enabled == nil then
  77                 module:log("debug", "Default is %s (for %s)", log_by_default, room.jid);
  78                 return log_by_default;
  79         end
  80         module:log("debug", "Logging in room %s is %s", room.jid, enabled);
  81         return enabled;
  82 end
  83
  84 if not log_all_rooms then
  85         module:hook("muc-config-form", function(event)
  86                 local room, form = event.room, event.form;
  87                 table.insert(form,
  88                 {
  89                         name = muc_form_enable,
  90                         type = "boolean",
  91                         label = "Enable archiving?",
  92                         value = archiving_enabled(room),
  93                 }
  94                 );
  95         end);
  96
  97         module:hook("muc-config-submitted/"..muc_form_enable, function(event)
  98                 event.room._data.archiving = event.value;
  99                 event.status_codes[event.value and "170" or "171"] = true;
 100         end);
 101 end
 102
 103 -- Note: We ignore the 'with' field as this is internally used for stanza types
 104 local query_form = dataform {
 105         { name = "FORM_TYPE"; type = "hidden"; value = xmlns_mam; };
 106         { name = "with"; type = "jid-single"; };
 107         { name = "start"; type = "text-single" };
 108         { name = "end"; type = "text-single"; };
 109 };
 110
 111 -- Serve form
 112 module:hook("iq-get/bare/"..xmlns_mam..":query", function(event)
 113         local origin, stanza = event.origin, event.stanza;
 114         origin.send(st.reply(stanza):tag("query", { xmlns = xmlns_mam }):add_child(query_form:form()));
 115         return true;
 116 end);
 117
 118 -- Handle archive queries
 119 module:hook("iq-set/bare/"..xmlns_mam..":query", function(event)
 120         local origin, stanza = event.origin, event.stanza;
 121         local room_jid = stanza.attr.to;
 122         local room_node = jid_split(room_jid);
 123         local orig_from = stanza.attr.from;
 124         local query = stanza.tags[1];
 125
 126         local room = get_room_from_jid(room_jid);
 127         if not room then
 128                 origin.send(st.error_reply(stanza, "cancel", "item-not-found"))
 129                 return true;
 130         end
 131         local from = jid_bare(orig_from);
 132
 133         -- Banned or not a member of a members-only room?
 134         local from_affiliation = room:get_affiliation(from);
 135         if from_affiliation == "outcast" -- banned
 136                 or room:get_members_only() and not from_affiliation then -- members-only, not a member
 137                 origin.send(st.error_reply(stanza, "auth", "forbidden"))
 138                 return true;
 139         end
 140
 141         local qid = query.attr.queryid;
 142
 143         -- Search query parameters
 144         local qstart, qend;
 145         local form = query:get_child("x", "jabber:x:data");
 146         if form then
 147                 local form_type, err = get_form_type(form);
 148                 if form_type ~= xmlns_mam then
 149                         origin.send(st.error_reply(stanza, "modify", "bad-request", "Unexpected FORM_TYPE, expected '"..xmlns_mam.."'"));
 150                         return true;
 151                 end
 152                 form, err = query_form:data(form);
 153                 if err then
 154                         origin.send(st.error_reply(stanza, "modify", "bad-request", select(2, next(err))));
 155                         return true;
 156                 end
 157                 qstart, qend = form["start"], form["end"];
 158         end
 159
 160         if qstart or qend then -- Validate timestamps
 161                 local vstart, vend = (qstart and timestamp_parse(qstart)), (qend and timestamp_parse(qend))
 162                 if (qstart and not vstart) or (qend and not vend) then
 163                         origin.send(st.error_reply(stanza, "modify", "bad-request", "Invalid timestamp"))
 164                         return true;
 165                 end
 166                 qstart, qend = vstart, vend;
 167         end
 168
 169         module:log("debug", "Archive query id %s from %s until %s)",
 170                 tostring(qid),
 171                 qstart and timestamp(qstart) or "the dawn of time",
 172                 qend and timestamp(qend) or "now");
 173
 174         -- RSM stuff
 175         local qset = rsm.get(query);
 176         local qmax = m_min(qset and qset.max or default_max_items, max_max_items);
 177         local reverse = qset and qset.before or false;
 178
 179         local before, after = qset and qset.before, qset and qset.after;
 180         if type(before) ~= "string" then before = nil; end
 181
 182         -- Load all the data!
 183         local data, err = archive:find(room_node, {
 184                 start = qstart; ["end"] = qend; -- Time range
 185                 limit = qmax + 1;
 186                 before = before; after = after;
 187                 reverse = reverse;
 188                 with = "message<groupchat";
 189         });
 190
 191         if not data then
 192                 if err == "item-not-found" then
 193                         origin.send(st.error_reply(stanza, "modify", "item-not-found"));
 194                 else
 195                         origin.send(st.error_reply(stanza, "cancel", "internal-server-error"));
 196                 end
 197                 return true;
 198         end
 199         local total = tonumber(err);
 200
 201         local msg_reply_attr = { to = stanza.attr.from, from = stanza.attr.to };
 202
 203         local results = {};
 204
 205         -- Wrap it in stuff and deliver
 206         local first, last;
 207         local count = 0;
 208         local complete = "true";
 209         for id, item, when in data do
 210                 count = count + 1;
 211                 if count > qmax then
 212                         complete = nil;
 213                         break;
 214                 end
 215                 local fwd_st = st.message(msg_reply_attr)
 216                         :tag("result", { xmlns = xmlns_mam, queryid = qid, id = id })
 217                                 :tag("forwarded", { xmlns = xmlns_forward })
 218                                         :tag("delay", { xmlns = xmlns_delay, stamp = timestamp(when) }):up();
 219
 220                 -- Strip <x> tag, containing the original senders JID, unless the room makes this public
 221                 if room:get_whois() ~= "anyone" then
 222                         item:maptags(function (tag)
 223                                 if tag.name == "x" and tag.attr.xmlns == xmlns_muc_user then
 224                                         return nil;
 225                                 end
 226                                 return tag;
 227                         end);
 228                 end
 229                 if not is_stanza(item) then
 230                         item = st.deserialize(item);
 231                 end
 232                 item.attr.to = nil;
 233                 item.attr.xmlns = "jabber:client";
 234                 fwd_st:add_child(item);
 235
 236                 if not first then first = id; end
 237                 last = id;
 238
 239                 if reverse then
 240                         results[count] = fwd_st;
 241                 else
 242                         origin.send(fwd_st);
 243                 end
 244         end
 245
 246         if reverse then
 247                 for i = #results, 1, -1 do
 248                         origin.send(results[i]);
 249                 end
 250                 first, last = last, first;
 251         end
 252
 253         -- That's all folks!
 254         module:log("debug", "Archive query %s completed", qid);
 255
 256         origin.send(st.reply(stanza)
 257                 :tag("fin", { xmlns = xmlns_mam, queryid = qid, complete = complete })
 258                         :add_child(rsm.generate {
 259                                 first = first, last = last, count = total }));
 260         return true;
 261 end);
 262
 263 module:hook("muc-get-history", function (event)
 264         local room = event.room;
 265         if not archiving_enabled(room) then return end
 266         local room_jid = room.jid;
 267         local maxstanzas = event.maxstanzas;
 268         local maxchars = event.maxchars;
 269         local since = event.since;
 270         local to = event.to;
 271
 272         if maxstanzas == 0 or maxchars == 0 then
 273                 return -- No history requested
 274         end
 275
 276         if not maxstanzas or maxstanzas > get_historylength(room) then
 277                 maxstanzas = get_historylength(room);
 278         end
 279
 280         if room._history and #room._history >= maxstanzas then
 281                 return -- It can deal with this itself
 282         end
 283
 284         -- Load all the data!
 285         local query = {
 286                 limit = maxstanzas;
 287                 start = since;
 288                 reverse = true;
 289                 with = "message<groupchat";
 290         }
 291         local data, err = archive:find(jid_split(room_jid), query);
 292
 293         if not data then
 294                 module:log("error", "Could not fetch history: %s", err);
 295                 return
 296         end
 297
 298         local history, i = {}, 1;
 299
 300         for id, item, when in data do
 301                 item.attr.to = to;
 302                 item:tag("delay", { xmlns = "urn:xmpp:delay", from = room_jid, stamp = timestamp(when) }):up(); -- XEP-0203
 303                 item:tag("stanza-id", { xmlns = xmlns_st_id, by = room_jid, id = id }):up();
 304                 if room:get_whois() ~= "anyone" then
 305                         item:maptags(function (tag)
 306                                 if tag.name == "x" and tag.attr.xmlns == xmlns_muc_user then
 307                                         return nil;
 308                                 end
 309                                 return tag;
 310                         end);
 311                 end
 312                 if maxchars then
 313                         local chars = #tostring(item);
 314                         if maxchars - chars < 0 then
 315                                 break
 316                         end
 317                         maxchars = maxchars - chars;
 318                 end
 319                 history[i], i = item, i+1;
 320                 -- module:log("debug", item);
 321         end
 322         function event.next_stanza()
 323                 i = i - 1;
 324                 return history[i];
 325         end
 326         return true;
 327 end, 1);
 328
 329 module:hook("muc-broadcast-messages", function (event)
 330         local room, stanza = event.room, event.stanza;
 331
 332         -- Filter out <stanza-id> that claim to be from us
 333         stanza:maptags(function (tag)
 334                 if tag.name == "stanza-id" and tag.attr.xmlns == xmlns_st_id
 335                 and jid_prep(tag.attr.by) == room.jid then
 336                         return nil;
 337                 end
 338                 if tag.name == "x" and tag.attr.xmlns == xmlns_muc_user then
 339                         return nil;
 340                 end
 341                 return tag;
 342         end);
 343
 344 end, 0);
 345
 346 -- Handle messages
 347 local function save_to_history(self, stanza)
 348         local room_node, room_host = jid_split(self.jid);
 349
 350         local stored_stanza = stanza;
 351
 352         if stanza.name == "message" and self:get_whois() == "anyone" then
 353                 stored_stanza = st.clone(stanza);
 354                 stored_stanza.attr.to = nil;
 355                 local actor = jid_bare(self._occupants[stanza.attr.from].jid);
 356                 local affiliation = self:get_affiliation(actor) or "none";
 357                 local role = self:get_role(actor) or self:get_default_role(affiliation);
 358                 stored_stanza:add_direct_child(st.stanza("x", { xmlns = xmlns_muc_user })
 359                         :tag("item", { affiliation = affiliation; role = role; jid = actor }));
 360         end
 361
 362         -- Policy check
 363         if not archiving_enabled(self) then return end -- Don't log
 364
 365         -- Save the type in the 'with' field, allows storing presence without conflicts
 366         local with = stanza.name
 367         if stanza.attr.type then
 368                 with = with .. "<" .. stanza.attr.type
 369         end
 370
 371         -- And stash it
 372         local time = time_now();
 373         local id, err = archive:append(room_node, nil, stored_stanza, time, with);
 374
 375         if not id and err == "quota-limit" then
 376                 if type(cleanup_after) == "number" then
 377                         module:log("debug", "Room '%s' over quota, cleaning archive", room_node);
 378                         local cleaned = archive:delete(room_node, {
 379                                 ["end"] = (os.time() - cleanup_after);
 380                         });
 381                         if cleaned then
 382                                 id, err = archive:append(room_node, nil, stored_stanza, time, with);
 383                         end
 384                 end
 385                 if not id and (archive.caps and archive.caps.truncate) then
 386                         module:log("debug", "User '%s' over quota, truncating archive", room_node);
 387                         local truncated = archive:delete(room_node, {
 388                                 truncate = archive_item_limit - 1;
 389                         });
 390                         if truncated then
 391                                 id, err = archive:append(room_node, nil, stored_stanza, time, with);
 392                         end
 393                 end
 394         end
 395
 396         if id then
 397                 schedule_cleanup(room_node);
 398                 stanza:add_direct_child(st.stanza("stanza-id", { xmlns = xmlns_st_id, by = self.jid, id = id }));
 399         end
 400 end
 401
 402 module:hook("muc-add-history", function (event)
 403         local room, stanza = event.room, event.stanza;
 404         save_to_history(room, stanza);
 405 end);
 406
 407 if module:get_option_boolean("muc_log_presences", false) then
 408         module:hook("muc-occupant-joined", function (event)
 409                 save_to_history(event.room, st.stanza("presence", { from = event.nick }):tag("x", { xmlns = "http://jabber.org/protocol/muc" }));
 410         end);
 411         module:hook("muc-occupant-left", function (event)
 412                 save_to_history(event.room, st.stanza("presence", { type = "unavailable", from = event.nick }));
 413         end);
 414 end
 415
 416 if not archive.delete then
 417         module:log("warn", "Storage driver %s does not support deletion", archive._provided_by);
 418         module:log("warn", "Archived message will persist after a room has been destroyed");
 419 else
 420         module:hook("muc-room-destroyed", function(event)
 421                 local room_node = jid_split(event.room.jid);
 422                 archive:delete(room_node);
 423         end);
 424 end
 425
 426 -- And role/affiliation changes?
 427
 428 module:add_feature(xmlns_mam);
 429
 430 module:hook("muc-disco#info", function(event)
 431         if archiving_enabled(event.room) then
 432                 event.reply:tag("feature", {var=xmlns_mam}):up();
 433         end
 434 end);
 435
 436 -- Cleanup
 437
 438 if cleanup_after ~= "never" then
 439         local cleanup_storage = module:open_store("muc_log_cleanup");
 440         local cleanup_map = module:open_store("muc_log_cleanup", "map");
 441
 442         local day = 86400;
 443         local multipliers = { d = day, w = day * 7, m = 31 * day, y = 365.2425 * day };
 444         local n, m = cleanup_after:lower():match("(%d+)%s*([dwmy]?)");
 445         if not n then
 446                 module:log("error", "Could not parse muc_log_expires_after string %q", cleanup_after);
 447                 return false;
 448         end
 449
 450         cleanup_after = tonumber(n) * ( multipliers[m] or 1 );
 451
 452         module:log("debug", "muc_log_expires_after = %d -- in seconds", cleanup_after);
 453
 454         if not archive.delete then
 455                 module:log("error", "muc_log_expires_after set but mod_%s does not support deleting", archive._provided_by);
 456                 return false;
 457         end
 458
 459         -- For each day, store a set of rooms that have new messages. To expire
 460         -- messages, we collect the union of sets of rooms from dates that fall
 461         -- outside the cleanup range.
 462
 463         local last_date = require "util.cache".new(module:get_option_number("muc_log_cleanup_date_cache_size", 1000));
 464         function schedule_cleanup(roomname, date)
 465                 date = date or datestamp();
 466                 if last_date:get(roomname) == date then return end
 467                 local ok = cleanup_map:set(date, roomname, true);
 468                 if ok then
 469                         last_date:set(roomname, date);
 470                 end
 471         end
 472
 473         cleanup_runner = require "util.async".runner(function ()
 474                 local rooms = {};
 475                 local cut_off = datestamp(os.time() - cleanup_after);
 476                 for date in cleanup_storage:users() do
 477                         if date <= cut_off then
 478                                 module:log("debug", "Messages from %q should be expired", date);
 479                                 local messages_this_day = cleanup_storage:get(date);
 480                                 if messages_this_day then
 481                                         for room in pairs(messages_this_day) do
 482                                                 rooms[room] = true;
 483                                         end
 484                                         if date < cut_off then
 485                                                 -- Messages from the same day as the cut-off might not have expired yet,
 486                                                 -- but all earlier will have, so clear storage for those days.
 487                                                 cleanup_storage:set(date, nil);
 488                                         end
 489                                 end
 490                         end
 491                 end
 492                 local sum, num_rooms = 0, 0;
 493                 for room in pairs(rooms) do
 494                         local ok, err = archive:delete(room, { ["end"] = os.time() - cleanup_after; })
 495                         if ok then
 496                                 num_rooms = num_rooms + 1;
 497                                 sum = sum + (tonumber(ok) or 0);
 498                         end
 499                 end
 500                 module:log("info", "Deleted %d expired messages for %d rooms", sum, num_rooms);
 501         end);
 502
 503         cleanup_task = module:add_timer(1, function ()
 504                 cleanup_runner:run(true);
 505                 return cleanup_interval;
 506         end);
 507 else
 508         module:log("debug", "Archive expiry disabled");
 509 end