Merge pull request #7518 from jitsi/aaronkvanmeerten/jibri-queue-component-modules

FEAT: prosody jibri queue component module
This commit is contained in:
Aaron van Meerten 2020-08-18 10:16:39 -05:00 committed by GitHub
commit c3329ec931
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 653 additions and 29 deletions

View File

@ -0,0 +1,5 @@
local jibri_queue_component
= module:get_option_string(
"jibri_queue_component", "jibriqueue"..module.host);
module:add_identity("component", "jibri-queue", jibri_queue_component);

View File

@ -0,0 +1,559 @@
local st = require "util.stanza";
local jid = require "util.jid";
local http = require "net.http";
local json = require "cjson";
local inspect = require('inspect');
local socket = require "socket";
local uuid_gen = require "util.uuid".generate;
local jwt = require "luajwtjitsi";
local it = require "util.iterators";
local neturl = require "net.url";
local parse = neturl.parseQuery;
local get_room_from_jid = module:require "util".get_room_from_jid;
local room_jid_match_rewrite = module:require "util".room_jid_match_rewrite;
local is_healthcheck_room = module:require "util".is_healthcheck_room;
local room_jid_split_subdomain = module:require "util".room_jid_split_subdomain;
local internal_room_jid_match_rewrite = module:require "util".internal_room_jid_match_rewrite;
local async_handler_wrapper = module:require "util".async_handler_wrapper;
-- this basically strips the domain from the conference.domain address
local parentHostName = string.gmatch(tostring(module.host), "%w+.(%w.+)")();
if parentHostName == nil then
log("error", "Failed to start - unable to get parent hostname");
return;
end
local parentCtx = module:context(parentHostName);
if parentCtx == nil then
log("error",
"Failed to start - unable to get parent context for host: %s",
tostring(parentHostName));
return;
end
local token_util = module:require "token/util".new(parentCtx);
local ASAPKeyServer;
local ASAPKeyPath;
local ASAPKeyId;
local ASAPIssuer;
local ASAPAudience;
local ASAPAcceptedIssuers;
local ASAPAcceptedAudiences;
local ASAPTTL;
local ASAPTTL_THRESHOLD;
local ASAPKey;
local JibriRegion;
local disableTokenVerification;
local muc_component_host;
local external_api_url;
local jwtKeyCacheSize;
local jwtKeyCache;
local function load_config()
ASAPKeyServer = module:get_option_string("asap_key_server");
if ASAPKeyServer then
module:log("debug", "ASAP Public Key URL %s", ASAPKeyServer);
token_util:set_asap_key_server(ASAPKeyServer);
end
ASAPKeyPath
= module:get_option_string("asap_key_path", '/etc/prosody/certs/asap.key');
ASAPKeyId
= module:get_option_string("asap_key_id", 'jitsi');
ASAPIssuer
= module:get_option_string("asap_issuer", 'jitsi');
ASAPAudience
= module:get_option_string("asap_audience", 'jibri-queue');
ASAPAcceptedIssuers
= module:get_option_array('asap_accepted_issuers',{'jibri-queue'});
module:log("debug", "ASAP Accepted Issuers %s", ASAPAcceptedIssuers);
token_util:set_asap_accepted_issuers(ASAPAcceptedIssuers);
ASAPAcceptedAudiences
= module:get_option_array('asap_accepted_audiences',{'*'});
module:log("debug", "ASAP Accepted Audiences %s", ASAPAcceptedAudiences);
token_util:set_asap_accepted_audiences(ASAPAcceptedAudiences);
-- do not require room to be set on tokens for jibri queue
token_util:set_asap_require_room_claim(false);
ASAPTTL
= module:get_option_number("asap_ttl", 3600);
ASAPTTL_THRESHOLD
= module:get_option_number("asap_ttl_threshold", 600);
queueServiceURL
= module:get_option_string("jibri_queue_url");
JibriRegion
= module:get_option_string("jibri_region", 'default');
-- option to enable/disable token verifications
disableTokenVerification
= module:get_option_boolean("disable_jibri_queue_token_verification", false);
muc_component_host
= module:get_option_string("muc_component");
external_api_url = module:get_option_string("external_api_url",tostring(parentHostName));
module:log("debug", "External advertised API URL", external_api_url);
-- TODO: Figure out a less arbitrary default cache size.
jwtKeyCacheSize
= module:get_option_number("jwt_pubkey_cache_size", 128);
jwtKeyCache = require"util.cache".new(jwtKeyCacheSize);
if queueServiceURL == nil then
log("error", "No jibri_queue_url specified. No service to contact!");
return;
end
if muc_component_host == nil then
log("error", "No muc_component specified. No muc to operate on for jibri queue!");
return;
end
-- Read ASAP key once on module startup
local f = io.open(ASAPKeyPath, "r");
if f then
ASAPKey = f:read("*all");
f:close();
if not ASAPKey then
module:log("warn", "No ASAP Key read from %s, disabling jibri queue component plugin", ASAPKeyPath);
return
end
else
module:log("warn", "Error reading ASAP Key %s, disabling jibri queue component plugin", ASAPKeyPath);
return
end
return true;
end
local function reload_config()
module:log("info", "Reloading configuration for jibri queue component");
local config_success = load_config();
-- clear ASAP public key cache on config reload
token_util:clear_asap_cache();
if not config_success then
log("error", "Unsuccessful reconfiguration, jibri queue component may misbehave");
end
end
local config_success = load_config();
if not config_success then
log("error", "Unsuccessful configuration step, jibri queue component disabled")
return;
end
local http_headers = {
["User-Agent"] = "Prosody ("..prosody.version.."; "..prosody.platform..")",
["Content-Type"] = "application/json"
};
-- we use async to detect Prosody 0.10 and earlier
local have_async = pcall(require, "util.async");
if not have_async then
module:log("warn", "conference duration will not work with Prosody version 0.10 or less.");
return;
end
log("info", "Starting jibri queue handling for %s", muc_component_host);
local function round(num, numDecimalPlaces)
local mult = 10^(numDecimalPlaces or 0)
return math.floor(num * mult + 0.5) / mult
end
local function generateToken(audience)
audience = audience or ASAPAudience
local t = os.time()
local err
local exp_key = 'asap_exp.'..audience
local token_key = 'asap_token.'..audience
local exp = jwtKeyCache:get(exp_key)
local token = jwtKeyCache:get(token_key)
--if we find a token and it isn't too far from expiry, then use it
if token ~= nil and exp ~= nil then
exp = tonumber(exp)
if (exp - t) > ASAPTTL_THRESHOLD then
return token
end
end
--expiry is the current time plus TTL
exp = t + ASAPTTL
local payload = {
iss = ASAPIssuer,
aud = audience,
nbf = t,
exp = exp,
}
-- encode
local alg = "RS256"
token, err = jwt.encode(payload, ASAPKey, alg, {kid = ASAPKeyId})
if not err then
token = 'Bearer '..token
jwtKeyCache:set(exp_key,exp)
jwtKeyCache:set(token_key,token)
return token
else
return ''
end
end
local function sendIq(participant,action,requestId,time,position,token)
local iqId = uuid_gen();
local from = module:get_host();
local outStanza = st.iq({type = 'set', from = from, to = participant, id = iqId}):tag("jibri-queue",
{ xmlns = 'http://jitsi.org/protocol/jibri-queue', requestId = requestId, action = action });
if token then
outStanza:tag("token"):text(token):up()
end
if time then
outStanza:tag("time"):text(tostring(time)):up()
end
if position then
outStanza:tag("position"):text(tostring(position)):up()
end
module:send(outStanza);
end
local function cb(content_, code_, response_, request_)
if code_ == 200 or code_ == 204 then
module:log("debug", "URL Callback: Code %s, Content %s, Request (host %s, path %s, body %s), Response: %s",
code_, content_, request_.host, request_.path, inspect(request_.body), inspect(response_));
else
module:log("warn", "URL Callback non successful: Code %s, Content %s, Request (%s), Response: %s",
code_, content_, inspect(request_), inspect(response_));
end
end
local function sendEvent(type,room_address,participant,requestId,replyIq,replyError)
local event_ts = round(socket.gettime()*1000);
local node, host, resource, target_subdomain = room_jid_split_subdomain(room_address);
local room_param = '';
if target_subdomain then
room_param = target_subdomain..'/'..node;
else
room_param = node;
end
local out_event = {
["conference"] = room_address,
["roomParam"] = room_param,
["eventType"] = type,
["participant"] = participant,
["externalApiUrl"] = external_api_url.."/jibriqueue/update",
["requestId"] = requestId,
["region"] = JibriRegion,
}
module:log("debug","Sending event %s",inspect(out_event));
local headers = http_headers or {}
headers['Authorization'] = generateToken()
module:log("debug","Sending headers %s",inspect(headers));
local requestURL = queueServiceURL.."/job/recording"
if type=="LeaveQueue" then
requestURL = requestURL .."/cancel"
end
local request = http.request(requestURL, {
headers = headers,
method = "POST",
body = json.encode(out_event)
}, function (content_, code_, response_, request_)
if code_ == 200 or code_ == 204 then
module:log("debug", "URL Callback: Code %s, Content %s, Request (host %s, path %s, body %s), Response: %s",
code_, content_, request_.host, request_.path, inspect(request_.body), inspect(response_));
if (replyIq) then
module:log("debug", "sending reply IQ %s",inspect(replyIq));
module:send(replyIq);
end
else
module:log("warn", "URL Callback non successful: Code %s, Content %s, Request (%s), Response: %s",
code_, content_, inspect(request_), inspect(response_));
if (replyError) then
module:log("warn", "sending reply error IQ %s",inspect(replyError));
module:send(replyError);
end
end
end);
end
function clearRoomQueueByOccupant(room, occupant)
room.jibriQueue[occupant.jid] = nil;
end
function addRoomQueueByOccupant(room, occupant, requestId)
room.jibriQueue[occupant.jid] = requestId;
end
-- receives iq from client currently connected to the room
function on_iq(event)
local requestId;
-- Check the type of the incoming stanza to avoid loops:
if event.stanza.attr.type == "error" then
return; -- We do not want to reply to these, so leave.
end
if event.stanza.attr.to == module:get_host() then
if event.stanza.attr.type == "set" then
local reply = st.reply(event.stanza);
local replyError = st.error_reply(event.stanza,'cancel','internal-server-error',"Queue Server Error");
local jibriQueue
= event.stanza:get_child('jibri-queue', 'http://jitsi.org/protocol/jibri-queue');
if jibriQueue then
module:log("debug", "Received Jibri Queue Request: %s ",inspect(jibriQueue));
local roomAddress = jibriQueue.attr.room;
local room = get_room_from_jid(room_jid_match_rewrite(roomAddress));
if not room then
module:log("warn", "No room found %s", roomAddress);
return false;
end
local from = event.stanza.attr.from;
local occupant = room:get_occupant_by_real_jid(from);
if not occupant then
module:log("warn", "No occupant %s found for %s", from, roomAddress);
return false;
end
local action = jibriQueue.attr.action;
if action == 'join' then
-- join action, so send event out
requestId = uuid_gen();
module:log("debug","Received join queue request for jid %s occupant %s requestId %s",roomAddress,occupant.jid,requestId);
-- now handle new jibri queue message
addRoomQueueByOccupant(room, occupant, requestId);
reply:add_child(st.stanza("jibri-queue", { xmlns = 'http://jitsi.org/protocol/jibri-queue', requestId = requestId})):up()
replyError:add_child(st.stanza("jibri-queue", { xmlns = 'http://jitsi.org/protocol/jibri-queue', requestId = requestId})):up()
module:log("debug","Sending JoinQueue event for jid %s occupant %s reply %s",roomAddress,occupant.jid,inspect(reply));
sendEvent('JoinQueue',roomAddress,occupant.jid,requestId,reply,replyError);
end
if action == 'leave' then
requestId = jibriQueue.attr.requestId;
module:log("debug","Received leave queue request for jid %s occupant %s requestId %s",roomAddress,occupant.jid,requestId);
-- TODO: check that requestId is the same as cached value
clearRoomQueueByOccupant(room, occupant);
reply:add_child(st.stanza("jibri-queue", { xmlns = 'http://jitsi.org/protocol/jibri-queue', requestId = requestId})):up()
replyError:add_child(st.stanza("jibri-queue", { xmlns = 'http://jitsi.org/protocol/jibri-queue', requestId = requestId})):up()
module:log("debug","Sending LeaveQueue event for jid %s occupant %s reply %s",roomAddress,occupant.jid,inspect(reply));
sendEvent('LeaveQueue',roomAddress,occupant.jid,requestId,reply,replyError);
end
else
module:log("warn","Jibri Queue Stanza missing child %s",inspect(event.stanza))
end
end
end
return true
end
-- create recorder queue cache for the room
function room_created(event)
local room = event.room;
if is_healthcheck_room(room.jid) then
return;
end
room.jibriQueue = {};
end
-- Conference ended, clear all queue cache jids
function room_destroyed(event)
local room = event.room;
if is_healthcheck_room(room.jid) then
return;
end
for jid, x in pairs(room.jibriQueue) do
if x then
sendEvent('LeaveQueue',internal_room_jid_match_rewrite(room.jid),jid,x);
end
end
end
-- Occupant left remove it from the queue if it joined the queue
function occupant_leaving(event)
local room = event.room;
if is_healthcheck_room(room.jid) then
return;
end
local occupant = event.occupant;
local requestId = room.jibriQueue[occupant.jid];
-- check if user has cached queue request
if requestId then
-- remove occupant from queue cache, signal backend
room.jibriQueue[occupant.jid] = nil;
sendEvent('LeaveQueue',internal_room_jid_match_rewrite(room.jid),occupant.jid,requestId);
end
end
module:hook("iq/host", on_iq);
-- executed on every host added internally in prosody, including components
function process_host(host)
if host == muc_component_host then -- the conference muc component
module:log("debug","Hook to muc events on %s", host);
local muc_module = module:context(host);
muc_module:hook("muc-room-created", room_created, -1);
-- muc_module:hook("muc-occupant-joined", occupant_joined, -1);
muc_module:hook("muc-occupant-pre-leave", occupant_leaving, -1);
muc_module:hook("muc-room-destroyed", room_destroyed, -1);
end
end
if prosody.hosts[muc_component_host] == nil then
module:log("debug","No muc component found, will listen for it: %s", muc_component_host)
-- when a host or component is added
prosody.events.add_handler("host-activated", process_host);
else
process_host(muc_component_host);
end
module:log("info", "Loading jibri_queue_component");
--- Verifies room name, domain name with the values in the token
-- @param token the token we received
-- @param room_name the room name
-- @param group name of the group (optional)
-- @param session the session to use for storing token specific fields
-- @return true if values are ok or false otherwise
function verify_token(token, room_jid, session)
if disableTokenVerification then
return true;
end
-- if not disableTokenVerification and we do not have token
-- stop here, cause the main virtual host can have guest access enabled
-- (allowEmptyToken = true) and we will allow access to rooms info without
-- a token
if token == nil then
log("warn", "no token provided");
return false;
end
session.auth_token = token;
local verified, reason, message = token_util:process_and_verify_token(session);
if not verified then
log("warn", "not a valid token %s: %s", tostring(reason), tostring(message));
log("debug", "invalid token %s", token);
return false;
end
return true;
end
--- Handles request for updating jibri queue status
-- @param event the http event, holds the request query
-- @return GET response, containing a json with response details
function handle_update_jibri_queue(event)
local body = json.decode(event.request.body);
module:log("debug","Update Jibri Queue Event Received: %s",inspect(body));
local token = event.request.headers["authorization"];
if not token then
token = ''
else
local prefixStart, prefixEnd = token:find("Bearer ");
if prefixStart ~= 1 then
module:log("error", "REST event: Invalid authorization header format. The header must start with the string 'Bearer '");
return { status_code = 403; };
end
token = token:sub(prefixEnd + 1);
end
local user_jid = body["participant"];
local roomAddress = body["conference"];
local userJWT = body["token"];
local action = body["action"];
local time = body["time"];
local position = body["position"];
local requestId = body["requestId"];
if not action then
if userJWT then
action = 'token';
else
action = 'info';
end
end
local room_jid = room_jid_match_rewrite(roomAddress);
if not verify_token(token, room_jid, {}) then
log("error", "REST event: Invalid token for room %s to route action %s for requestId %s", roomAddress, action, requestId);
return { status_code = 403; };
end
local room = get_room_from_jid(room_jid);
if (not room) then
log("error", "REST event: no room found %s to route action %s for requestId %s", roomAddress, action, requestId);
return { status_code = 404; };
end
local occupant = room:get_occupant_by_real_jid(user_jid);
if not occupant then
log("warn", "REST event: No occupant %s found for %s to route action %s for requestId %s", user_jid, roomAddress, action, requestId);
return { status_code = 404; };
end
if not room.jibriQueue[occupant.jid] then
log("warn", "REST event: No queue request found for occupant %s in conference %s to route action %s for requestId %s",occupant.jid,room.jid, action, requestId)
return { status_code = 404; };
end
if not requestId then
requestId = room.jibriQueue[occupant.jid];
end
if action == 'token' and userJWT then
log("debug", "REST event: Token received for occupant %s in conference %s requestId %s, clearing room queue");
clearRoomQueueByOccupant(room, occupant);
end
log("debug", "REST event: Sending update for occupant %s in conference %s to route action %s for requestId %s",occupant.jid,room.jid, action, requestId);
sendIq(occupant.jid,action,requestId,time,position,userJWT);
return { status_code = 200; };
end
module:depends("http");
module:provides("http", {
default_path = "/";
name = "jibriqueue";
route = {
["POST /jibriqueue/update"] = function (event) return async_handler_wrapper(event,handle_update_jibri_queue) end;
};
});
module:hook_global('config-reloaded', reload_config);

