jiti-meet/resources/prosody-plugins/mod_reservations.lua

627 lines
24 KiB
Lua

--- This is a port of Jicofo's Reservation System as a prosody module
-- ref: https://github.com/jitsi/jicofo/blob/master/doc/reservation.md
--
-- We try to retain the same behaviour and interfaces where possible, but there
-- is some difference:
-- * In the event that the DELETE call fails, Jicofo's reservation
-- system retains reservation data and allows re-creation of room if requested by
-- the same creator without making further call to the API; this module does not
-- offer this behaviour. Re-creation of a closed room will behave like a new meeting
-- and trigger a new API call to validate the reservation.
-- * Jicofo's reservation system expect int-based conflict_id. We take any sensible string.
--
-- In broad strokes, this module works by intercepting Conference IQs sent to focus component
-- and buffers it until reservation is confirmed (by calling the provided API endpoint).
-- The IQ events are routed on to focus component if reservation is valid, or error
-- response is sent back to the origin if reservation is denied. Events are routed as usual
-- if the room already exists.
--
--
-- Installation:
-- =============
--
-- Under domain config,
-- 1. add "reservations" to modules_enabled.
-- 2. Specify URL base for your API endpoint using "reservations_api_prefix" (required)
-- 3. Optional config:
-- * set "reservations_api_timeout" to change API call timeouts (defaults to 20 seconds)
-- * set "reservations_api_headers" to specify custom HTTP headers included in
-- all API calls e.g. to provide auth tokens.
-- * set "reservations_api_retry_count" to the number of times API call failures are retried (defaults to 3)
-- * set "reservations_api_retry_delay" seconds to wait between retries (defaults to 3s)
-- * set "reservations_api_should_retry_for_code" to a function that takes an HTTP response code and
-- returns true if API call should be retried. By default, retries are done for 5XX
-- responses. Timeouts are never retried, and HTTP call failures are always retried.
-- * set "reservations_enable_max_occupants" to true to enable integration with
-- mod_muc_max_occupants. Setting thia will allow optional "max_occupants"
-- payload from API to influence max occupants allowed for a given room.
--
--
-- Example config:
--
-- VirtualHost "jitmeet.example.com"
-- modules_enabled = {
-- "reservations";
-- }
-- reservations_api_prefix = "http://reservation.example.com"
--
-- --- The following are all optional
-- reservations_api_headers = {
-- ["Authorization"] = "Bearer TOKEN-237958623045";
-- }
-- reservations_api_timeout = 10 -- timeout if API does not respond within 10s
-- reservations_api_retry_count = 5 -- retry up to 5 times
-- reservations_api_retry_delay = 1 -- wait 1s between retries
-- reservations_api_should_retry_for_code = function (code)
-- return code >= 500 or code == 408
-- end
--
local jid = require 'util.jid';
local http = require "net.http";
local json = require "util.json";
local st = require "util.stanza";
local timer = require 'util.timer';
local datetime = require 'util.datetime';
local get_room_from_jid = module:require "util".get_room_from_jid;
local is_healthcheck_room = module:require "util".is_healthcheck_room;
local room_jid_match_rewrite = module:require "util".room_jid_match_rewrite;
local api_prefix = module:get_option("reservations_api_prefix");
local api_headers = module:get_option("reservations_api_headers");
local api_timeout = module:get_option("reservations_api_timeout", 20);
local api_retry_count = tonumber(module:get_option("reservations_api_retry_count", 3));
local api_retry_delay = tonumber(module:get_option("reservations_api_retry_delay", 3));
local max_occupants_enabled = module:get_option("reservations_enable_max_occupants", false);
-- Option for user to control HTTP response codes that will result in a retry.
-- Defaults to returning true on any 5XX code or 0
local api_should_retry_for_code = module:get_option("reservations_api_should_retry_for_code", function (code)
return code >= 500;
end)
local muc_component_host = module:get_option_string("main_muc");
-- How often to check and evict expired reservation data
local expiry_check_period = 60;
-- Cannot proceed if "reservations_api_prefix" not configured
if not api_prefix then
module:log("error", "reservations_api_prefix not specified. Disabling %s", module:get_name());
return;
end
-- get/infer focus component hostname so we can intercept IQ bound for it
local focus_component_host = module:get_option_string("focus_component");
if not focus_component_host then
local muc_domain_base = module:get_option_string("muc_mapper_domain_base");
if not muc_domain_base then
module:log("error", "Could not infer focus domain. Disabling %s", module:get_name());
return;
end
focus_component_host = 'focus.'..muc_domain_base;
end
-- common HTTP headers added to all API calls
local http_headers = {
["User-Agent"] = "Prosody ("..prosody.version.."; "..prosody.platform..")";
};
if api_headers then -- extra headers from config
for key, value in pairs(api_headers) do
http_headers[key] = value;
end
end
--- Utils
--- Converts int timestamp to datetime string compatible with Java SimpleDateFormat
-- @param t timestamps in seconds. Supports int (as returned by os.time()) or higher
-- precision (as returned by socket.gettime())
-- @return formatted datetime string (yyyy-MM-dd'T'HH:mm:ss.SSSX)
local function to_java_date_string(t)
local t_secs, mantissa = math.modf(t);
local ms_str = (mantissa == 0) and '.000' or tostring(mantissa):sub(2,5);
local date_str = os.date("!%Y-%m-%dT%H:%M:%S", t_secs);
return date_str..ms_str..'Z';
end
--- Start non-blocking HTTP call
-- @param url URL to call
-- @param options options table as expected by net.http where we provide optional headers, body or method.
-- @param callback if provided, called with callback(response_body, response_code) when call complete.
-- @param timeout_callback if provided, called without args when request times out.
-- @param retries how many times to retry on failure; 0 means no retries.
local function async_http_request(url, options, callback, timeout_callback, retries)
local completed = false;
local timed_out = false;
local retries = retries or api_retry_count;
local function cb_(response_body, response_code)
if not timed_out then -- request completed before timeout
completed = true;
if (response_code == 0 or api_should_retry_for_code(response_code)) and retries > 0 then
module:log("warn", "API Response code %d. Will retry after %ds", response_code, api_retry_delay);
timer.add_task(api_retry_delay, function()
async_http_request(url, options, callback, timeout_callback, retries - 1)
end)
return;
end
if callback then
callback(response_body, response_code)
end
end
end
local request = http.request(url, options, cb_);
timer.add_task(api_timeout, function ()
timed_out = true;
if not completed then
http.destroy_request(request);
if timeout_callback then
timeout_callback()
end
end
end);
end
--- Returns current timestamp
local function now()
-- Don't really need higher precision of socket.gettime(). Besides, we loose
-- milliseconds precision when converting back to timestamp from date string
-- when we use datetime.parse(t), so let's be consistent.
return os.time();
end
--- Start RoomReservation implementation
-- Status enums used in RoomReservation:meta.status
local STATUS = {
PENDING = 0;
SUCCESS = 1;
FAILED = -1;
}
local RoomReservation = {};
RoomReservation.__index = RoomReservation;
function newRoomReservation(room_jid, creator_jid)
return setmetatable({
room_jid = room_jid;
-- Reservation metadata. store as table so we can set and read atomically.
-- N.B. This should always be updated using self.set_status_*
meta = {
status = STATUS.PENDING;
mail_owner = jid.bare(creator_jid);
conflict_id = nil;
start_time = now(); -- timestamp, in seconds
expires_at = nil; -- timestamp, in seconds
error_text = nil;
error_code = nil;
};
-- Array of pending events that we need to route once API call is complete
pending_events = {};
-- Set true when API call trigger has been triggered (by enqueue of first event)
api_call_triggered = false;
}, RoomReservation);
end
--- Extracts room name from room jid
function RoomReservation:get_room_name()
return jid.node(self.room_jid);
end
--- Checks if reservation data is expires and should be evicted from store
function RoomReservation:is_expired()
return self.meta.expires_at ~= nil and now() > self.meta.expires_at;
end
--- Main entry point for handing and routing events.
function RoomReservation:enqueue_or_route_event(event)
if self.meta.status == STATUS.PENDING then
table.insert(self.pending_events, event)
if not self.api_call_triggered == true then
self:call_api_create_conference();
end
else
-- API call already complete. Immediately route without enqueueing.
-- This could happen if request comes in between the time reservation approved
-- and when Jicofo actually creates the room.
module:log("debug", "Reservation details already stored. Skipping queue for %s", self.room_jid);
self:route_event(event);
end
end
--- Updates status and initiates event routing. Called internally when API call complete.
function RoomReservation:set_status_success(start_time, duration, mail_owner, conflict_id, max_occupants)
module:log("info", "Reservation created successfully for %s", self.room_jid);
self.meta = {
status = STATUS.SUCCESS;
mail_owner = mail_owner or self.meta.mail_owner;
conflict_id = conflict_id;
start_time = start_time;
expires_at = start_time + duration;
error_text = nil;
error_code = nil;
}
if max_occupants_enabled and max_occupants then
self.meta.max_occupants = max_occupants
end
self:route_pending_events()
end
--- Updates status and initiates error response to pending events. Called internally when API call complete.
function RoomReservation:set_status_failed(error_code, error_text)
module:log("info", "Reservation creation failed for %s - (%s) %s", self.room_jid, error_code, error_text);
self.meta = {
status = STATUS.FAILED;
mail_owner = self.meta.mail_owner;
conflict_id = nil;
start_time = self.meta.start_time;
-- Retain reservation rejection for a short while so we have time to report failure to
-- existing clients and not trigger a re-query too soon.
-- N.B. Expiry could take longer since eviction happens periodically.
expires_at = now() + 30;
error_text = error_text;
error_code = error_code;
}
self:route_pending_events()
end
--- Triggers routing of all enqueued events
function RoomReservation:route_pending_events()
if self.meta.status == STATUS.PENDING then -- should never be called while PENDING. check just in case.
return;
end
module:log("debug", "Routing all pending events for %s", self.room_jid);
local event;
while #self.pending_events ~= 0 do
event = table.remove(self.pending_events);
self:route_event(event)
end
end
--- Event routing implementation
function RoomReservation:route_event(event)
-- this should only be called after API call complete and status no longer PENDING
assert(self.meta.status ~= STATUS.PENDING, "Attempting to route event while API call still PENDING")
local meta = self.meta;
local origin, stanza = event.origin, event.stanza;
if meta.status == STATUS.FAILED then
module:log("debug", "Route: Sending reservation error to %s", stanza.attr.from);
self:reply_with_error(event, meta.error_code, meta.error_text);
else
if meta.status == STATUS.SUCCESS then
if self:is_expired() then
module:log("debug", "Route: Sending reservation expiry to %s", stanza.attr.from);
self:reply_with_error(event, 419, "Reservation expired");
else
module:log("debug", "Route: Forwarding on event from %s", stanza.attr.from);
prosody.core_post_stanza(origin, stanza, false); -- route iq to intended target (focus)
end
else
-- this should never happen unless dev made a mistake. Block by default just in case.
module:log("error", "Reservation for %s has invalid state %s. Rejecting request.", self.room_jid, meta.status);
self:reply_with_error(event, 500, "Failed to determine reservation state");
end
end
end
--- Generates reservation-error stanza and sends to event origin.
function RoomReservation:reply_with_error(event, error_code, error_text)
local stanza = event.stanza;
local id = stanza.attr.id;
local to = stanza.attr.from;
local from = stanza.attr.to;
event.origin.send(
st.iq({ type="error", to=to, from=from, id=id })
:tag("error", { type="cancel" })
:tag("service-unavailable", { xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" }):up()
:tag("text", { xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" }):text(error_text):up()
:tag("reservation-error", { xmlns="http://jitsi.org/protocol/focus", ["error-code"]=tostring(error_code) })
);
end
--- Initiates non-blocking API call to validate reservation
function RoomReservation:call_api_create_conference()
self.api_call_triggered = true;
local url = api_prefix..'/conference';
local request_data = {
name = self:get_room_name();
start_time = to_java_date_string(self.meta.start_time);
mail_owner = self.meta.mail_owner;
}
local http_options = {
body = http.formencode(request_data); -- because Jicofo reservation encodes as form data instead JSON
method = 'POST';
headers = http_headers;
}
module:log("debug", "Sending POST /conference for %s", self.room_jid);
async_http_request(url, http_options, function (response_body, response_code)
self:on_api_create_conference_complete(response_body, response_code);
end, function ()
self:on_api_call_timeout();
end);
end
--- Parses and validates HTTP response body for conference payload
-- Ref: https://github.com/jitsi/jicofo/blob/master/doc/reservation.md
-- @return nil if invalid, or table with payload parsed from JSON response
function RoomReservation:parse_conference_response(response_body)
local data = json.decode(response_body);
if data == nil then -- invalid JSON payload
module:log("error", "Invalid JSON response from API - %s", response_body);
return;
end
if data.name == nil or data.name:lower() ~= self:get_room_name() then
module:log("error", "Missing or mismatching room name - %s", data.name);
return;
end
if data.id == nil then
module:log("error", "Missing id");
return;
end
if data.mail_owner == nil then
module:log("error", "Missing mail_owner");
return;
end
local duration = tonumber(data.duration);
if duration == nil then
module:log("error", "Missing or invalid duration - %s", data.duration);
return;
end
data.duration = duration;
-- if optional max_occupants field set, cast to number
if data.max_occupants ~= nil then
local max_occupants = tonumber(data.max_occupants)
if max_occupants == nil or max_occupants < 1 then
-- N.B. invalid max_occupants rejected even if max_occupants_enabled=false
module:log("error", "Invalid value for max_occupants - %s", data.max_occupants);
return;
end
data.max_occupants = max_occupants
end
local start_time = datetime.parse(data.start_time); -- N.B. we lose milliseconds portion of the date
if start_time == nil then
module:log("error", "Missing or invalid start_time - %s", data.start_time);
return;
end
data.start_time = start_time;
return data;
end
--- Parses and validates HTTP error response body for API call.
-- Expect JSON with a "message" field.
-- @return message string, or generic error message if invalid payload.
function RoomReservation:parse_error_message_from_response(response_body)
local data = json.decode(response_body);
if data ~= nil and data.message ~= nil then
module:log("debug", "Invalid error response body. Will use generic error message.");
return data.message;
else
return "Rejected by reservation server";
end
end
--- callback on API timeout
function RoomReservation:on_api_call_timeout()
self:set_status_failed(500, 'Reservation lookup timed out');
end
--- callback on API response
function RoomReservation:on_api_create_conference_complete(response_body, response_code)
if response_code == 200 or response_code == 201 then
self:handler_conference_data_returned_from_api(response_body);
elseif response_code == 409 then
self:handle_conference_already_exist(response_body);
elseif response_code == nil then -- warrants a retry, but this should be done automatically by the http call method.
self:set_status_failed(500, 'Could not contact reservation server');
else
self:set_status_failed(response_code, self:parse_error_message_from_response(response_body));
end
end
function RoomReservation:handler_conference_data_returned_from_api(response_body)
local data = self:parse_conference_response(response_body);
if not data then -- invalid response from API
module:log("error", "API returned success code but invalid payload");
self:set_status_failed(500, 'Invalid response from reservation server');
else
self:set_status_success(data.start_time, data.duration, data.mail_owner, data.id, data.max_occupants)
end
end
function RoomReservation:handle_conference_already_exist(response_body)
local data = json.decode(response_body);
if data == nil or data.conflict_id == nil then
-- yes, in the case of 409, API expected to return "id" as "conflict_id".
self:set_status_failed(409, 'Invalid response from reservation server');
else
local url = api_prefix..'/conference/'..data.conflict_id;
local http_options = {
method = 'GET';
headers = http_headers;
}
async_http_request(url, http_options, function(response_body, response_code)
if response_code == 200 then
self:handler_conference_data_returned_from_api(response_body);
else
self:set_status_failed(response_code, self:parse_error_message_from_response(response_body));
end
end, function ()
self:on_api_call_timeout();
end);
end
end
--- End RoomReservation
--- Store reservations lookups that are still pending or with room still active
local reservations = {}
local function get_or_create_reservations(room_jid, creator_jid)
if reservations[room_jid] == nil then
module:log("debug", "Creating new reservation data for %s", room_jid);
reservations[room_jid] = newRoomReservation(room_jid, creator_jid);
end
return reservations[room_jid];
end
local function evict_expired_reservations()
local expired = {}
-- first, gather jids of expired rooms. So we don't remove from table while iterating.
for room_jid, res in pairs(reservations) do
if res:is_expired() then
table.insert(expired, room_jid);
end
end
local room;
for _, room_jid in ipairs(expired) do
room = get_room_from_jid(room_jid);
if room then
-- Close room if still active (reservation duration exceeded)
module:log("info", "Room exceeded reservation duration. Terminating %s", room_jid);
room:destroy(nil, "Scheduled conference duration exceeded.");
-- Rely on room_destroyed to calls DELETE /conference and drops reservation[room_jid]
else
module:log("error", "Reservation references expired room that is no longer active. Dropping %s", room_jid);
-- This should not happen unless evict_expired_reservations somehow gets triggered
-- between the time room is destroyed and room_destroyed callback is called. (Possible?)
-- But just in case, we drop the reservation to avoid repeating this path on every pass.
reservations[room_jid] = nil;
end
end
end
timer.add_task(expiry_check_period, function()
evict_expired_reservations();
return expiry_check_period;
end)
--- Intercept conference IQ to Jicofo handle reservation checks before allowing normal event flow
module:log("info", "Hook to global pre-iq/host");
module:hook("pre-iq/host", function(event)
local stanza = event.stanza;
if stanza.name ~= "iq" or stanza.attr.to ~= focus_component_host or stanza.attr.type ~= 'set' then
return; -- not IQ for jicofo. Ignore this event.
end
local conference = stanza:get_child('conference', 'http://jitsi.org/protocol/focus');
if conference == nil then
return; -- not Conference IQ. Ignore.
end
local room_jid = room_jid_match_rewrite(conference.attr.room);
if get_room_from_jid(room_jid) ~= nil then
module:log("debug", "Skip reservation check for existing room %s", room_jid);
return; -- room already exists. Continue with normal flow
end
local res = get_or_create_reservations(room_jid, stanza.attr.from);
res:enqueue_or_route_event(event); -- hand over to reservation obj to route event
return true;
end);
--- Forget reservation details once room destroyed so query is repeated if room re-created
local function room_destroyed(event)
local res;
local room = event.room
if not is_healthcheck_room(room.jid) then
res = reservations[room.jid]
-- drop reservation data for this room
reservations[room.jid] = nil
if res then -- just in case event triggered more than once?
module:log("info", "Dropped reservation data for destroyed room %s", room.jid);
local conflict_id = res.meta.conflict_id
if conflict_id then
local url = api_prefix..'/conference/'..conflict_id;
local http_options = {
method = 'DELETE';
headers = http_headers;
}
module:log("debug", "Sending DELETE /conference/%s", conflict_id);
async_http_request(url, http_options);
end
end
end
end
--- If max_occupants_enabled, update room max_occupants if returned by API
local function room_created(event)
local res;
local room = event.room
if max_occupants_enabled and not is_healthcheck_room(room.jid) then
res = reservations[room.jid]
if res and res.meta.max_occupants ~= nil then
module:log("info", "Setting max_occupants %d for room %s", res.meta.max_occupants, room.jid);
room._data.max_occupants = res.meta.max_occupants
end
end
end
function process_host(host)
if host == muc_component_host then -- the conference muc component
module:log("info", "Hook to muc-room-destroyed on %s", host);
module:context(host):hook("muc-room-destroyed", room_destroyed, -1);
if max_occupants_enabled then
module:log("info", "Hook to muc-room-created on %s (mod_muc_max_occupants integration enabled)", host);
module:context(host):hook("muc-room-created", room_created);
end
end
end
if prosody.hosts[muc_component_host] == nil then
module:log("info", "No muc component found, will listen for it: %s", muc_component_host)
prosody.events.add_handler("host-activated", process_host);
else
process_host(muc_component_host);
end