res_stasis_broadcast.c File Reference

Stasis application broadcast resource. More...

#include "asterisk.h"
#include <errno.h>
#include <regex.h>
#include "asterisk/astobj2.h"
#include "asterisk/channel.h"
#include "asterisk/http.h"
#include "asterisk/json.h"
#include "asterisk/lock.h"
#include "asterisk/module.h"
#include "asterisk/pbx.h"
#include "asterisk/stasis_app.h"
#include "asterisk/stasis_app_impl.h"
#include "asterisk/stasis_channels.h"
#include "asterisk/time.h"
#include "asterisk/utils.h"
#include "asterisk/stasis_app_broadcast.h"

Include dependency graph for res_stasis_broadcast.c:

Go to the source code of this file.

Data Structures
struct	stasis_broadcast_ctx
	Broadcast context stored on channel. More...

Macros
#define	ALTERNATION_DEPTH_THRESHOLD   2
	Group depth threshold for alternation limits.
#define	AST_API_MODULE   /* Mark this as the module providing the API */
#define	BROADCAST_BUCKETS   37
#define	BROADCAST_POLL_INTERVAL_MS   200
	Interval in ms between hangup checks while waiting for a claim.
#define	MAX_ALTERNATIONS   20
	Maximum alternations allowed in deeply nested groups.
#define	MAX_BROADCAST_TIMEOUT_MS   (24 * 60 * 60 * 1000)
	Maximum broadcast timeout in milliseconds (24 hours)
#define	MAX_GROUP_DEPTH   10
	Maximum depth for regex group nesting.
#define	MAX_NESTED_QUANTIFIERS   3
	Maximum number of nested quantifiers in regex.
#define	MAX_QUANTIFIER_BOUND   100
	Maximum value for brace quantifier bounds {m,n}.
#define	MAX_REGEX_LENGTH   256
	Maximum length for app_filter regex pattern.

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
struct ast_module *	AST_MODULE_SELF_SYM (void)
static struct stasis_broadcast_ctx *	broadcast_ctx_create (const char *channel_id, const char *app_filter, unsigned int flags)
	Create a new broadcast context.
static void	broadcast_ctx_destructor (void *obj)
	Destructor for broadcast context.
static void	broadcast_datastore_destroy (void *data)
	Destructor for broadcast datastore.
static int	load_module (void)
static int	send_broadcast_event (struct ast_channel *chan, struct stasis_broadcast_ctx *ctx)
	Create and send broadcast event to all applications.
int AST_OPTIONAL_API_NAME()	stasis_app_broadcast_channel (struct ast_channel *chan, int timeout_ms, const char *app_filter, unsigned int flags)
	Start a broadcast for a channel.
void AST_OPTIONAL_API_NAME()	stasis_app_broadcast_cleanup (const char *channel_id)
	Clean up broadcast context for a channel.
int AST_OPTIONAL_API_NAME()	stasis_app_broadcast_wait (struct ast_channel *chan, int timeout_ms)
	Wait for a broadcast channel to be claimed.
char *AST_OPTIONAL_API_NAME()	stasis_app_broadcast_winner (const char *channel_id)
	Get the winner app name for a broadcast channel.
int AST_OPTIONAL_API_NAME()	stasis_app_claim_channel (const char *channel_id, const char *app_name)
	Attempt to claim a broadcast channel.
static int	unload_module (void)
static int	validate_regex_pattern (const char *pattern)
	Validate a regex pattern for safety.

Variables
static struct ast_module_info	__mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER , .description = "Stasis application broadcast" , .key = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_EXTENDED, .load = load_module, .unload = unload_module, .requires = "res_stasis,res_ari,http", .load_pri = AST_MODPRI_APP_DEPEND - 1, }
static const struct ast_module_info *	ast_module_info = &__mod_info
static struct ao2_container *	broadcast_contexts
	Container for all active broadcast contexts.
static const struct ast_datastore_info	broadcast_datastore_info
	Datastore information for broadcast context.

Detailed Description

Stasis application broadcast resource.

Author

Daniel Donoghue danie.nosp@m.l.do.nosp@m.noghu.nosp@m.e@au.nosp@m.rorai.nosp@m.nnov.nosp@m.ation.nosp@m..com

Definition in file res_stasis_broadcast.c.

Macro Definition Documentation

ALTERNATION_DEPTH_THRESHOLD

#define ALTERNATION_DEPTH_THRESHOLD   2

Group depth threshold for alternation limits.

Definition at line 71 of file res_stasis_broadcast.c.

AST_API_MODULE

#define AST_API_MODULE   /* Mark this as the module providing the API */

Definition at line 50 of file res_stasis_broadcast.c.

BROADCAST_BUCKETS

#define BROADCAST_BUCKETS   37

Definition at line 53 of file res_stasis_broadcast.c.

BROADCAST_POLL_INTERVAL_MS

#define BROADCAST_POLL_INTERVAL_MS   200

Interval in ms between hangup checks while waiting for a claim.

Definition at line 77 of file res_stasis_broadcast.c.