View File

@ -93,6 +93,8 @@ function Util.new(module)
--array of accepted audiences: by default only includes our appId
self.acceptedAudiences = module:get_option_array('asap_accepted_audiences',{'*'})
self.requireRoomClaim = module:get_option_boolean('asap_require_room_claim', true);
if self.asapKeyServer and not have_async then
module:log("error", "requires a version of Prosody with util.async");
return nil;
@ -102,7 +104,23 @@ function Util.new(module)
end
function Util:set_asap_key_server(asapKeyServer)
self.asapKeyServer = asapKeyServer
self.asapKeyServer = asapKeyServer;
end
function Util:set_asap_accepted_issuers(acceptedIssuers)
self.acceptedIssuers = acceptedIssuers;
end
function Util:set_asap_accepted_audiences(acceptedAudiences)
self.acceptedAudiences = acceptedAudiences;
end
function Util:set_asap_require_room_claim(checkRoom)
self.requireRoomClaim = checkRoom;
end
function Util:clear_asap_cache()
self.cache = require"util.cache".new(cacheSize);
end
--- Returns the public key by keyID
@ -114,18 +132,41 @@ function Util:get_public_key(keyId)
-- If the key is not found in the cache.
module:log("debug", "Cache miss for key: "..keyId);
local code;
local timeout_occurred;
local wait, done = async.waiter();
local function cb(content_, code_, response_, request_)
content, code = content_, code_;
if code == 200 or code == 204 then
self.cache:set(keyId, content);
else
module:log("warn", "Error on public key request: Code %s, Content %s",
code_, content_);
end
done();
end
local keyurl = path.join(self.asapKeyServer, hex.to(sha256(keyId))..'.pem');
local function cb(content_, code_, response_, request_)
if timeout_occurred == nil then
content, code = content_, code_;
if code == 200 or code == 204 then
self.cache:set(keyId, content);
else
module:log("warn", "Error on public key request: Code %s, Content %s",
code_, content_);
end
done();
else
module:log("warn", "public key reply delivered after timeout from: %s",keyurl);
end
end
-- TODO: Is the done() call racey? Can we cancel this if the request
-- succeedes?
local function cancel()
-- TODO: This check is racey. Not likely to be a problem, but we should
-- still stick a mutex on content / code at some point.
if code == nil then
timeout_occurred = true;
module:log("warn", "Timeout %s seconds fetching public key from: %s",http_timeout,keyurl);
if http.destroy_request ~= nil then
http.destroy_request(request);
end
done();
end
end
module:log("debug", "Fetching public key from: "..keyurl);
-- We hash the key ID to work around some legacy behavior and make
@ -136,19 +177,6 @@ function Util:get_public_key(keyId)
method = "GET"
}, cb);
-- TODO: Is the done() call racey? Can we cancel this if the request
-- succeedes?
local function cancel()
-- TODO: This check is racey. Not likely to be a problem, but we should
-- still stick a mutex on content / code at some point.
if code == nil then
-- no longer present in prosody 0.11, so check before calling
if http.destroy_request ~= nil then
http.destroy_request(request);
end
done();
end
end
timer.add_task(http_timeout, cancel);
wait();
@ -169,6 +197,10 @@ end
-- @param 'acceptedIssuers' list of issuers to check
-- @return nil and error string or true for accepted claim
function Util:verify_issuer(issClaim, acceptedIssuers)
if not acceptedIssuers then
acceptedIssuers = self.acceptedIssuers
end
module:log("debug","verify_issuer claim: %s against accepted: %s",issClaim, acceptedIssuers);
for i, iss in ipairs(acceptedIssuers) do
if issClaim == iss then
--claim matches an accepted issuer so return success
@ -183,6 +215,7 @@ end
-- @param 'aud' claim from the token to verify
-- @return nil and error string or true for accepted claim
function Util:verify_audience(audClaim)
module:log("debug","verify_audience claim: %s against accepted: %s",audClaim, self.acceptedAudiences);
for i, aud in ipairs(self.acceptedAudiences) do
if aud == '*' then
--* indicates to accept any audience in the claims so return success
@ -223,9 +256,11 @@ function Util:verify_token(token, secret, acceptedIssuers)
return nil, issCheckErr;
end
local roomClaim = claims["room"];
if roomClaim == nil then
return nil, "'room' claim is missing";
if self.requireRoomClaim then
local roomClaim = claims["room"];
if roomClaim == nil then
return nil, "'room' claim is missing";
end
end
local audClaim = claims["aud"];

