Prequisites for ARI Outbound Websockets

stasis:
* Added stasis_app_is_registered().
* Added stasis_app_control_mark_failed().
* Added stasis_app_control_is_failed().
* Fixed res_stasis_device_state so unsubscribe all works properly.
* Modified stasis_app_unregister() to unsubscribe from all event sources.
* Modified stasis_app_exec to return -1 if stasis_app_control_is_failed()
  returns true.

http:
* Added ast_http_create_basic_auth_header().

md5:
* Added define for MD5_DIGEST_LENGTH.

tcptls:
* Added flag to ast_tcptls_session_args to suppress connection log messages
  to give callers more control over logging.

http_websocket:
* Add flag to ast_websocket_client_options to suppress connection log messages
  to give callers more control over logging.
* Added username and password to ast_websocket_client_options to support
  outbound basic authentication.
* Added ast_websocket_result_to_str().

include/asterisk/http.h

@@ -141,6 +141,20 @@ struct ast_http_auth {
  */
 struct ast_http_auth *ast_http_get_auth(struct ast_variable *headers);
 
+/*!
+ * \brief Create an HTTP authorization header.
+ *
+ * The returned ast_variable must be freed with ast_variables_destroy()
+ *
+ * \param userid User ID or "<userid>:<password>".
+ * \param password Password if not in userid.
+ *
+ * \return ast_variable with name="Authorization" and value="Basic <base64enc>"
+ * \retval NULL if memory allocation failed.
+ */
+struct ast_variable *ast_http_create_basic_auth_header(const char *userid,
+	const char *password);
+
 /*! \brief Register a URI handler */
 int ast_http_uri_link(struct ast_http_uri *urihandler);
 

include/asterisk/http_websocket.h

@@ -397,7 +397,7 @@ AST_OPTIONAL_API(const char *, ast_websocket_session_id, (struct ast_websocket *
  * \brief Result code for a websocket client.
  */
 enum ast_websocket_result {
-	WS_OK,
+	WS_OK = 0,
 	WS_ALLOCATE_ERROR,
 	WS_KEY_ERROR,
 	WS_URI_PARSE_ERROR,
@@ -411,6 +411,7 @@ enum ast_websocket_result {
 	WS_NOT_SUPPORTED,
 	WS_WRITE_ERROR,
 	WS_CLIENT_START_ERROR,
+	WS_UNAUTHORIZED,
 };
 
 /*!
@@ -468,6 +469,9 @@ struct ast_websocket_client_options {
 	 * Secure websocket credentials
 	 */
 	struct ast_tls_config *tls_cfg;
+	const char *username;          /*!< Auth username */
+	const char *password;          /*!< Auth password */
+	int suppress_connection_msgs;  /*!< Suppress connection log messages */
 };
 
 /*!
@@ -510,4 +514,13 @@ AST_OPTIONAL_API(const char *, ast_websocket_client_accept_protocol,
  */
 AST_OPTIONAL_API(int, ast_websocket_set_timeout, (struct ast_websocket *session, int timeout), {return -1;});
 
+/*!
+ * \brief Convert a websocket result code to a string.
+ *
+ * \param result The result code to convert
+ *
+ * \return A string representation of the result code
+ */
+AST_OPTIONAL_API(const char *, ast_websocket_result_to_str, (enum ast_websocket_result result), {return "";});
+
 #endif

include/asterisk/md5.h

@@ -23,6 +23,10 @@
 #ifndef _ASTERISK_MD5_H
 #define _ASTERISK_MD5_H
 
+#ifndef MD5_DIGEST_LENGTH
+#define MD5_DIGEST_LENGTH 16
+#endif
+
 struct MD5Context {
 	uint32_t buf[4];
 	uint32_t bits[2];
@@ -33,7 +37,7 @@ struct MD5Context {
 void MD5Init(struct MD5Context *context);
 void MD5Update(struct MD5Context *context, unsigned char const *buf,
 	       unsigned len);
-void MD5Final(unsigned char digest[16], struct MD5Context *context);
+void MD5Final(unsigned char digest[MD5_DIGEST_LENGTH], struct MD5Context *context);
 void MD5Transform(uint32_t buf[4], uint32_t const in[16]);
 
 #endif /* _ASTERISK_MD5_H */

include/asterisk/stasis_app.h

@@ -85,6 +85,16 @@ struct ao2_container *stasis_app_get_all(void);
  */
 struct stasis_app *stasis_app_get_by_name(const char *name);
 
+/*!
+ * \brief Check if a Stasis application is registered.
+ *
+ * \param name The name of the registered Stasis application
+ *
+ * \return 1 if the application is registered.
+ * \return 0 if the application is not registered.
+ */
+int stasis_app_is_registered(const char *name);
+
 /*!
  * \brief Register a new Stasis application.
  *
@@ -116,7 +126,7 @@ int stasis_app_register(const char *app_name, stasis_app_cb handler, void *data)
 int stasis_app_register_all(const char *app_name, stasis_app_cb handler, void *data);
 
 /*!
- * \brief Unregister a Stasis application.
+ * \brief Unregister a Stasis application and unsubscribe from all event sources.
  * \param app_name Name of the application to unregister.
  */
 void stasis_app_unregister(const char *app_name);
@@ -447,6 +457,20 @@ void stasis_app_control_execute_until_exhausted(
 int stasis_app_control_is_done(
 	struct stasis_app_control *control);
 
+/*!
+ * \brief Set the failed flag on a control structure
+ *
+ * \param control Control object to be updated
+ */
+void stasis_app_control_mark_failed(struct stasis_app_control *control);
+
+/*!
+ * \brief Check if a control object is marked as "failed"
+ *
+ * \param control Control object to check
+ */
+int stasis_app_control_is_failed(const struct stasis_app_control *control);
+
 /*!
  * \brief Flush the control command queue.
  * \since 13.9.0

include/asterisk/tcptls.h

@@ -142,6 +142,7 @@ struct ast_tcptls_session_args {
 	void *(*worker_fn)(void *); /*!< the function in charge of doing the actual work */
 	const char *name;
 	struct ast_tls_config *old_tls_cfg; /*!< copy of the SSL configuration to determine whether changes have been made */
+	int suppress_connection_msgs; /*!< suppress connection messages to allow caller to manage logging */
 };
 
 /*! \brief

main/http.c

@@ -1665,6 +1665,50 @@ struct ast_http_auth *ast_http_get_auth(struct ast_variable *headers)
 	return NULL;
 }
 
+struct ast_variable *ast_http_create_basic_auth_header(const char *userid,
+	const char *password)
+{
+	int encoded_size = 0;
+	int userinfo_len = 0;
+	RAII_VAR(char *, userinfo, NULL, ast_free);
+	char *encoded_userinfo = NULL;
+	struct ast_variable *auth_header = NULL;
+
+	if (ast_strlen_zero(userid)) {
+		return NULL;
+	}
+
+	if (strchr(userid, ':')) {
+		userinfo = ast_strdup(userid);
+		userinfo_len = strlen(userinfo);
+	} else {
+		if (ast_strlen_zero(password)) {
+			return NULL;
+		}
+		userinfo_len = ast_asprintf(&userinfo, "%s:%s", userid, password);
+	}
+	if (!userinfo) {
+		return NULL;
+	}
+
+	/*
+	 * The header value is "Basic " + base64(userinfo).
+	 * Doubling the userinfo length then adding the length
+	 * of the "Basic " prefix is a conservative estimate of the
+	 * final encoded size.
+	 */
+	encoded_size = userinfo_len * 2 * sizeof(char) + 1 + BASIC_LEN;
+	encoded_userinfo = ast_alloca(encoded_size);
+	strcpy(encoded_userinfo, BASIC_PREFIX); /* Safe */
+	ast_base64encode(encoded_userinfo + BASIC_LEN, (unsigned char *)userinfo,
+		userinfo_len, encoded_size - BASIC_LEN);
+
+	auth_header = ast_variable_new("Authorization",
+		encoded_userinfo, "");
+
+	return auth_header;
+}
+
 int ast_http_response_status_line(const char *buf, const char *version, int code)
 {
 	int status_code;

main/md5.c

@@ -117,7 +117,7 @@ void MD5Update(struct MD5Context *ctx, unsigned char const *buf, unsigned len)
  * Final wrapup - pad to 64-byte boundary with the bit pattern
  * 1 0* (64-bit count of bits processed, MSB-first)
  */
-void MD5Final(unsigned char digest[16], struct MD5Context *ctx)
+void MD5Final(unsigned char digest[MD5_DIGEST_LENGTH], struct MD5Context *ctx)
 {
 	unsigned count;
 	unsigned char *p;

main/tcptls.c

@@ -379,7 +379,8 @@ static void __ssl_setup_certs(struct ast_tls_config *cfg, const size_t cert_file
 }
 #endif
 
-static int __ssl_setup(struct ast_tls_config *cfg, int client)
+static int __ssl_setup(struct ast_tls_config *cfg, int client,
+	int suppress_progress_msgs)
 {
 #ifndef DO_SSL
 	if (cfg->enabled) {
@@ -534,8 +535,10 @@ static int __ssl_setup(struct ast_tls_config *cfg, int client)
 				if (SSL_CTX_set_tmp_dh(cfg->ssl_ctx, dh)) {
 					long options = SSL_OP_CIPHER_SERVER_PREFERENCE | SSL_OP_SINGLE_DH_USE | SSL_OP_SINGLE_ECDH_USE;
 					options = SSL_CTX_set_options(cfg->ssl_ctx, options);
+					if (!suppress_progress_msgs) {
 						ast_verb(2, "TLS/SSL DH initialized, PFS cipher-suites enabled\n");
 					}
+				}
 				DH_free(dh);
 			}
 			BIO_free(bio);
@@ -548,7 +551,9 @@ static int __ssl_setup(struct ast_tls_config *cfg, int client)
 	#endif
 	/* SSL_CTX_set_ecdh_auto(cfg->ssl_ctx, on); requires OpenSSL 1.0.2 which wraps: */
 	if (SSL_CTX_ctrl(cfg->ssl_ctx, SSL_CTRL_SET_ECDH_AUTO, 1, NULL)) {
+		if (!suppress_progress_msgs) {
 			ast_verb(2, "TLS/SSL ECDH initialized (automatic), faster PFS ciphers enabled\n");
+		}
 #if !defined(OPENSSL_NO_ECDH) && (OPENSSL_VERSION_NUMBER >= 0x10000000L) && (OPENSSL_VERSION_NUMBER < 0x10100000L)
 	} else {
 		/* enables AES-128 ciphers, to get AES-256 use NID_secp384r1 */
@@ -562,14 +567,16 @@ static int __ssl_setup(struct ast_tls_config *cfg, int client)
 #endif
 	}
 
+	if (!suppress_progress_msgs) {
 		ast_verb(2, "TLS/SSL certificate ok\n");	/* We should log which one that is ok. This message doesn't really make sense in production use */
+	}
 	return 1;
 #endif
 }
 
 int ast_ssl_setup(struct ast_tls_config *cfg)
 {
-	return __ssl_setup(cfg, 0);
+	return __ssl_setup(cfg, 0, 0);
 }
 
 void ast_ssl_teardown(struct ast_tls_config *cfg)
@@ -653,8 +660,10 @@ struct ast_tcptls_session_instance *ast_tcptls_client_start_timeout(
 	}
 
 	if (socket_connect(desc->accept_fd, &desc->remote_address, timeout)) {
+		if (!desc->suppress_connection_msgs) {
 			ast_log(LOG_WARNING, "Unable to connect %s to %s: %s\n", desc->name,
 					ast_sockaddr_stringify(&desc->remote_address), strerror(errno));
+		}
 
 		ao2_ref(tcptls_session, -1);
 		return NULL;
@@ -663,8 +672,7 @@ struct ast_tcptls_session_instance *ast_tcptls_client_start_timeout(
 	ast_fd_clear_flags(desc->accept_fd, O_NONBLOCK);
 
 	if (desc->tls_cfg) {
-		desc->tls_cfg->enabled = 1;
-		__ssl_setup(desc->tls_cfg, 1);
+		__ssl_setup(desc->tls_cfg, 1, desc->suppress_connection_msgs);
 	}
 
 	return handle_tcptls_connection(tcptls_session);

res/res_http_websocket.c

@@ -1091,7 +1091,8 @@ int AST_OPTIONAL_API_NAME(ast_websocket_remove_protocol)(const char *name, ast_w
  * The returned host will contain the address and optional port while
  * path will contain everything after the address/port if included.
  */
-static int websocket_client_parse_uri(const char *uri, char **host, struct ast_str **path)
+static int websocket_client_parse_uri(const char *uri, char **host,
+	struct ast_str **path, char **userinfo)
 {
 	struct ast_uri *parsed_uri = ast_uri_parse_websocket(uri);
 
@@ -1100,6 +1101,7 @@ static int websocket_client_parse_uri(const char *uri, char **host, struct ast_s
 	}
 
 	*host = ast_uri_make_host_with_port(parsed_uri);
+	*userinfo = ast_strdup(ast_uri_user_info(parsed_uri));
 
 	if (ast_uri_path(parsed_uri) || ast_uri_query(parsed_uri)) {
 		*path = ast_str_create(64);
@@ -1212,6 +1214,10 @@ struct websocket_client {
 	struct ast_tcptls_session_args *args;
 	/*! tcptls connection instance */
 	struct ast_tcptls_session_instance *ser;
+	/*! Authentication userid:password */
+	char *userinfo;
+	/*! Suppress connection log messages */
+	int suppress_connection_msgs;
 };
 
 static void websocket_client_destroy(void *obj)
@@ -1226,6 +1232,7 @@ static void websocket_client_destroy(void *obj)
 	ast_free(client->key);
 	ast_free(client->resource_name);
 	ast_free(client->host);
+	ast_free(client->userinfo);
 }
 
 static struct ast_websocket * websocket_client_create(
@@ -1239,9 +1246,17 @@ static struct ast_websocket * websocket_client_create(
 		return NULL;
 	}
 
+	if (!ast_uuid_generate_str(ws->session_id, sizeof(ws->session_id))) {
+		ast_log(LOG_ERROR, "Unable to allocate websocket session_id\n");
+		ao2_ref(ws, -1);
+		*result = WS_ALLOCATE_ERROR;
+		return NULL;
+	}
+
 	if (!(ws->client = ao2_alloc(
 		      sizeof(*ws->client), websocket_client_destroy))) {
 		ast_log(LOG_ERROR, "Unable to allocate websocket client\n");
+		ao2_ref(ws, -1);
 		*result = WS_ALLOCATE_ERROR;
 		return NULL;
 	}
@@ -1253,22 +1268,34 @@ static struct ast_websocket * websocket_client_create(
 	}
 
 	if (websocket_client_parse_uri(
-		    options->uri, &ws->client->host, &ws->client->resource_name)) {
+			options->uri, &ws->client->host, &ws->client->resource_name,
+			&ws->client->userinfo)) {
 		ao2_ref(ws, -1);
 		*result = WS_URI_PARSE_ERROR;
 		return NULL;
 	}
 
+	if (ast_strlen_zero(ws->client->userinfo)
+		&& !ast_strlen_zero(options->username)
+		&& !ast_strlen_zero(options->password)) {
+		ast_asprintf(&ws->client->userinfo, "%s:%s", options->username,
+			options->password);
+	}
+
 	if (!(ws->client->args = websocket_client_args_create(
 		      ws->client->host, options->tls_cfg, result))) {
 		ao2_ref(ws, -1);
 		return NULL;
 	}
-	ws->client->protocols = ast_strdup(options->protocols);
 
+	ws->client->suppress_connection_msgs = options->suppress_connection_msgs;
+	ws->client->args->suppress_connection_msgs = options->suppress_connection_msgs;
+	ws->client->protocols = ast_strdup(options->protocols);
 	ws->client->version = 13;
 	ws->opcode = -1;
 	ws->reconstruct = DEFAULT_RECONSTRUCTION_CEILING;
+	ws->timeout = options->timeout;
+
 	return ws;
 }
 
@@ -1289,17 +1316,29 @@ static enum ast_websocket_result websocket_client_handle_response_code(
 	case 101:
 		return 0;
 	case 400:
+		if (!client->suppress_connection_msgs) {
 			ast_log(LOG_ERROR, "Received response 400 - Bad Request "
 				"- from %s\n", client->host);
+		}
 		return WS_BAD_REQUEST;
+	case 401:
+		if (!client->suppress_connection_msgs) {
+			ast_log(LOG_ERROR, "Received response 401 - Unauthorized "
+				"- from %s\n", client->host);
+		}
+		return WS_UNAUTHORIZED;
 	case 404:
+		if (!client->suppress_connection_msgs) {
 			ast_log(LOG_ERROR, "Received response 404 - Request URL not "
 				"found - from %s\n", client->host);
+		}
 		return WS_URL_NOT_FOUND;
 	}
 
+	if (!client->suppress_connection_msgs) {
 		ast_log(LOG_ERROR, "Invalid HTTP response code %d from %s\n",
 			response_code, client->host);
+	}
 	return WS_INVALID_RESPONSE;
 }
 
@@ -1384,29 +1423,49 @@ static enum ast_websocket_result websocket_client_handshake_get_response(
 		WS_OK : WS_HEADER_MISSING;
 }
 
+#define optional_header_spec "%s%s%s"
+#define print_optional_header(test, name, value) \
+	test ? name : "", \
+	test ? value : "", \
+	test ? "\r\n" : ""
+
 static enum ast_websocket_result websocket_client_handshake(
 	struct websocket_client *client)
 {
-	char protocols[100] = "";
+	size_t protocols_len = 0;
+	struct ast_variable *auth_header = NULL;
+	size_t res;
 
-	if (!ast_strlen_zero(client->protocols)) {
-		sprintf(protocols, "Sec-WebSocket-Protocol: %s\r\n",
-			client->protocols);
+	if (!ast_strlen_zero(client->userinfo)) {
+		auth_header = ast_http_create_basic_auth_header(client->userinfo, NULL);
+		if (!auth_header) {
+			ast_log(LOG_ERROR, "Unable to allocate client websocket userinfo\n");
+			return WS_ALLOCATE_ERROR;
 		}
+	}
+
+	protocols_len = client->protocols ? strlen(client->protocols) : 0;
 
-	if (ast_iostream_printf(client->ser->stream,
+	res = ast_iostream_printf(client->ser->stream,
 		"GET /%s HTTP/1.1\r\n"
 		"Sec-WebSocket-Version: %d\r\n"
 		"Upgrade: websocket\r\n"
 		"Connection: Upgrade\r\n"
 		"Host: %s\r\n"
+		optional_header_spec
+		optional_header_spec
 		"Sec-WebSocket-Key: %s\r\n"
-			"%s\r\n",
+		"\r\n",
 		client->resource_name ? ast_str_buffer(client->resource_name) : "",
 		client->version,
 		client->host,
-			client->key,
-			protocols) < 0) {
+		print_optional_header(auth_header, "Authorization: ", auth_header->value),
+		print_optional_header(protocols_len, "Sec-WebSocket-Protocol: ", client->protocols),
+		client->key
+	);
+
+	ast_variables_destroy(auth_header);
+	if (res < 0) {
 		ast_log(LOG_ERROR, "Failed to send handshake.\n");
 		return WS_WRITE_ERROR;
 	}
@@ -1530,6 +1589,33 @@ int AST_OPTIONAL_API_NAME(ast_websocket_write_string)
 				   (char *)buf, len);
 }
 
+const char *websocket_result_string_map[] = {
+	[WS_OK] = "OK",
+	[WS_ALLOCATE_ERROR] = "Allocation error",
+	[WS_KEY_ERROR] = "Key error",
+	[WS_URI_PARSE_ERROR] = "URI parse error",
+	[WS_URI_RESOLVE_ERROR] = "URI resolve error",
+	[WS_BAD_STATUS] = "Bad status line",
+	[WS_INVALID_RESPONSE] = "Invalid response code",
+	[WS_BAD_REQUEST] = "Bad request",
+	[WS_URL_NOT_FOUND] = "URL not found",
+	[WS_HEADER_MISMATCH] = "Header mismatch",
+	[WS_HEADER_MISSING] = "Header missing",
+	[WS_NOT_SUPPORTED] = "Not supported",
+	[WS_WRITE_ERROR] = "Write error",
+	[WS_CLIENT_START_ERROR] = "Client start error",
+	[WS_UNAUTHORIZED] = "Unauthorized"
+};
+
+const char *AST_OPTIONAL_API_NAME(ast_websocket_result_to_str)
+	(enum ast_websocket_result result)
+{
+	if (!ARRAY_IN_BOUNDS(result, websocket_result_string_map)) {
+		return "unknown";
+	}
+	return websocket_result_string_map[result];
+}
+
 static int load_module(void)
 {
 	websocketuri.data = websocket_server_internal_create();

res/res_stasis.c

@@ -1667,6 +1667,9 @@ int stasis_app_exec(struct ast_channel *chan, const char *app_name, int argc,
 	 */
 	cleanup();
 
+	 if (stasis_app_control_is_failed(control)) {
+		 res = -1;
+	 }
 	/* The control needs to be removed from the controls container in
 	 * case a new PBX is started and ends up coming back into Stasis.
 	 */
@@ -1741,6 +1744,19 @@ struct stasis_app *stasis_app_get_by_name(const char *name)
 	return find_app_by_name(name);
 }
 
+int stasis_app_is_registered(const char *name)
+{
+	struct stasis_app *app = find_app_by_name(name);
+
+	/*
+	 * It's safe to unref app here because we're not actually
+	 * using it or returning it.
+	 */
+	ao2_cleanup(app);
+
+	return app != NULL;
+}
+
 static int append_name(void *obj, void *arg, int flags)
 {
 	struct stasis_app *app = obj;
@@ -1832,6 +1848,8 @@ int stasis_app_register_all(const char *app_name, stasis_app_cb handler, void *d
 void stasis_app_unregister(const char *app_name)
 {
 	struct stasis_app *app;
+	struct stasis_app_event_source *source;
+	int res;
 
 	if (!app_name) {
 		return;
@@ -1848,6 +1866,22 @@ void stasis_app_unregister(const char *app_name)
 		return;
 	}
 
+	/* Unsubscribe from all event sources. */
+	AST_RWLIST_RDLOCK(&event_sources);
+	AST_LIST_TRAVERSE(&event_sources, source, next) {
+		if (!source->unsubscribe || !source->is_subscribed
+			|| !source->is_subscribed(app, NULL)) {
+			continue;
+		}
+
+		res = source->unsubscribe(app, NULL);
+		if (res) {
+			ast_log(LOG_WARNING, "%s: Error unsubscribing from event source '%s'\n",
+				app_name, source->scheme);
+		}
+	}
+	AST_RWLIST_UNLOCK(&event_sources);
+
 	app_deactivate(app);
 
 	/* There's a decent chance that app is ready for cleanup. Go ahead

res/res_stasis_device_state.c

@@ -351,7 +351,7 @@ static int is_subscribed_device_state_lock(struct stasis_app *app, const char *n
 	int is_subscribed;
 
 	ao2_lock(device_state_subscriptions);
-	is_subscribed = is_subscribed_device_state(app, name);
+	is_subscribed = is_subscribed_device_state(app, S_OR(name, DEVICE_STATE_ALL));
 	ao2_unlock(device_state_subscriptions);
 
 	return is_subscribed;
@@ -409,7 +409,7 @@ static int unsubscribe_device_state(struct stasis_app *app, const char *name)
 	struct device_state_subscription *sub;
 
 	ao2_lock(device_state_subscriptions);
-	sub = find_device_state_subscription(app, name);
+	sub = find_device_state_subscription(app, S_OR(name, DEVICE_STATE_ALL));
 	if (sub) {
 		remove_device_state_subscription(sub);
 	}

res/stasis/control.c

@@ -102,6 +102,11 @@ struct stasis_app_control {
 	 * When set, /c app_stasis should exit and continue in the dialplan.
 	 */
 	unsigned int is_done:1;
+	/*!
+	 * When set, /c app_stasis should exit indicating failure and continue
+	 * in the dialplan.
+	 */
+	unsigned int failed:1;
 };
 
 static void control_dtor(void *obj)
@@ -368,6 +373,17 @@ void control_mark_done(struct stasis_app_control *control)
 	ao2_unlock(control->command_queue);
 }
 
+void stasis_app_control_mark_failed(struct stasis_app_control *control)
+{
+	control->failed = 1;
+}
+
+int stasis_app_control_is_failed(const struct stasis_app_control *control)
+{
+	return control->failed;
+}
+
+
 struct stasis_app_control_continue_data {
 	char context[AST_MAX_CONTEXT];
 	char extension[AST_MAX_EXTENSION];