MAX_ALTERNATIONS

#define MAX_ALTERNATIONS   20

Maximum alternations allowed in deeply nested groups.

Definition at line 68 of file res_stasis_broadcast.c.

MAX_BROADCAST_TIMEOUT_MS

#define MAX_BROADCAST_TIMEOUT_MS   (24 * 60 * 60 * 1000)

Maximum broadcast timeout in milliseconds (24 hours)

Definition at line 74 of file res_stasis_broadcast.c.

MAX_GROUP_DEPTH

#define MAX_GROUP_DEPTH   10

Maximum depth for regex group nesting.

Definition at line 59 of file res_stasis_broadcast.c.

MAX_NESTED_QUANTIFIERS

#define MAX_NESTED_QUANTIFIERS   3

Maximum number of nested quantifiers in regex.

Definition at line 62 of file res_stasis_broadcast.c.

MAX_QUANTIFIER_BOUND

#define MAX_QUANTIFIER_BOUND   100

Maximum value for brace quantifier bounds {m,n}.

Definition at line 65 of file res_stasis_broadcast.c.

MAX_REGEX_LENGTH

#define MAX_REGEX_LENGTH   256

Maximum length for app_filter regex pattern.

Definition at line 56 of file res_stasis_broadcast.c.

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 933 of file res_stasis_broadcast.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 933 of file res_stasis_broadcast.c.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 933 of file res_stasis_broadcast.c.

broadcast_ctx_create()

static struct stasis_broadcast_ctx * broadcast_ctx_create ( const char *  channel_id, const char *  app_filter, unsigned int  flags  )

static

Create a new broadcast context.

Validates and compiles the app_filter regex if provided. On regex failure the context is still created but broadcasts will be sent to all applications (i.e. no filtering).

Definition at line 151 of file res_stasis_broadcast.c.

{
    struct stasis_broadcast_ctx *ctx;
 
    ctx = ao2_alloc(sizeof(*ctx), broadcast_ctx_destructor);
    if (!ctx) {
        return NULL;
    }
 
    /* ao2_alloc zeroes the struct; only set non-zero fields explicitly */
    ast_copy_string(ctx->channel_id, channel_id, sizeof(ctx->channel_id));
    ctx->flags = flags;
    ctx->container = ao2_bump(broadcast_contexts);
    ast_cond_init(&ctx->cond, NULL);
 
    /* Validate and compile app_filter regex if provided */
    if (!ast_strlen_zero(app_filter)) {
        ast_copy_string(ctx->app_filter, app_filter, sizeof(ctx->app_filter));
        if (validate_regex_pattern(app_filter) != 0) {
            ast_log(LOG_WARNING,
                "Channel %s: rejecting app_filter regex as potentially dangerous: %s\n",
                channel_id, app_filter);
        } else if (regcomp(&ctx->compiled_filter, app_filter,
                REG_EXTENDED | REG_NOSUB) != 0) {
            ast_log(LOG_WARNING,
                "Channel %s: failed to compile app_filter regex '%s'\n",
                channel_id, app_filter);
        } else {
            ctx->filter_compiled = 1;
        }
 
        if (!ctx->filter_compiled) {
            ast_log(LOG_WARNING,
                "Channel %s: proceeding without application filtering due to invalid regex\n",
                channel_id);
        }
    }
 
    ast_debug(1, "Created broadcast context for channel %s (filter: %s, flags: 0x%x)\n",
        ctx->channel_id,
        ctx->filter_compiled ? ctx->app_filter : "none",
        ctx->flags);
 
    return ctx;
}

References ao2_alloc, ao2_bump, stasis_broadcast_ctx::app_filter, ast_cond_init, ast_copy_string(), ast_debug, ast_log, ast_strlen_zero(), broadcast_contexts, broadcast_ctx_destructor(), stasis_broadcast_ctx::channel_id, stasis_broadcast_ctx::compiled_filter, stasis_broadcast_ctx::cond, stasis_broadcast_ctx::container, stasis_broadcast_ctx::filter_compiled, stasis_broadcast_ctx::flags, LOG_WARNING, NULL, and validate_regex_pattern().

Referenced by stasis_app_broadcast_channel().

broadcast_ctx_destructor()

static void broadcast_ctx_destructor ( void *  obj )

static

Destructor for broadcast context.

Definition at line 132 of file res_stasis_broadcast.c.

{
    struct stasis_broadcast_ctx *ctx = obj;
    ast_free(ctx->winner_app);
    if (ctx->filter_compiled) {
        regfree(&ctx->compiled_filter);
    }
    ao2_cleanup(ctx->container);
    ast_cond_destroy(&ctx->cond);
}

References ao2_cleanup, ast_cond_destroy, ast_free, stasis_broadcast_ctx::compiled_filter, stasis_broadcast_ctx::cond, stasis_broadcast_ctx::container, stasis_broadcast_ctx::filter_compiled, and stasis_broadcast_ctx::winner_app.

Referenced by broadcast_ctx_create().

broadcast_datastore_destroy()

