prosody
36ba170c4fd0
plugins/mod_smacks.lua

Find changesets by keywords by author, files, the commit message, revision number or hash, or revset expression.

Software / code / prosody

Comparison

plugins/mod_smacks.lua @ 12688:36ba170c4fd0

mod_smacks: Split enable handling to stages, to allow easier SASL2 integration
author Matthew Wild <mwild1@gmail.com>
date Mon, 29 Aug 2022 15:45:52 +0100
parent 12682:464a22f2751c
child 12689:1bc2220cd6ec
comparison
equal deleted inserted replaced
12687:5b69ecaf3427 12688:36ba170c4fd0
105 tail = { condition = "undefined-condition"; text = "Client acknowledged less stanzas than already acknowledged" }; 105 tail = { condition = "undefined-condition"; text = "Client acknowledged less stanzas than already acknowledged" };
106 pop = { condition = "internal-server-error"; text = "Something went wrong with Stream Management" }; 106 pop = { condition = "internal-server-error"; text = "Something went wrong with Stream Management" };
107 overflow = { condition = "resource-constraint", text = "Too many unacked stanzas remaining, session can't be resumed" } 107 overflow = { condition = "resource-constraint", text = "Too many unacked stanzas remaining, session can't be resumed" }
108 }); 108 });
109 109
110 local resume_errors = require "util.error".init("mod_smacks", xmlns_sm3, { 110 local enable_errors = require "util.error".init("mod_smacks", xmlns_sm3, {
111 already_enabled = { condition = "unexpected-request", text = "Stream management is already enabled" };
112 bind_required = { condition = "unexpected-request", text = "Client must bind a resource before enabling stream management" };
113 unavailable = { condition = "service-unavailable", text = "Stream management is not available for this stream" };
114 -- Resumption
111 expired = { condition = "item-not-found", text = "Session expired, and cannot be resumed" }; 115 expired = { condition = "item-not-found", text = "Session expired, and cannot be resumed" };
112 already_bound = { condition = "unexpected-request", text = "Cannot resume another session after a resource is bound" }; 116 already_bound = { condition = "unexpected-request", text = "Cannot resume another session after a resource is bound" };
113 unknown_session = { condition = "item-not-found", text = "Unknown session" }; 117 unknown_session = { condition = "item-not-found", text = "Unknown session" };
114 }); 118 });
115 119
125 end 129 end
126 session.delayed_ack_timer = nil; 130 session.delayed_ack_timer = nil;
127 end 131 end
128 132
129 local function can_do_smacks(session, advertise_only) 133 local function can_do_smacks(session, advertise_only)
130 if session.smacks then return false, "unexpected-request", "Stream management is already enabled"; end 134 if session.smacks then return false, enable_errors.new("already_enabled"); end
131 135
132 local session_type = session.type; 136 local session_type = session.type;
133 if session.username then 137 if session.username then
134 if not(advertise_only) and not(session.resource) then -- Fail unless we're only advertising sm 138 if not(advertise_only) and not(session.resource) then -- Fail unless we're only advertising sm
135 return false, "unexpected-request", "Client must bind a resource before enabling stream management"; 139 return false, enable_errors.new("bind_required");
136 end 140 end
137 return true; 141 return true;
138 elseif s2s_smacks and (session_type == "s2sin" or session_type == "s2sout") then 142 elseif s2s_smacks and (session_type == "s2sin" or session_type == "s2sout") then
139 return true; 143 return true;
140 end 144 end
141 return false, "service-unavailable", "Stream management is not available for this stream"; 145 return false, enable_errors.new("unavailable");
142 end 146 end
143 147
144 module:hook("stream-features", 148 module:hook("stream-features",
145 function (event) 149 function (event)
146 if can_do_smacks(event.origin, true) then 150 if can_do_smacks(event.origin, true) then
292 wrap_session_out(session, resume); 296 wrap_session_out(session, resume);
293 wrap_session_in(session, resume); 297 wrap_session_in(session, resume);
294 return session; 298 return session;
295 end 299 end
296 300
297 function handle_enable(session, stanza, xmlns_sm) 301 function do_enable(session, stanza)
298 local ok, err, err_text = can_do_smacks(session); 302 local ok, err = can_do_smacks(session);
299 if not ok then 303 if not ok then
300 session.log("warn", "Failed to enable smacks: %s", err_text); -- TODO: XEP doesn't say we can send error text, should it? 304 session.log("warn", "Failed to enable smacks: %s", err.text); -- TODO: XEP doesn't say we can send error text, should it?
301 (session.sends2s or session.send)(st.stanza("failed", { xmlns = xmlns_sm }):tag(err, { xmlns = xmlns_errors})); 305 return nil, err;
302 return true;
303 end 306 end
304 307
305 if session.username then 308 if session.username then
306 local old_sessions, err = all_old_sessions:get(session.username); 309 local old_sessions, err = all_old_sessions:get(session.username);
307 session.log("debug", "Old sessions: %q", old_sessions) 310 session.log("debug", "Old sessions: %q", old_sessions)
318 elseif err then 321 elseif err then
319 session.log("error", "Unable to retrieve old resumption counters: %s", err); 322 session.log("error", "Unable to retrieve old resumption counters: %s", err);
320 end 323 end
321 end 324 end
322 325
323 session.log("debug", "Enabling stream management");
324 session.smacks = xmlns_sm;
325
326 wrap_session(session, false);
327
328 local resume_max;
329 local resume_token; 326 local resume_token;
330 local resume = stanza.attr.resume; 327 local resume = stanza.attr.resume;
331 if resume == "true" or resume == "1" then 328 if resume == "true" or resume == "1" then
332 resume_token = new_id(); 329 resume_token = new_id();
333 track_session(session, resume_token); 330 end
334 resume_max = tostring(resume_timeout); 331
335 end 332 return {
336 (session.sends2s or session.send)(st.stanza("enabled", { xmlns = xmlns_sm, id = resume_token, resume = resume, max = resume_max })); 333 id = resume_token;
334 resume_max = resume_token and tostring(resume_timeout) or nil;
335 session = session;
336 finish = function ()
337 session.log("debug", "Enabling stream management");
338
339 track_session(session, resume_token);
340 wrap_session(session, false);
341
342 end;
343 };
344 end
345
346 function handle_enable(session, stanza, xmlns_sm)
347 local enabled, err = do_enable(session, stanza);
348 if not enabled then
349 (session.sends2s or session.send)(st.stanza("failed", { xmlns = xmlns_sm }):add_error(err));
350 return true;
351 end
352
353 session.smacks = xmlns_sm;
354
355 (session.sends2s or session.send)(st.stanza("enabled", {
356 xmlns = xmlns_sm;
357 id = enabled.id;
358 resume = enabled.id and "1" or nil;
359 max = enabled.resume_max;
360 }));
361
362 enabled.finish();
363
337 return true; 364 return true;
338 end 365 end
339 module:hook_tag(xmlns_sm2, "enable", function (session, stanza) return handle_enable(session, stanza, xmlns_sm2); end, 100); 366 module:hook_tag(xmlns_sm2, "enable", function (session, stanza) return handle_enable(session, stanza, xmlns_sm2); end, 100);
340 module:hook_tag(xmlns_sm3, "enable", function (session, stanza) return handle_enable(session, stanza, xmlns_sm3); end, 100); 367 module:hook_tag(xmlns_sm3, "enable", function (session, stanza) return handle_enable(session, stanza, xmlns_sm3); end, 100);
341 368
534 module:hook("s2sin-destroyed", handle_s2s_destroyed); 561 module:hook("s2sin-destroyed", handle_s2s_destroyed);
535 562
536 function do_resume(session, stanza) 563 function do_resume(session, stanza)
537 if session.full_jid then 564 if session.full_jid then
538 session.log("warn", "Tried to resume after resource binding"); 565 session.log("warn", "Tried to resume after resource binding");
539 return nil, resume_errors.new("already_bound"); 566 return nil, enable_errors.new("already_bound");
540 end 567 end
541 568
542 local id = stanza.attr.previd; 569 local id = stanza.attr.previd;
543 local original_session = session_registry[jid.join(session.username, session.host, id)]; 570 local original_session = session_registry[jid.join(session.username, session.host, id)];
544 if not original_session then 571 if not original_session then
545 local old_session = old_session_registry:get(session.username, id); 572 local old_session = old_session_registry:get(session.username, id);
546 if old_session then 573 if old_session then
547 session.log("debug", "Tried to resume old expired session with id %s", id); 574 session.log("debug", "Tried to resume old expired session with id %s", id);
548 clear_old_session(session, id); 575 clear_old_session(session, id);
549 resumption_expired(1); 576 resumption_expired(1);
550 return nil, resume_errors.new("expired", { h = old_session.h }); 577 return nil, enable_errors.new("expired", { h = old_session.h });
551 end 578 end
552 session.log("debug", "Tried to resume non-existent session with id %s", id); 579 session.log("debug", "Tried to resume non-existent session with id %s", id);
553 return nil, resume_errors.new("unknown_session"); 580 return nil, enable_errors.new("unknown_session");
554 end 581 end
555 582
556 if original_session.hibernating_watchdog then 583 if original_session.hibernating_watchdog then
557 original_session.log("debug", "Letting the watchdog go"); 584 original_session.log("debug", "Letting the watchdog go");
558 original_session.hibernating_watchdog:cancel(); 585 original_session.hibernating_watchdog:cancel();