websocket: Consider pending SSL data when waiting for socket input

When TLS is in use, checking the readiness of the underlying FD is insufficient
for determining if there is data available to be read. So before polling the
FD, check if there is any buffered data in the TLS layer and use that first.

ASTERISK-28562 #close
Reported by: Robert Sutton

Change-Id: I95fcb3e2004700d5cf8e5ee04943f0115b15e10d

include/asterisk/http_websocket.h

@@ -337,6 +337,20 @@ AST_OPTIONAL_API(void, ast_websocket_unref, (struct ast_websocket *session), {re
  */
 AST_OPTIONAL_API(int, ast_websocket_fd, (struct ast_websocket *session), { errno = ENOSYS; return -1;});
 
+/*!
+ * \brief Wait for the WebSocket session to be ready to be read.
+ * \since 16.8.0
+ * \since 17.2.0
+ *
+ * \param session Pointer to the WebSocket session
+ * \param timeout the number of milliseconds to wait
+ *
+ * \retval -1 if error occurred
+ * \retval 0 if the timeout expired
+ * \retval 1 if the WebSocket session is ready for reading
+ */
+AST_OPTIONAL_API(int, ast_websocket_wait_for_input, (struct ast_websocket *session, int timeout), { errno = ENOSYS; return -1; });
+
 /*!
  * \brief Get the remote address for a WebSocket connected session.
  *

include/asterisk/iostream.h

@@ -126,6 +126,20 @@ void ast_iostream_set_exclusive_input(struct ast_iostream *stream, int exclusive
  */
 int ast_iostream_get_fd(struct ast_iostream *stream);
 
+/*!
+ * \brief Wait for input on the iostream's file descriptor
+ * \since 16.8.0
+ * \since 17.2.0
+ *
+ * \param stream A pointer to an iostream
+ * \param timeout the number of milliseconds to wait
+ *
+ * \retval -1 if error occurred
+ * \retval 0 if the timeout expired
+ * \retval 1 if the stream is ready for reading
+ */
+int ast_iostream_wait_for_input(struct ast_iostream *stream, int timeout);
+
 /*!
  * \brief Make an iostream non-blocking.
  *

main/iostream.c

@@ -86,6 +86,20 @@ int ast_iostream_get_fd(struct ast_iostream *stream)
 	return stream->fd;
 }
 
+int ast_iostream_wait_for_input(struct ast_iostream *stream, int timeout)
+{
+#if defined(DO_SSL)
+	/* Because SSL is read in blocks, it's possible that the last time we read we
+	   got more than we asked for and it is now buffered inside OpenSSL. If that
+	   is the case, calling ast_wait_for_input() will block until the fd is ready
+	   for reading again, which might never happen. */
+	if (stream->ssl && SSL_pending(stream->ssl)) {
+		return 1;
+	}
+#endif
+	return ast_wait_for_input(stream->fd, timeout);
+}
+
 void ast_iostream_nonblock(struct ast_iostream *stream)
 {
 	ast_fd_set_flags(stream->fd, O_NONBLOCK);

res/res_http_websocket.c

@@ -427,6 +427,11 @@ int AST_OPTIONAL_API_NAME(ast_websocket_fd)(struct ast_websocket *session)
 	return session->closing ? -1 : ast_iostream_get_fd(session->stream);
 }
 
+int AST_OPTIONAL_API_NAME(ast_websocket_wait_for_input)(struct ast_websocket *session, int timeout)
+{
+	return session->closing ? -1 : ast_iostream_wait_for_input(session->stream, timeout);
+}
+
 struct ast_sockaddr * AST_OPTIONAL_API_NAME(ast_websocket_remote_address)(struct ast_websocket *session)
 {
 	return &session->remote_address;
@@ -545,8 +550,8 @@ static inline int ws_safe_read(struct ast_websocket *session, char *buf, size_t
 				break;
 			}
 		}
-		if (ast_wait_for_input(ast_iostream_get_fd(session->stream), 1000) < 0) {
-			ast_log(LOG_ERROR, "ast_wait_for_input returned err: %s\n", strerror(errno));
+		if (ast_iostream_wait_for_input(session->stream, 1000) < 0) {
+			ast_log(LOG_ERROR, "ast_iostream_wait_for_input returned err: %s\n", strerror(errno));
 			*opcode = AST_WEBSOCKET_OPCODE_CLOSE;
 			session->closing = 1;
 			ao2_unlock(session);
@@ -974,7 +979,7 @@ static void websocket_echo_callback(struct ast_websocket *session, struct ast_va
 		goto end;
 	}
 
-	while ((res = ast_wait_for_input(ast_websocket_fd(session), -1)) > 0) {
+	while ((res = ast_websocket_wait_for_input(session, -1)) > 0) {
 		char *payload;
 		uint64_t payload_len;
 		enum ast_websocket_opcode opcode;

res/res_pjsip_transport_websocket.c

@@ -392,7 +392,7 @@ static void websocket_cb(struct ast_websocket *session, struct ast_variable *par
 	transport = create_data.transport;
 	read_data.transport = transport;
 
-	while (ast_wait_for_input(ast_websocket_fd(session), -1) > 0) {
+	while (ast_websocket_wait_for_input(session, -1) > 0) {
 		enum ast_websocket_opcode opcode;
 		int fragmented;