static void broadcast_datastore_destroy ( void *  data )

static

Destructor for broadcast datastore.

Called when the channel is destroyed. Ensures the broadcast context is unlinked from the global container even if the caller never reached stasis_app_broadcast_cleanup (e.g. abnormal channel teardown).

Definition at line 112 of file res_stasis_broadcast.c.

{
    struct stasis_broadcast_ctx *ctx = data;
 
    if (ctx->container) {
        ao2_unlink(ctx->container, ctx);
    }
    ao2_cleanup(ctx);
}

References ao2_cleanup, ao2_unlink, and stasis_broadcast_ctx::container.

load_module()

static int load_module ( void  )

static

Definition at line 899 of file res_stasis_broadcast.c.

{
    broadcast_contexts = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
        BROADCAST_BUCKETS, stasis_broadcast_ctx_hash_fn, NULL, stasis_broadcast_ctx_cmp_fn);
 
    if (!broadcast_contexts) {
        return AST_MODULE_LOAD_DECLINE;
    }
 
    ast_debug(1, "Stasis broadcast module loaded\n");
    return AST_MODULE_LOAD_SUCCESS;
}

References AO2_ALLOC_OPT_LOCK_MUTEX, ao2_container_alloc_hash, ast_debug, AST_MODULE_LOAD_DECLINE, AST_MODULE_LOAD_SUCCESS, BROADCAST_BUCKETS, broadcast_contexts, and NULL.

send_broadcast_event()

static int send_broadcast_event ( struct ast_channel *  chan, struct stasis_broadcast_ctx *  ctx  )

static

Create and send broadcast event to all applications.

Uses the compiled regex cached in ctx for application filtering.

Definition at line 412 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct ast_json *, event, NULL, ast_json_unref);
    RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
    RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
    struct ao2_iterator iter;
    char *app_name;
    const char *caller = NULL;
    const char *called = NULL;
 
    /* Get snapshot and caller/called info under a single channel lock */
    ast_channel_lock(chan);
    snapshot = ao2_bump(ast_channel_snapshot(chan));
    caller = ast_strdupa(S_OR(ast_channel_caller(chan)->id.number.str, ""));
    called = ast_strdupa(S_OR(ast_channel_exten(chan), ""));
    ast_channel_unlock(chan);
 
    /* Build the broadcast event.  Channel variables configured in
     * ari.conf "channelvars" are already included in the channel
     * snapshot produced by ast_channel_snapshot_to_json(). */
    event = ast_json_pack("{s: s, s: o, s: o, s: s?, s: s?}",
        "type", "CallBroadcast",
        "timestamp", ast_json_timeval(ast_tvnow(), NULL),
        "channel", ast_channel_snapshot_to_json(snapshot, NULL),
        "caller", caller,
        "called", called);
 
    if (!event) {
        ast_log(LOG_ERROR, "Channel %s: failed to create broadcast event\n",
            ast_channel_name(chan));
        return -1;
    }
 
    /* Get all registered applications */
    apps = stasis_app_get_all();
    if (!apps) {
        ast_log(LOG_ERROR, "Channel %s: failed to get stasis applications\n",
            ast_channel_name(chan));
        return -1;
    }
 
    ast_debug(2, "Broadcasting to %d registered Stasis applications\n",
        ao2_container_count(apps));
 
    /*
     * Broadcast to all matching applications.
     *
     * We collect matching apps into a plain array, Fisher-Yates shuffle it,
     * then call stasis_app_send() for each.  stasis_app_send() writes
     * directly to each app's WebSocket socket synchronously on the calling
     * thread.  The shuffle ensures no single ARI application is consistently
     * first to receive the event — every app gets a fair chance to claim the
     * channel regardless of its position in the ao2 hash container.
     */
    {
        int app_count;
        char **matching_arr;
        int n = 0;
        int i;
 
        app_count = ao2_container_count(apps);
        if (app_count == 0) {
            ast_debug(2, "Channel %s: no Stasis applications registered\n",
                ast_channel_uniqueid(chan));
            return 0;
        }
 
        matching_arr = ast_malloc(app_count * sizeof(*matching_arr));
        if (!matching_arr) {
            ast_log(LOG_ERROR, "Channel %s: failed to allocate matching apps array\n",
                ast_channel_name(chan));
            return -1;
        }
 
        /* First pass: collect all matching app names (transfer refs to array) */
        iter = ao2_iterator_init(apps, 0);
        while ((app_name = ao2_iterator_next(&iter)) && n < app_count) {
            if (ctx->filter_compiled &&
                regexec(&ctx->compiled_filter, app_name, 0, NULL, 0) == REG_NOMATCH) {
                ast_debug(3, "App '%s' does not match filter, skipping\n", app_name);
                ao2_ref(app_name, -1);
                continue;
            }
            matching_arr[n++] = app_name; /* ref transferred to array */
        }
        ao2_iterator_destroy(&iter);
 
        ast_debug(2, "Broadcasting channel %s to %d matching applications\n",
            ast_channel_uniqueid(chan), n);
 
        /* Fisher-Yates shuffle: randomise delivery order so no app is
         * consistently first to receive the broadcast event. */
        for (i = n - 1; i > 0; i--) {
            int j = ast_random() % (i + 1);
            char *tmp = matching_arr[i];
            matching_arr[i] = matching_arr[j];
            matching_arr[j] = tmp;
        }
 
        /*
         * Second pass: send to each matching app.  A deep copy of the event
         * is required for each call because stasis_app_send() mutates the
         * message in-place (adds "asterisk_id" via ast_json_object_set).
         */
        for (i = 0; i < n; i++) {
            char *match_name = matching_arr[i];
            struct ast_json *event_copy;
 
            ast_debug(3, "Sending broadcast to app '%s'\n", match_name);
 
            event_copy = ast_json_deep_copy(event);
            if (!event_copy) {
                ast_log(LOG_ERROR,
                    "Channel %s: failed to deep-copy event for app '%s'\n",
                    ast_channel_uniqueid(chan), match_name);
                ao2_ref(match_name, -1);
                continue;
            }
 
            stasis_app_send(match_name, event_copy);
            ast_json_unref(event_copy);
            ao2_ref(match_name, -1);
        }
 
        ast_free(matching_arr);
    }
 
    return 0;
}