View File

@ -18,14 +18,20 @@ local escaped_muc_domain_prefix = muc_domain_prefix:gsub("%p", "%%%1");
local target_subdomain_pattern
= "^"..escaped_muc_domain_prefix..".([^%.]+)%."..escaped_muc_domain_base;
-- Utility function to split room JID to include room name and subdomain
local function room_jid_split_subdomain(room_jid)
local node, host, resource = jid.split(room_jid);
local target_subdomain = host and host:match(target_subdomain_pattern);
return node, host, resource, target_subdomain
end
--- Utility function to check and convert a room JID from
-- virtual room1@muc.foo.example.com to real [foo]room1@muc.example.com
-- @param room_jid the room jid to match and rewrite if needed
-- @return returns room jid [foo]room1@muc.example.com when it has subdomain
-- otherwise room1@muc.example.com(the room_jid value untouched)
local function room_jid_match_rewrite(room_jid)
local node, host, resource = jid.split(room_jid);
local target_subdomain = host and host:match(target_subdomain_pattern);
local node, host, resource, target_subdomain = room_jid_split_subdomain(room_jid);
if not target_subdomain then
module:log("debug", "No need to rewrite out 'to' %s", room_jid);
return room_jid;
@ -38,6 +44,23 @@ local function room_jid_match_rewrite(room_jid)
return room_jid
end
local function internal_room_jid_match_rewrite(room_jid)
local node, host, resource = jid.split(room_jid);
if host ~= muc_domain or not node then
module:log("debug", "No need to rewrite %s (not from the MUC host)", room_jid);
return room_jid;
end
local target_subdomain, target_node = node:match("^%[([^%]]+)%](.+)$");
if not (target_node and target_subdomain) then
module:log("debug", "Not rewriting... unexpected node format: %s", node);
return room_jid;
end
-- Ok, rewrite room_jid address to pretty format
local new_node, new_host, new_resource = target_node, muc_domain_prefix..".".. target_subdomain.."."..muc_domain_base, resource;
room_jid = jid.join(new_node, new_host, new_resource);
module:log("debug", "Rewrote to %s", room_jid);
return room_jid
end
--- Finds and returns room by its jid
-- @param room_jid the room jid to search in the muc component
@ -191,5 +214,7 @@ return {
get_room_from_jid = get_room_from_jid;
async_handler_wrapper = async_handler_wrapper;
room_jid_match_rewrite = room_jid_match_rewrite;
room_jid_split_subdomain = room_jid_split_subdomain;
internal_room_jid_match_rewrite = internal_room_jid_match_rewrite;
update_presence_identity = update_presence_identity;
};