References ao2_bump, ao2_cleanup, ao2_container_count(), ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_ref, app_name(), ast_channel_caller(), ast_channel_exten(), ast_channel_lock, ast_channel_name(), ast_channel_snapshot_to_json(), ast_channel_uniqueid(), ast_channel_unlock, ast_debug, ast_free, ast_json_deep_copy(), ast_json_pack(), ast_json_timeval(), ast_json_unref(), ast_log, ast_malloc, ast_random(), ast_strdupa, ast_tvnow(), stasis_broadcast_ctx::compiled_filter, stasis_broadcast_ctx::filter_compiled, LOG_ERROR, NULL, RAII_VAR, S_OR, stasis_app_get_all(), and stasis_app_send().

Referenced by stasis_app_broadcast_channel().

stasis_app_broadcast_channel()

int AST_OPTIONAL_API_NAME() stasis_app_broadcast_channel	(	struct ast_channel *	chan,
		int	timeout_ms,
		const char *	app_filter,
		unsigned int	flags
	)

Start a broadcast for a channel.

Parameters

chan	The channel to broadcast
timeout_ms	Timeout in milliseconds
app_filter	Optional regex filter for applications

Returns

0 on success, -1 on error

Definition at line 550 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    struct ast_datastore *datastore;
 
    if (!chan) {
        return -1;
    }
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    /* Remove any previous broadcast datastore from a prior attempt.
     * This supports failover scenarios where StasisBroadcast() is
     * called multiple times for the same channel.  The datastore
     * destructor unlinks the old context from the container. */
    {
        struct ast_datastore *old_ds;
        ast_channel_lock(chan);
        old_ds = ast_channel_datastore_find(chan, &broadcast_datastore_info, NULL);
        if (old_ds) {
            ast_channel_datastore_remove(chan, old_ds);
        }
        ast_channel_unlock(chan);
        if (old_ds) {
            ast_datastore_free(old_ds);
        }
    }
 
    /* Create broadcast context (validates and compiles app_filter regex) */
    ctx = broadcast_ctx_create(ast_channel_uniqueid(chan), app_filter, flags);
    if (!ctx) {
        ast_log(LOG_ERROR, "Channel %s: failed to create broadcast context\n",
            ast_channel_uniqueid(chan));
        return -1;
    }
 
    /* Store context in container */
    ao2_link(broadcast_contexts, ctx);
 
    /* Create and attach datastore to channel */
    datastore = ast_datastore_alloc(&broadcast_datastore_info, ast_channel_uniqueid(chan));
    if (!datastore) {
        ast_log(LOG_ERROR, "Channel %s: failed to allocate broadcast datastore\n",
            ast_channel_uniqueid(chan));
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
 
    datastore->data = ao2_bump(ctx);
    ast_channel_lock(chan);
    if (ast_channel_datastore_add(chan, datastore)) {
        ast_channel_unlock(chan);
        ast_log(LOG_ERROR, "Channel %s: failed to attach broadcast datastore\n",
            ast_channel_uniqueid(chan));
        ast_datastore_free(datastore);
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
    ast_channel_unlock(chan);
 
    ast_debug(1, "Starting broadcast for channel %s (timeout: %dms, filter: %s)\n",
        ast_channel_uniqueid(chan), timeout_ms, app_filter ? app_filter : "none");
 
    /* Send broadcast event to all matching applications */
    if (send_broadcast_event(chan, ctx) != 0) {
        ast_log(LOG_ERROR, "Channel %s: failed to send broadcast event\n",
            ast_channel_uniqueid(chan));
        ast_channel_lock(chan);
        ast_channel_datastore_remove(chan, datastore);
        ast_channel_unlock(chan);
        ast_datastore_free(datastore);
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
 
    return 0;
}

References ao2_bump, ao2_cleanup, ao2_link, ao2_unlink, ast_channel_datastore_add(), ast_channel_datastore_find(), ast_channel_datastore_remove(), ast_channel_lock, ast_channel_uniqueid(), ast_channel_unlock, ast_datastore_alloc, ast_datastore_free(), ast_debug, ast_log, broadcast_contexts, broadcast_ctx_create(), broadcast_datastore_info, ast_datastore::data, LOG_ERROR, NULL, RAII_VAR, and send_broadcast_event().

Referenced by stasis_broadcast_exec().

stasis_app_broadcast_cleanup()

void AST_OPTIONAL_API_NAME() stasis_app_broadcast_cleanup

(

const char *

channel_id

)

Clean up broadcast context for a channel.

This is the normal-path cleanup called by the dialplan application after the broadcast completes. The channel datastore destructor (broadcast_datastore_destroy) also unlinks the context as a safety net for abnormal teardown; ao2_unlink is idempotent so the double call is harmless.

Parameters

channel_id

The unique ID of the channel

Definition at line 885 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
 
    if (ast_strlen_zero(channel_id) || !broadcast_contexts) {
        return;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
    if (ctx) {
        ast_debug(3, "Cleaning up broadcast context for %s\n", channel_id);
    }
}

References ao2_cleanup, ao2_find, ast_debug, ast_strlen_zero(), broadcast_contexts, NULL, OBJ_SEARCH_KEY, OBJ_UNLINK, and RAII_VAR.

Referenced by stasis_broadcast_exec().

stasis_app_broadcast_wait()

int AST_OPTIONAL_API_NAME() stasis_app_broadcast_wait	(	struct ast_channel *	chan,
		int	timeout_ms
	)

Wait for a broadcast channel to be claimed.

Blocks until the channel is claimed, the timeout expires, or the channel hangs up. The hangup check runs every BROADCAST_POLL_INTERVAL_MS so that a dead channel does not tie up a PBX thread for the full timeout period.

Parameters

chan	The channel
timeout_ms	Maximum time to wait in milliseconds

Returns

0 if claimed within timeout, -1 otherwise

Definition at line 782 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    const char *channel_id;
    struct timeval deadline;
    int result = -1;
 
    if (!chan) {
        return -1;
    }
 
    channel_id = ast_channel_uniqueid(chan);
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        ast_log(LOG_WARNING, "No broadcast context for channel %s\n", channel_id);
        return -1;
    }
 
    /* Cap excessive timeouts to prevent arithmetic overflow */
    if (timeout_ms < 0) {
        timeout_ms = 0;
    } else if (timeout_ms > MAX_BROADCAST_TIMEOUT_MS) {
        timeout_ms = MAX_BROADCAST_TIMEOUT_MS;
    }
 
    /* Calculate absolute deadline */
    deadline = ast_tvadd(ast_tvnow(),
        ast_tv(timeout_ms / 1000, (timeout_ms % 1000) * 1000));
 
    ao2_lock(ctx);
    while (!ctx->claimed) {
        struct timeval now;
        struct timespec poll_spec;
        long remaining_ms;
        long poll_ms;
        int wait_result;
 
        /* Check for hangup so we don't block on a dead channel */
        if (ast_check_hangup(chan)) {
            ast_debug(3, "Channel %s hung up during broadcast wait\n",
                channel_id);
            break;
        }
 
        /* Check if we've passed the overall deadline */
        now = ast_tvnow();
        remaining_ms = ast_tvdiff_ms(deadline, now);
        if (remaining_ms <= 0) {
            ast_debug(3, "Broadcast timeout for channel %s after %dms\n",
                channel_id, timeout_ms);
            break;
        }
 
        /* Sleep for the shorter of the remaining time and the poll interval */
        poll_ms = remaining_ms;
        if (poll_ms > BROADCAST_POLL_INTERVAL_MS) {
            poll_ms = BROADCAST_POLL_INTERVAL_MS;
        }
 
        poll_spec.tv_sec = now.tv_sec + (poll_ms / 1000);
        poll_spec.tv_nsec = (long)(now.tv_usec) * 1000L
            + (long)(poll_ms % 1000) * 1000000L;
        while (poll_spec.tv_nsec >= 1000000000) {
            poll_spec.tv_sec++;
            poll_spec.tv_nsec -= 1000000000;
        }
 
        wait_result = ast_cond_timedwait(&ctx->cond, ao2_object_get_lockaddr(ctx), &poll_spec);
        if (wait_result != 0 && wait_result != ETIMEDOUT) {
            ast_log(LOG_WARNING,
                "Channel %s: unexpected error waiting for claim: %s (%d)\n",
                channel_id, strerror(wait_result), wait_result);
            break;
        }
        /* Loop back: re-check claimed, then hangup, then deadline */
    }
 
    if (ctx->claimed) {
        ast_debug(1, "Channel %s claimed by %s\n",
            channel_id, ctx->winner_app);
        result = 0;
    }
    ao2_unlock(ctx);
 
    return result;
}

References ao2_cleanup, ao2_find, ao2_lock, ao2_object_get_lockaddr(), ao2_unlock, ast_channel_uniqueid(), ast_check_hangup(), ast_cond_timedwait, ast_debug, ast_log, ast_tv(), ast_tvadd(), ast_tvdiff_ms(), ast_tvnow(), broadcast_contexts, BROADCAST_POLL_INTERVAL_MS, LOG_WARNING, MAX_BROADCAST_TIMEOUT_MS, NULL, OBJ_SEARCH_KEY, RAII_VAR, result, and wait_result().

Referenced by stasis_broadcast_exec().

stasis_app_broadcast_winner()

char *AST_OPTIONAL_API_NAME() stasis_app_broadcast_winner

(

const char *

channel_id

)

Get the winner app name for a broadcast channel.

Parameters

channel_id

The unique ID of the channel

Returns

A copy of the winner app name (caller must free with ast_free), or NULL if not claimed or not found

Definition at line 743 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    char *winner = NULL;
 
    if (ast_strlen_zero(channel_id)) {
        return NULL;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        return NULL;
    }
 
    ao2_lock(ctx);
    if (ctx->claimed) {
        winner = ast_strdup(ctx->winner_app);
    }
    /* Mark the broadcast as finished so no new claims can succeed.
     * This closes the race window between reading the winner and
     * the subsequent broadcast_cleanup call. */
    ctx->finished = 1;
    ao2_unlock(ctx);
 
    return winner;
}

References ao2_cleanup, ao2_find, ao2_lock, ao2_unlock, ast_strdup, ast_strlen_zero(), broadcast_contexts, NULL, OBJ_SEARCH_KEY, and RAII_VAR.

Referenced by stasis_broadcast_exec().

stasis_app_claim_channel()

int AST_OPTIONAL_API_NAME() stasis_app_claim_channel	(	const char *	channel_id,
		const char *	app_name
	)

Attempt to claim a broadcast channel.

Parameters

channel_id	The unique ID of the channel
app_name	The name of the application claiming the channel

Returns

0 if claim successful, -1 if channel not found, -2 if already claimed

Definition at line 637 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
 
    if (ast_strlen_zero(channel_id) || ast_strlen_zero(app_name)) {
        return -1;
    }
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    /* Find broadcast context */
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        ast_debug(1, "No broadcast context found for channel %s\n", channel_id);
        return -1;
    }
 
    /* Atomically check and set claimed flag.
     * Check claimed before finished: if the channel was claimed and then the
     * broadcast finished, a late claim should return -2 (409 Conflict) rather
     * than -1 (404) so callers can distinguish "already taken" from "not found". */
    ao2_lock(ctx);
    if (ctx->claimed) {
        ast_debug(1, "Channel %s already claimed by %s (attempt by %s denied)\n",
            channel_id, ctx->winner_app ? ctx->winner_app : "(unknown)", app_name);
        ao2_unlock(ctx);
        return -2;
    }
    if (ctx->finished) {
        ast_debug(1, "Channel %s broadcast already finished (late claim by %s rejected)\n",
            channel_id, app_name);
        ao2_unlock(ctx);
        return -1;
    }
    ctx->winner_app = ast_strdup(app_name);
    if (!ctx->winner_app) {
        ast_log(LOG_ERROR,
            "Failed to allocate winner app name for channel %s\n",
            channel_id);
        ao2_unlock(ctx);
        return -1;
    }
    ctx->claimed = 1;
    ast_verb(3, "Channel %s claimed by application %s\n",
        channel_id, app_name);
    /* Signal waiting thread that channel was claimed */
    ast_cond_signal(&ctx->cond);
    ao2_unlock(ctx);
 
    /* Send CallClaimed event to matching apps */
    if (!(ctx->flags & STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED)) {
        RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
        RAII_VAR(struct ast_json *, event, NULL, ast_json_unref);
        RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
        struct ao2_iterator iter;
        char *app_name_iter;
 
        snapshot = ast_channel_snapshot_get_latest(channel_id);
        if (snapshot) {
            event = ast_json_pack("{s: s, s: o, s: o, s: s}",
                "type", "CallClaimed",
                "timestamp", ast_json_timeval(ast_tvnow(), NULL),
                "channel", ast_channel_snapshot_to_json(snapshot, NULL),
                "winner_app", app_name);
        }
        if (event) {
            apps = stasis_app_get_all();
        }
        if (apps) {
            iter = ao2_iterator_init(apps, 0);
            while ((app_name_iter = ao2_iterator_next(&iter))) {
                struct ast_json *event_copy;
 
                /* Only send to apps that matched the original broadcast filter */
                if (ctx->filter_compiled &&
                    regexec(&ctx->compiled_filter, app_name_iter,
                        0, NULL, 0) == REG_NOMATCH) {
                    ao2_ref(app_name_iter, -1);
                    continue;
                }
 
                event_copy = ast_json_deep_copy(event);
                if (!event_copy) {
                    ao2_ref(app_name_iter, -1);
                    continue;
                }
 
                stasis_app_send(app_name_iter, event_copy);
                ast_json_unref(event_copy);
                ao2_ref(app_name_iter, -1);
            }
            ao2_iterator_destroy(&iter);
        }
    }
 
    return 0;
}

References ao2_cleanup, ao2_find, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_lock, ao2_ref, ao2_unlock, app_name(), ast_channel_snapshot_get_latest(), ast_channel_snapshot_to_json(), ast_cond_signal, ast_debug, ast_json_deep_copy(), ast_json_pack(), ast_json_timeval(), ast_json_unref(), ast_log, ast_strdup, ast_strlen_zero(), ast_tvnow(), ast_verb, broadcast_contexts, LOG_ERROR, NULL, OBJ_SEARCH_KEY, RAII_VAR, stasis_app_get_all(), stasis_app_send(), and STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED.

Referenced by ast_ari_events_claim_channel().

unload_module()

static int unload_module ( void  )

static

Definition at line 912 of file res_stasis_broadcast.c.

{
    /* NULL the global pointer before releasing the reference so that
     * concurrent lookups see NULL (safe) rather than a freed pointer. */
    {
        struct ao2_container *old_contexts = broadcast_contexts;
        broadcast_contexts = NULL;
        ao2_cleanup(old_contexts);
    }
 
    ast_debug(1, "Stasis broadcast module unloaded\n");
    return 0;
}

References ao2_cleanup, ast_debug, broadcast_contexts, and NULL.

validate_regex_pattern()

static int validate_regex_pattern ( const char *  pattern )

static

Validate a regex pattern for safety.

Checks that the regex pattern is within length limits and doesn't contain patterns that could cause excessive backtracking or denial of service.

Parameters

pattern

The regex pattern to validate

Returns

0 if valid, -1 if invalid

Definition at line 207 of file res_stasis_broadcast.c.

{
    size_t len;
    int group_depth = 0;
    int quantified_groups = 0;
    int in_class = 0; /* Inside [...] */
    /* Track alternations per group depth. Index 0 is outside groups and unused. */
    int alternations_per_depth[MAX_GROUP_DEPTH + 1] = { 0 };
    const char *p;
 
    if (ast_strlen_zero(pattern)) {
        return 0; /* Empty pattern is valid (will be skipped) */
    }
 
    /* Check maximum length to prevent excessive regex compilation time */
    len = strlen(pattern);
    if (len > MAX_REGEX_LENGTH) {
        ast_debug(3, "Regex pattern exceeds maximum length of %d characters (got %zu)\n",
            MAX_REGEX_LENGTH, len);
        return -1;
    }
 
    /*
     * Check for potentially dangerous patterns that could cause
     * excessive regex compilation or matching time. Look for:
     * - Excessive group nesting depth
     * - Too many quantified groups (groups followed by +, *, or ?)
     *
     * Note: This is a heuristic approach that catches common dangerous
     * patterns. Combined with the length limit, it provides reasonable
     * protection against ReDoS while allowing legitimate regex usage.
     */
    for (p = pattern; *p; p++) {
        /* Handle character classes: enter on unescaped '[' and exit on unescaped ']' */
        if (!in_class && *p == '[' && (p == pattern || *(p - 1) != '\\')) {
            in_class = 1;
            /* In POSIX ERE, ']' immediately after '[' or '[^' is a
             * literal, not the end of the class.  Advance past the
             * optional negation caret and the literal ']' so the
             * main loop does not leave in_class prematurely. */
            if (*(p + 1) == '^') {
                p++;
            }
            if (*(p + 1) == ']') {
                p++;
            }
            continue;
        } else if (in_class) {
            if (*p == '\\') {
                /* Skip the next escaped character inside character class */
                if (*(p + 1)) {
                    p++;
                }
                continue;
            }
            if (*p == ']') {
                in_class = 0;
            }
            /* Ignore everything inside character classes for heuristics */
            continue;
        }
        switch (*p) {
        case '(':
            group_depth++;
            if (group_depth > MAX_GROUP_DEPTH) {
                ast_debug(3, "Regex pattern has too many nested groups (max %d)\n",
                    MAX_GROUP_DEPTH);
                return -1;
            }
            /* Reset alternation counter for newly entered group depth */
            alternations_per_depth[group_depth] = 0;
            break;
        case ')':
            if (group_depth > 0) {
                /* Clear alternations count for this depth before leaving */
                alternations_per_depth[group_depth] = 0;
                group_depth--;
            }
            break;
        case '+':
        case '*':
        case '?':
            /*
             * Count quantified groups - patterns like (...)+ or (...)*
             * Too many of these can cause slow matching on certain inputs.
             */
            if (p > pattern && *(p - 1) == ')') {
                quantified_groups++;
            }
            break;
        case '{': {
            /* Parse POSIX quantifier {m}, {m,}, {m,n} with overflow and bound checks */
            const char *q = p + 1;
            long m = 0, n = -1; /* n=-1 means open upper bound */
            int valid = 0;
            int digit;
            int overflow = 0;
 
            if (*q >= '0' && *q <= '9') {
                /* Parse m safely */
                while (*q >= '0' && *q <= '9') {
                    digit = (*q - '0');
                    if (m > (LONG_MAX - digit) / 10) { /* overflow on next step */
                        overflow = 1;
                        break;
                    }
                    m = (m * 10) + digit;
                    if (m > MAX_QUANTIFIER_BOUND) { /* early bound exceed */
                        overflow = 1;
                        break;
                    }
                    q++;
                }
                if (!overflow && *q == ',') {
                    q++;
                    if (*q >= '0' && *q <= '9') {
                        long nn = 0;
                        while (*q >= '0' && *q <= '9') {
                            digit = (*q - '0');
                            if (nn > (LONG_MAX - digit) / 10) {
                                overflow = 1;
                                break;
                            }
                            nn = (nn * 10) + digit;
                            if (nn > MAX_QUANTIFIER_BOUND) {
                                overflow = 1;
                                break;
                            }
                            q++;
                        }
                        n = nn;
                    } else {
                        n = -1; /* open upper bound */
                    }
                } else if (!overflow) {
                    n = m; /* {m} */
                }
                if (!overflow && *q == '}') {
                    valid = 1;
                }
            }
            if (overflow) {
                ast_debug(3, "Regex quantifier overflow or exceeds max bound (max %d)\n", MAX_QUANTIFIER_BOUND);
                return -1;
            }
            if (valid) {
                /* Additional bounds check (defensive) */
                if (m > MAX_QUANTIFIER_BOUND || (n != -1 && n > MAX_QUANTIFIER_BOUND)) {
                    ast_debug(3, "Regex quantifier bounds too large (max %d)\n", MAX_QUANTIFIER_BOUND);
                    return -1;
                }
                if (p > pattern && *(p - 1) == ')') {
                    quantified_groups++;
                }
                p = q; /* q currently points to '}' */
            }
            break;
        }
        case '|':
            if (group_depth > 0) {
                alternations_per_depth[group_depth]++;
                if (group_depth > ALTERNATION_DEPTH_THRESHOLD &&
                    alternations_per_depth[group_depth] > MAX_ALTERNATIONS) {
                    ast_debug(3,
                        "Regex has too many alternations in deep group (depth %d, count %d, max %d)\n",
                        group_depth,
                        alternations_per_depth[group_depth],
                        MAX_ALTERNATIONS);
                    return -1;
                }
            }
            break;
        case '\\':
            /*
             * Skip the next character entirely from heuristic processing.
             * This ensures escaped characters (metacharacters in BRE or literals
             * in ERE like \‍(, \‍), \+, \*, \?, etc.) do not affect group depth
             * or quantified group counts.
             */
            if (*(p + 1)) {
                p++;
            }
            /* Continue to next loop iteration without evaluating the escaped char */
            continue;
        }
    }
 
    /*
     * Reject patterns with too many quantified groups, as these are
     * often indicators of potentially slow patterns that could be
     * exploited for denial of service.
     */
    if (quantified_groups > MAX_NESTED_QUANTIFIERS) {
        ast_debug(3, "Regex pattern has too many quantified groups (max %d)\n",
            MAX_NESTED_QUANTIFIERS);
        return -1;
    }
 
    return 0;
}

References ALTERNATION_DEPTH_THRESHOLD, ast_debug, ast_strlen_zero(), digit, len(), MAX_ALTERNATIONS, MAX_GROUP_DEPTH, MAX_NESTED_QUANTIFIERS, MAX_QUANTIFIER_BOUND, and MAX_REGEX_LENGTH.

Referenced by broadcast_ctx_create().

Variable Documentation

__mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER , .description = "Stasis application broadcast" , .key = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .support_level = AST_MODULE_SUPPORT_EXTENDED, .load = load_module, .unload = unload_module, .requires = "res_stasis,res_ari,http", .load_pri = AST_MODPRI_APP_DEPEND - 1, }

static

Definition at line 933 of file res_stasis_broadcast.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 933 of file res_stasis_broadcast.c.

broadcast_contexts

struct ao2_container* broadcast_contexts

static

Container for all active broadcast contexts.

Definition at line 104 of file res_stasis_broadcast.c.

Referenced by broadcast_ctx_create(), load_module(), stasis_app_broadcast_channel(), stasis_app_broadcast_cleanup(), stasis_app_broadcast_wait(), stasis_app_broadcast_winner(), stasis_app_claim_channel(), and unload_module().

broadcast_datastore_info

const struct ast_datastore_info broadcast_datastore_info

static

Initial value:

= {
    .type = "stasis_broadcast_context",
    .destroy = broadcast_datastore_destroy,
}

Datastore information for broadcast context.

Definition at line 123 of file res_stasis_broadcast.c.

                                                                  {
    .type = "stasis_broadcast_context",
    .destroy = broadcast_datastore_destroy,
};

Referenced by stasis_app_broadcast_channel().