Implement internal abstraction for iostreams

fopencookie/funclose is a non-standard API and should not be used
in portable software. Additionally, the way FILE's fd is used in
non-blocking mode is undefined behaviour and cannot be relied on.

This introduces internal abstraction for io streams, that allows
implementing the desired virtualization of read/write operations
with necessary timeout handling.

ASTERISK-24515 #close
ASTERISK-24517 #close

Change-Id: Id916aef418b665ced6a7489aef74908b6e376e85

5 files changed, 739 insertions, 834 deletions

diff --git a/main/http.c b/main/http.c
index 77feb397b..9aff4d167 100644
--- a/main/http.c
+++ b/main/http.c
@@ -449,11 +449,13 @@ void ast_http_send(struct ast_tcptls_session_instance *ser,
 	struct timeval now = ast_tvnow();
 	struct ast_tm tm;
 	char timebuf[80];
+	char buf[256];
+	int len;
 	int content_length = 0;
 	int close_connection;
 	struct ast_str *server_header_field = ast_str_create(MAX_SERVER_NAME_LENGTH);
 
-	if (!ser || !ser->f || !server_header_field) {
+	if (!ser || !server_header_field) {
 		/* The connection is not open. */
 		ast_free(http_header);
 		ast_free(out);
@@ -503,7 +505,7 @@ void ast_http_send(struct ast_tcptls_session_instance *ser,
 	}
 
 	/* send http header */
-	fprintf(ser->f,
+	ast_iostream_printf(ser->stream,
 		"HTTP/1.1 %d %s\r\n"
 		"%s"
 		"Date: %s\r\n"
@@ -524,18 +526,16 @@ void ast_http_send(struct ast_tcptls_session_instance *ser,
 	/* send content */
 	if (method != AST_HTTP_HEAD || status_code >= 400) {
 		if (out && ast_str_strlen(out)) {
-			if (fwrite(ast_str_buffer(out), ast_str_strlen(out), 1, ser->f) != 1) {
+			len = ast_str_strlen(out);
+			if (ast_iostream_write(ser->stream, ast_str_buffer(out), len) != len) {
 				ast_log(LOG_ERROR, "fwrite() failed: %s\n", strerror(errno));
 				close_connection = 1;
 			}
 		}
 
 		if (fd) {
-			char buf[256];
-			int len;
-
 			while ((len = read(fd, buf, sizeof(buf))) > 0) {
-				if (fwrite(buf, len, 1, ser->f) != 1) {
+				if (ast_iostream_write(ser->stream, buf, len) != len) {
 					ast_log(LOG_WARNING, "fwrite() failed: %s\n", strerror(errno));
 					close_connection = 1;
 					break;
@@ -567,7 +567,7 @@ void ast_http_create_response(struct ast_tcptls_session_instance *ser, int statu
 		ast_free(http_header_data);
 		ast_free(server_address);
 		ast_free(out);
-		if (ser && ser->f) {
+		if (ser) {
 			ast_debug(1, "HTTP closing session. OOM.\n");
 			ast_tcptls_close_session_file(ser);
 		}
@@ -921,9 +921,9 @@ static int http_body_read_contents(struct ast_tcptls_session_instance *ser, char
 {
 	int res;
 
-	/* Stay in fread until get all the expected data or timeout. */
-	res = fread(buf, length, 1, ser->f);
-	if (res < 1) {
+	/* Stream is in exclusive mode so we get it all if possible. */
+	res = ast_iostream_read(ser->stream, buf, length);
+	if (res < length) {
 		ast_log(LOG_WARNING, "Short HTTP request %s (Wanted %d)\n",
 			what_getting, length);
 		return -1;
@@ -945,23 +945,12 @@ static int http_body_read_contents(struct ast_tcptls_session_instance *ser, char
  */
 static int http_body_discard_contents(struct ast_tcptls_session_instance *ser, int length, const char *what_getting)
 {
-	int res;
-	char buf[MAX_HTTP_LINE_LENGTH];/* Discard buffer */
-
-	/* Stay in fread until get all the expected data or timeout. */
-	while (sizeof(buf) < length) {
-		res = fread(buf, sizeof(buf), 1, ser->f);
-		if (res < 1) {
-			ast_log(LOG_WARNING, "Short HTTP request %s (Wanted %zu of remaining %d)\n",
-				what_getting, sizeof(buf), length);
-			return -1;
-		}
-		length -= sizeof(buf);
-	}
-	res = fread(buf, length, 1, ser->f);
-	if (res < 1) {
-		ast_log(LOG_WARNING, "Short HTTP request %s (Wanted %d of remaining %d)\n",
-			what_getting, length, length);
+	ssize_t res;
+
+	res = ast_iostream_discard(ser->stream, length);
+	if (res < length) {
+		ast_log(LOG_WARNING, "Short HTTP request %s (Wanted %d but got %zd)\n",
+			what_getting, length, res);
 		return -1;
 	}
 	return 0;
@@ -1037,7 +1026,7 @@ static int http_body_get_chunk_length(struct ast_tcptls_session_instance *ser)
 	char header_line[MAX_HTTP_LINE_LENGTH];
 
 	/* get the line of hexadecimal giving chunk-size w/ optional chunk-extension */
-	if (!fgets(header_line, sizeof(header_line), ser->f)) {
+	if (ast_iostream_gets(ser->stream, header_line, sizeof(header_line)) <= 0) {
 		ast_log(LOG_WARNING, "Short HTTP read of chunked header\n");
 		return -1;
 	}
@@ -1065,8 +1054,8 @@ static int http_body_check_chunk_sync(struct ast_tcptls_session_instance *ser)
 	char chunk_sync[2];
 
 	/* Stay in fread until get the expected CRLF or timeout. */
-	res = fread(chunk_sync, sizeof(chunk_sync), 1, ser->f);
-	if (res < 1) {
+	res = ast_iostream_read(ser->stream, chunk_sync, sizeof(chunk_sync));
+	if (res < sizeof(chunk_sync)) {
 		ast_log(LOG_WARNING, "Short HTTP chunk sync read (Wanted %zu)\n",
 			sizeof(chunk_sync));
 		return -1;
@@ -1095,7 +1084,7 @@ static int http_body_discard_chunk_trailer_headers(struct ast_tcptls_session_ins
 	char header_line[MAX_HTTP_LINE_LENGTH];
 
 	for (;;) {
-		if (!fgets(header_line, sizeof(header_line), ser->f)) {
+		if (ast_iostream_gets(ser->stream, header_line, sizeof(header_line)) <= 0) {
 			ast_log(LOG_WARNING, "Short HTTP read of chunked trailer header\n");
 			return -1;
 		}
@@ -1758,7 +1747,7 @@ static int http_request_headers_get(struct ast_tcptls_session_instance *ser, str
 		char *name;
 		char *value;
 
-		if (!fgets(header_line, sizeof(header_line), ser->f)) {
+		if (ast_iostream_gets(ser->stream, header_line, sizeof(header_line)) <= 0) {
 			ast_http_error(ser, 400, "Bad Request", "Timeout");
 			return -1;
 		}
@@ -1832,7 +1821,7 @@ static int httpd_process_request(struct ast_tcptls_session_instance *ser)
 	int res;
 	char request_line[MAX_HTTP_LINE_LENGTH];
 
-	if (!fgets(request_line, sizeof(request_line), ser->f)) {
+	if (ast_iostream_gets(ser->stream, request_line, sizeof(request_line)) <= 0) {
 		return -1;
 	}
 
@@ -1913,11 +1902,10 @@ static int httpd_process_request(struct ast_tcptls_session_instance *ser)
 static void *httpd_helper_thread(void *data)
 {
 	struct ast_tcptls_session_instance *ser = data;
-	struct protoent *p;
-	int flags;
 	int timeout;
+	int arg = 1;
 
-	if (!ser || !ser->f) {
+	if (!ser) {
 		ao2_cleanup(ser);
 		return NULL;
 	}
@@ -1934,23 +1922,11 @@ static void *httpd_helper_thread(void *data)
 	 * This is necessary to prevent delays (caused by buffering) as we
 	 * write to the socket in bits and pieces.
 	 */
-	p = getprotobyname("tcp");
-	if (p) {
-		int arg = 1;
-
-		if (setsockopt(ser->fd, p->p_proto, TCP_NODELAY, (char *) &arg, sizeof(arg) ) < 0) {
-			ast_log(LOG_WARNING, "Failed to set TCP_NODELAY on HTTP connection: %s\n", strerror(errno));
-			ast_log(LOG_WARNING, "Some HTTP requests may be slow to respond.\n");
-		}
-	} else {
-		ast_log(LOG_WARNING, "Failed to set TCP_NODELAY on HTTP connection, getprotobyname(\"tcp\") failed\n");
+	if (setsockopt(ast_iostream_get_fd(ser->stream), IPPROTO_TCP, TCP_NODELAY, (char *) &arg, sizeof(arg) ) < 0) {
+		ast_log(LOG_WARNING, "Failed to set TCP_NODELAY on HTTP connection: %s\n", strerror(errno));
 		ast_log(LOG_WARNING, "Some HTTP requests may be slow to respond.\n");
 	}
-
-	/* make sure socket is non-blocking */
-	flags = fcntl(ser->fd, F_GETFL);
-	flags |= O_NONBLOCK;
-	fcntl(ser->fd, F_SETFL, flags);
+	ast_iostream_nonblock(ser->stream);
 
 	/* Setup HTTP worker private data to keep track of request body reading. */
 	ao2_cleanup(ser->private_data);
@@ -1973,23 +1949,17 @@ static void *httpd_helper_thread(void *data)
 	}
 
 	/* We can let the stream wait for data to arrive. */
-	ast_tcptls_stream_set_exclusive_input(ser->stream_cookie, 1);
+	ast_iostream_set_exclusive_input(ser->stream, 1);
 
 	for (;;) {
-		int ch;
-
 		/* Wait for next potential HTTP request message. */
-		ast_tcptls_stream_set_timeout_inactivity(ser->stream_cookie, timeout);
-		ch = fgetc(ser->f);
-		if (ch == EOF || ungetc(ch, ser->f) == EOF) {
-			/* Between request idle timeout */
-			ast_debug(1, "HTTP idle timeout or peer closed connection.\n");
+		ast_iostream_set_timeout_idle_inactivity(ser->stream, timeout, session_inactivity);
+		if (httpd_process_request(ser)) {
+			/* Break the connection or the connection closed */
 			break;
 		}
-
-		ast_tcptls_stream_set_timeout_inactivity(ser->stream_cookie, session_inactivity);
-		if (httpd_process_request(ser) || !ser->f || feof(ser->f)) {
-			/* Break the connection or the connection closed */
+		if (!ser->stream) {
+			/* Web-socket or similar that took the connection */
 			break;
 		}
 
@@ -2003,10 +1973,9 @@ static void *httpd_helper_thread(void *data)
 done:
 	ast_atomic_fetchadd_int(&session_count, -1);
 
-	if (ser->f) {
-		ast_debug(1, "HTTP closing session.  Top level\n");
-		ast_tcptls_close_session_file(ser);
-	}
+	ast_debug(1, "HTTP closing session.  Top level\n");
+	ast_tcptls_close_session_file(ser);
+
 	ao2_ref(ser, -1);
 	return NULL;
 }
diff --git a/main/iostream.c b/main/iostream.c
new file mode 100644
index 000000000..46abc18a5
--- /dev/null
+++ b/main/iostream.c
@@ -0,0 +1,553 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 1999 - 2015, Digium, Inc.
+ *
+ * Timo Teräs <timo.teras@iki.fi>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+#include <fcntl.h>
+#include <stdarg.h>
+
+#include "asterisk.h"
+#include "asterisk/utils.h"
+#include "asterisk/astobj2.h"
+#include "asterisk/iostream.h"
+
+struct ast_iostream {
+	SSL *ssl;
+	struct timeval start;
+	int fd;
+	int timeout;
+	int timeout_reset;
+	int exclusive_input;
+	int rbuflen;
+	char *rbufhead;
+	char rbuf[2048];
+};
+
+int ast_iostream_get_fd(struct ast_iostream *stream)
+{
+	return stream->fd;
+}
+
+void ast_iostream_nonblock(struct ast_iostream *stream)
+{
+	fcntl(stream->fd, F_SETFL, fcntl(stream->fd, F_GETFL) | O_NONBLOCK);
+}
+
+SSL *ast_iostream_get_ssl(struct ast_iostream *stream)
+{
+	return stream->ssl;
+}
+
+void ast_iostream_set_timeout_disable(struct ast_iostream *stream)
+{
+	ast_assert(stream != NULL);
+
+	stream->timeout = -1;
+	stream->timeout_reset = -1;
+}
+
+void ast_iostream_set_timeout_inactivity(struct ast_iostream *stream, int timeout)
+{
+	ast_assert(stream != NULL);
+
+	stream->start.tv_sec = 0;
+	stream->timeout = timeout;
+	stream->timeout_reset = timeout;
+}
+
+void ast_iostream_set_timeout_idle_inactivity(struct ast_iostream *stream, int timeout, int timeout_reset)
+{
+	ast_assert(stream != NULL);
+
+	stream->start.tv_sec = 0;
+	stream->timeout = timeout;
+	stream->timeout_reset = timeout_reset;
+}
+
+void ast_iostream_set_timeout_sequence(struct ast_iostream *stream, struct timeval start, int timeout)
+{
+	ast_assert(stream != NULL);
+
+	stream->start = start;
+	stream->timeout = timeout;
+	stream->timeout_reset = timeout;
+}
+
+void ast_iostream_set_exclusive_input(struct ast_iostream *stream, int exclusive_input)
+{
+	ast_assert(stream != NULL);
+
+	stream->exclusive_input = exclusive_input;
+}
+
+static ssize_t iostream_read(struct ast_iostream *stream, void *buf, size_t size)
+{
+	struct timeval start;
+	int ms;
+	int res;
+
+	if (stream->start.tv_sec) {
+		start = stream->start;
+	} else {
+		start = ast_tvnow();
+	}
+
+#if defined(DO_SSL)
+	if (stream->ssl) {
+		for (;;) {
+			res = SSL_read(stream->ssl, buf, size);
+			if (0 < res) {
+				/* We read some payload data. */
+				stream->timeout = stream->timeout_reset;
+				return res;
+			}
+			switch (SSL_get_error(stream->ssl, res)) {
+			case SSL_ERROR_ZERO_RETURN:
+				/* Report EOF for a shutdown */
+				ast_debug(1, "TLS clean shutdown alert reading data\n");
+				return 0;
+			case SSL_ERROR_WANT_READ:
+				if (!stream->exclusive_input) {
+					/* We cannot wait for data now. */
+					errno = EAGAIN;
+					return -1;
+				}
+				while ((ms = ast_remaining_ms(start, stream->timeout))) {
+					res = ast_wait_for_input(stream->fd, ms);
+					if (0 < res) {
+						/* Socket is ready to be read. */
+						break;
+					}
+					if (res < 0) {
+						if (errno == EINTR || errno == EAGAIN) {
+							/* Try again. */
+							continue;
+						}
+						ast_debug(1, "TLS socket error waiting for read data: %s\n",
+							strerror(errno));
+						return -1;
+					}
+				}
+				break;
+			case SSL_ERROR_WANT_WRITE:
+				while ((ms = ast_remaining_ms(start, stream->timeout))) {
+					res = ast_wait_for_output(stream->fd, ms);
+					if (0 < res) {
+						/* Socket is ready to be written. */
+						break;
+					}
+					if (res < 0) {
+						if (errno == EINTR || errno == EAGAIN) {
+							/* Try again. */
+							continue;
+						}
+						ast_debug(1, "TLS socket error waiting for write space: %s\n",
+							strerror(errno));
+						return -1;
+					}
+				}
+				break;
+			default:
+				/* Report EOF for an undecoded SSL or transport error. */
+				ast_debug(1, "TLS transport or SSL error reading data\n");
+				return 0;
+			}
+			if (!ms) {
+				/* Report EOF for a timeout */
+				ast_debug(1, "TLS timeout reading data\n");
+				return 0;
+			}
+		}
+	}
+#endif	/* defined(DO_SSL) */
+
+	for (;;) {
+		res = read(stream->fd, buf, size);
+		if (0 <= res) {
+			/* Got data or we cannot wait for it. */
+			stream->timeout = stream->timeout_reset;
+			return res;
+		}
+		if (!stream->exclusive_input) {
+			return res;
+		}
+		if (errno != EINTR && errno != EAGAIN) {
+			/* Not a retryable error. */
+			ast_debug(1, "TCP socket error reading data: %s\n",
+				strerror(errno));
+			return -1;
+		}
+		ms = ast_remaining_ms(start, stream->timeout);
+		if (!ms) {
+			/* Report EOF for a timeout */
+			ast_debug(1, "TCP timeout reading data\n");
+			return 0;
+		}
+		ast_wait_for_input(stream->fd, ms);
+	}
+}
+
+ssize_t ast_iostream_read(struct ast_iostream *stream, void *buf, size_t size)
+{
+	if (!size) {
+		/* You asked for no data you got no data. */
+		return 0;
+	}
+
+	if (!stream || stream->fd == -1) {
+		errno = EBADF;
+		return -1;
+	}
+
+	/* Get any remains from the read buffer */
+	if (stream->rbuflen) {
+		size_t r = size;
+		if (r > stream->rbuflen) {
+			r = stream->rbuflen;
+		}
+		memcpy(buf, stream->rbufhead, r);
+		stream->rbuflen -= r;
+		stream->rbufhead += r;
+		return r;
+	}
+
+	return iostream_read(stream, buf, size);
+}
+
+ssize_t ast_iostream_gets(struct ast_iostream *stream, char *buf, size_t count)
+{
+	ssize_t r;
+	char *newline;
+
+	do {
+		/* Search for newline */
+		newline = memchr(stream->rbufhead, '\n', stream->rbuflen);
+		if (newline) {
+			r = newline - stream->rbufhead + 1;
+			if (r > count-1) {
+				r = count-1;
+			}
+			break;
+		}
+
+		/* Enough data? */
+		if (stream->rbuflen >= count - 1) {
+			r = count - 1;
+			break;
+		}
+
+		/* Try to fill in line buffer */
+		if (stream->rbuflen && stream->rbuf != stream->rbufhead) {
+			memmove(&stream->rbuf, stream->rbufhead, stream->rbuflen);
+		}
+		stream->rbufhead = stream->rbuf;
+
+		r = iostream_read(stream, stream->rbufhead + stream->rbuflen, sizeof(stream->rbuf) - stream->rbuflen);
+		if (r <= 0) {
+			return r;
+		}
+		stream->rbuflen += r;
+	} while (1);
+
+	/* Return r bytes with termination byte */
+	memcpy(buf, stream->rbufhead, r);
+	buf[r] = 0;
+	stream->rbuflen -= r;
+	stream->rbufhead += r;
+
+	return r;
+}
+
+ssize_t ast_iostream_discard(struct ast_iostream *stream, size_t size)
+{
+	char buf[1024];
+	size_t remaining = size;
+	ssize_t ret;
+
+	while (remaining) {
+		ret = ast_iostream_read(stream, buf, remaining > sizeof(buf) ? sizeof(buf) : remaining);
+		if (ret < 0) {
+			return ret;
+		}
+		remaining -= ret;
+	}
+
+	return size;
+}
+
+ssize_t ast_iostream_write(struct ast_iostream *stream, const void *buf, size_t size)
+{
+	struct timeval start;
+	int ms;
+	int res;
+	int written;
+	int remaining;
+
+	if (!size) {
+		/* You asked to write no data you wrote no data. */
+		return 0;
+	}
+
+	if (!stream || stream->fd == -1) {
+		errno = EBADF;
+		return -1;
+	}
+
+	if (stream->start.tv_sec) {
+		start = stream->start;
+	} else {
+		start = ast_tvnow();
+	}
+
+#if defined(DO_SSL)
+	if (stream->ssl) {
+		written = 0;
+		remaining = size;
+		for (;;) {
+			res = SSL_write(stream->ssl, buf + written, remaining);
+			if (res == remaining) {
+				/* Everything was written. */
+				return size;
+			}
+			if (0 < res) {
+				/* Successfully wrote part of the buffer.  Try to write the rest. */
+				written += res;
+				remaining -= res;
+				continue;
+			}
+			switch (SSL_get_error(stream->ssl, res)) {
+			case SSL_ERROR_ZERO_RETURN:
+				ast_debug(1, "TLS clean shutdown alert writing data\n");
+				if (written) {
+					/* Report partial write. */
+					return written;
+				}
+				errno = EBADF;
+				return -1;
+			case SSL_ERROR_WANT_READ:
+				ms = ast_remaining_ms(start, stream->timeout);
+				if (!ms) {
+					/* Report partial write. */
+					ast_debug(1, "TLS timeout writing data (want read)\n");
+					return written;
+				}
+				ast_wait_for_input(stream->fd, ms);
+				break;
+			case SSL_ERROR_WANT_WRITE:
+				ms = ast_remaining_ms(start, stream->timeout);
+				if (!ms) {
+					/* Report partial write. */
+					ast_debug(1, "TLS timeout writing data (want write)\n");
+					return written;
+				}
+				ast_wait_for_output(stream->fd, ms);
+				break;
+			default:
+				/* Undecoded SSL or transport error. */
+				ast_debug(1, "TLS transport or SSL error writing data\n");
+				if (written) {
+					/* Report partial write. */
+					return written;
+				}
+				errno = EBADF;
+				return -1;
+			}
+		}
+	}
+#endif	/* defined(DO_SSL) */
+
+	written = 0;
+	remaining = size;
+	for (;;) {
+		res = write(stream->fd, buf + written, remaining);
+		if (res == remaining) {
+			/* Yay everything was written. */
+			return size;
+		}
+		if (0 < res) {
+			/* Successfully wrote part of the buffer.  Try to write the rest. */
+			written += res;
+			remaining -= res;
+			continue;
+		}
+		if (errno != EINTR && errno != EAGAIN) {
+			/* Not a retryable error. */
+			ast_debug(1, "TCP socket error writing: %s\n", strerror(errno));
+			if (written) {
+				return written;
+			}
+			return -1;
+		}
+		ms = ast_remaining_ms(start, stream->timeout);
+		if (!ms) {
+			/* Report partial write. */
+			ast_debug(1, "TCP timeout writing data\n");
+			return written;
+		}
+		ast_wait_for_output(stream->fd, ms);
+	}
+}
+
+ssize_t ast_iostream_printf(struct ast_iostream *stream, const void *fmt, ...)
+{
+	char sbuf[256], *buf = sbuf;
+	int len, len2, ret = -1;
+	va_list va;
+
+	va_start(va, fmt);
+	len = vsnprintf(buf, sizeof(sbuf), fmt, va);
+	va_end(va);
+
+	if (len > sizeof(sbuf)) {
+		buf = ast_malloc(len);
+		if (!buf) {
+			return -1;
+		}
+		va_start(va, fmt);
+		len2 = vsnprintf(buf, len, fmt, va);
+		va_end(va);
+		if (len2 > len) {
+			goto error;
+		}
+	}
+
+	if (ast_iostream_write(stream, buf, len) == len)
+		ret = len;
+
+error:
+	if (buf != sbuf) {
+		ast_free(buf);
+	}
+
+	return ret;
+}
+
+int ast_iostream_close(struct ast_iostream *stream)
+{
+	if (!stream) {
+		errno = EBADF;
+		return -1;
+	}
+
+	if (stream->fd != -1) {
+#if defined(DO_SSL)
+		if (stream->ssl) {
+			int res;
+
+			/*
+			 * According to the TLS standard, it is acceptable for an
+			 * application to only send its shutdown alert and then
+			 * close the underlying connection without waiting for
+			 * the peer's response (this way resources can be saved,
+			 * as the process can already terminate or serve another
+			 * connection).
+			 */
+			res = SSL_shutdown(stream->ssl);
+			if (res < 0) {
+				ast_log(LOG_ERROR, "SSL_shutdown() failed: %d\n",
+					SSL_get_error(stream->ssl, res));
+			}
+
+			if (!stream->ssl->server) {
+				/* For client threads, ensure that the error stack is cleared */
+				ERR_remove_state(0);
+			}
+
+			SSL_free(stream->ssl);
+			stream->ssl = NULL;
+		}
+#endif	/* defined(DO_SSL) */
+
+		/*
+		 * Issuing shutdown() is necessary here to avoid a race
+		 * condition where the last data written may not appear
+		 * in the TCP stream.  See ASTERISK-23548
+		 */
+		shutdown(stream->fd, SHUT_RDWR);
+		if (close(stream->fd)) {
+			ast_log(LOG_ERROR, "close() failed: %s\n", strerror(errno));
+		}
+		stream->fd = -1;
+	}
+	ao2_t_ref(stream, -1, "Closed ast_iostream");
+
+	return 0;
+}
+
+static void iostream_dtor(void *cookie)
+{
+#ifdef AST_DEVMODE
+	/* Since the ast_assert below is the only one using stream,
+	 * and ast_assert is only available with AST_DEVMODE, we
+	 * put this in a conditional to avoid compiler warnings. */
+	struct ast_iostream *stream = cookie;
+#endif
+
+	ast_assert(stream->fd == -1);
+}
+
+struct ast_iostream *ast_iostream_from_fd(int *fd)
+{
+	struct ast_iostream *stream;
+
+	stream = ao2_alloc_options(sizeof(*stream), iostream_dtor,
+		AO2_ALLOC_OPT_LOCK_NOLOCK);
+	if (stream) {
+		stream->timeout = -1;
+		stream->timeout_reset = -1;
+		stream->fd = *fd;
+		*fd = -1;
+	}
+
+	return stream;
+}
+
+int ast_iostream_start_tls(struct ast_iostream **pstream, SSL_CTX *ssl_ctx, int client)
+{
+#ifdef DO_SSL
+	struct ast_iostream *stream = *pstream;
+	int (*ssl_setup)(SSL *) = client ? SSL_connect : SSL_accept;
+	char err[256];
+
+	stream->ssl = SSL_new(ssl_ctx);
+	if (!stream->ssl) {
+		ast_log(LOG_ERROR, "Unable to create new SSL connection\n");
+		errno = ENOMEM;
+		return -1;
+	}
+
+	/*
+	 * This function takes struct ast_iostream **, so it can chain
+	 * SSL over any ast_iostream. For now we assume it's a file descriptor.
+	 * But later this should instead use BIO wrapper to tie SSL to another
+	 * ast_iostream.
+	 */
+	SSL_set_fd(stream->ssl, stream->fd);
+
+	if (ssl_setup(stream->ssl) <= 0) {
+		ast_log(LOG_ERROR, "Problem setting up ssl connection: %s\n",
+			ERR_error_string(ERR_get_error(), err));
+		errno = EIO;
+		return -1;
+	}
+
+	return 0;
+#else
+	ast_log(LOG_ERROR, "SSL not enabled in this build\n");
+	errno = ENOTSUP;
+	return -1;
+#endif
+}
diff --git a/main/manager.c b/main/manager.c
index ef1afb03d..f059015c7 100644
--- a/main/manager.c
+++ b/main/manager.c
@@ -1549,8 +1549,7 @@ static void acl_change_stasis_unsubscribe(void)
 struct mansession_session {
 				/*! \todo XXX need to document which fields it is protecting */
 	struct ast_sockaddr addr;	/*!< address we are connecting from */
-	FILE *f;		/*!< fdopen() on the underlying fd */
-	int fd;			/*!< descriptor used for output. Either the socket (AMI) or a temporary file (HTTP) */
+	struct ast_iostream *stream;	/*!< AMI stream */
 	int inuse;		/*!< number of HTTP sessions using this entry */
 	int needdestroy;	/*!< Whether an HTTP session should be destroyed */
 	pthread_t waiting_thread;	/*!< Sleeping thread using this descriptor */
@@ -1592,9 +1591,8 @@ enum mansession_message_parsing {
  */
 struct mansession {
 	struct mansession_session *session;
+	struct ast_iostream *stream;
 	struct ast_tcptls_session_instance *tcptls_session;
-	FILE *f;
-	int fd;
 	enum mansession_message_parsing parsing;
 	int write_error:1;
 	struct manager_custom_hook *hook;
@@ -2166,10 +2164,6 @@ static void session_destructor(void *obj)
 		ast_datastore_free(datastore);
 	}
 
-	if (session->f != NULL) {
-		fflush(session->f);
-		fclose(session->f);
-	}
 	if (eqe) {
 		ast_atomic_fetchadd_int(&eqe->usecount, -1);
 	}
@@ -2204,7 +2198,6 @@ static struct mansession_session *build_mansession(const struct ast_sockaddr *ad
 		return NULL;
 	}
 
-	newsession->fd = -1;
 	newsession->waiting_thread = AST_PTHREADT_NULL;
 	newsession->writetimeout = 100;
 	newsession->send_events = -1;
@@ -2617,7 +2610,7 @@ static char *handle_showmanconn(struct ast_cli_entry *e, int cmd, struct ast_cli
 				ast_sockaddr_stringify_addr(&session->addr),
 				(int) (session->sessionstart),
 				(int) (now - session->sessionstart),
-				session->fd,
+				session->stream ? ast_iostream_get_fd(session->stream) : -1,
 				session->inuse,
 				session->readperm,
 				session->writeperm);
@@ -2889,7 +2882,6 @@ int ast_hook_send_action(struct manager_custom_hook *hook, const char *msg)
 			 * This is necessary to meet the previous design of manager.c
 			 */
 			s.hook = hook;
-			s.f = (void*)1; /* set this to something so our request will make it through all functions that test it*/
 
 			ao2_lock(act_found);
 			if (act_found->registered && act_found->func) {
@@ -2920,9 +2912,8 @@ int ast_hook_send_action(struct manager_custom_hook *hook, const char *msg)
  */
 static int send_string(struct mansession *s, char *string)
 {
-	int res;
-	FILE *f = s->f ? s->f : s->session->f;
-	int fd = s->f ? s->fd : s->session->fd;
+	struct ast_iostream *stream = s->stream ? s->stream : s->session->stream;
+	int len, res;
 
 	/* It's a result from one of the hook's action invocation */
 	if (s->hook) {
@@ -2934,7 +2925,12 @@ static int send_string(struct mansession *s, char *string)
 		return 0;
 	}
 
-	if ((res = ast_careful_fwrite(f, fd, string, strlen(string), s->session->writetimeout))) {
+	len = strlen(string);
+	ast_iostream_set_timeout_inactivity(stream, s->session->writetimeout);
+	res = ast_iostream_write(stream, string, len);
+	ast_iostream_set_timeout_disable(stream);
+
+	if (res < len) {
 		s->write_error = 1;
 	}
 
@@ -2975,10 +2971,10 @@ void astman_append(struct mansession *s, const char *fmt, ...)
 		return;
 	}
 
-	if (s->f != NULL || s->session->f != NULL) {
+	if (s->tcptls_session != NULL && s->tcptls_session->stream != NULL) {
 		send_string(s, ast_str_buffer(buf));
 	} else {
-		ast_verbose("fd == -1 in astman_append, should not happen\n");
+		ast_verbose("No connection stream in astman_append, should not happen\n");
 	}
 }
 
@@ -4119,7 +4115,7 @@ static int action_waitevent(struct mansession *s, const struct message *m)
 			break;
 		}
 		if (s->session->managerid == 0) {	/* AMI session */
-			if (ast_wait_for_input(s->session->fd, 1000)) {
+			if (ast_wait_for_input(ast_iostream_get_fd(s->session->stream), 1000)) {
 				break;
 			}
 		} else {	/* HTTP session */
@@ -5924,7 +5920,7 @@ static int process_events(struct mansession *s)
 	int ret = 0;
 
 	ao2_lock(s->session);
-	if (s->session->f != NULL) {
+	if (s->session->stream != NULL) {
 		struct eventqent *eqe = s->session->last_ev;
 
 		while ((eqe = advance_event(eqe))) {
@@ -6466,7 +6462,7 @@ static int get_input(struct mansession *s, char *output)
 		s->session->waiting_thread = pthread_self();
 		ao2_unlock(s->session);
 
-		res = ast_wait_for_input(s->session->fd, timeout);
+		res = ast_wait_for_input(ast_iostream_get_fd(s->session->stream), timeout);
 
 		ao2_lock(s->session);
 		s->session->waiting_thread = AST_PTHREADT_NULL;
@@ -6484,7 +6480,7 @@ static int get_input(struct mansession *s, char *output)
 	}
 
 	ao2_lock(s->session);
-	res = fread(src + s->session->inlen, 1, maxlen - s->session->inlen, s->session->f);
+	res = ast_iostream_read(s->session->stream, src + s->session->inlen, maxlen - s->session->inlen);
 	if (res < 1) {
 		res = -1;	/* error return */
 	} else {
@@ -6617,13 +6613,12 @@ static void *session_do(void *data)
 	struct mansession s = {
 		.tcptls_session = data,
 	};
-	int flags;
 	int res;
+	int arg = 1;
 	struct ast_sockaddr ser_remote_address_tmp;
-	struct protoent *p;
 
 	if (ast_atomic_fetchadd_int(&unauth_sessions, +1) >= authlimit) {
-		fclose(ser->f);
+		ast_iostream_close(ser->stream);
 		ast_atomic_fetchadd_int(&unauth_sessions, -1);
 		goto done;
 	}
@@ -6632,7 +6627,7 @@ static void *session_do(void *data)
 	session = build_mansession(&ser_remote_address_tmp);
 
 	if (session == NULL) {
-		fclose(ser->f);
+		ast_iostream_close(ser->stream);
 		ast_atomic_fetchadd_int(&unauth_sessions, -1);
 		goto done;
 	}
@@ -6640,20 +6635,10 @@ static void *session_do(void *data)
 	/* here we set TCP_NODELAY on the socket to disable Nagle's algorithm.
 	 * This is necessary to prevent delays (caused by buffering) as we
 	 * write to the socket in bits and pieces. */
-	p = getprotobyname("tcp");
-	if (p) {
-		int arg = 1;
-		if( setsockopt(ser->fd, p->p_proto, TCP_NODELAY, (char *)&arg, sizeof(arg) ) < 0 ) {
-			ast_log(LOG_WARNING, "Failed to set manager tcp connection to TCP_NODELAY mode: %s\nSome manager actions may be slow to respond.\n", strerror(errno));
-		}
-	} else {
-		ast_log(LOG_WARNING, "Failed to set manager tcp connection to TCP_NODELAY, getprotobyname(\"tcp\") failed\nSome manager actions may be slow to respond.\n");
+	if (setsockopt(ast_iostream_get_fd(ser->stream), IPPROTO_TCP, TCP_NODELAY, (char *)&arg, sizeof(arg) ) < 0) {
+		ast_log(LOG_WARNING, "Failed to set manager tcp connection to TCP_NODELAY mode: %s\nSome manager actions may be slow to respond.\n", strerror(errno));
 	}
-
-	/* make sure socket is non-blocking */
-	flags = fcntl(ser->fd, F_GETFL);
-	flags |= O_NONBLOCK;
-	fcntl(ser->fd, F_SETFL, flags);
+	ast_iostream_nonblock(ser->stream);
 
 	ao2_lock(session);
 	/* Hook to the tail of the event queue */
@@ -6662,8 +6647,7 @@ static void *session_do(void *data)
 	ast_mutex_init(&s.lock);
 
 	/* these fields duplicate those in the 'ser' structure */
-	session->fd = s.fd = ser->fd;
-	session->f = s.f = ser->f;
+	session->stream = s.stream = ser->stream;
 	ast_sockaddr_copy(&session->addr, &ser_remote_address_tmp);
 	s.session = session;
 
@@ -6682,9 +6666,9 @@ static void *session_do(void *data)
 	 * We cannot let the stream exclusively wait for data to arrive.
 	 * We have to wake up the task to send async events.
 	 */
-	ast_tcptls_stream_set_exclusive_input(ser->stream_cookie, 0);
+	ast_iostream_set_exclusive_input(ser->stream, 0);
 
-	ast_tcptls_stream_set_timeout_sequence(ser->stream_cookie,
+	ast_iostream_set_timeout_sequence(ser->stream,
 		ast_tvnow(), authtimeout * 1000);
 
 	astman_append(&s, "Asterisk Call Manager/%s\r\n", AMI_VERSION);	/* welcome prompt */
@@ -6693,7 +6677,7 @@ static void *session_do(void *data)
 			break;
 		}
 		if (session->authenticated) {
-			ast_tcptls_stream_set_timeout_disable(ser->stream_cookie);
+			ast_iostream_set_timeout_disable(ser->stream);
 		}
 	}
 	/* session is over, explain why and terminate */
@@ -7552,23 +7536,9 @@ static void xml_translate(struct ast_str **out, char *in, struct ast_variable *g
 
 static void close_mansession_file(struct mansession *s)
 {
-	if (s->f) {
-		if (fclose(s->f)) {
-			ast_log(LOG_ERROR, "fclose() failed: %s\n", strerror(errno));
-		}
-		s->f = NULL;
-		s->fd = -1;
-	} else if (s->fd != -1) {
-		/*
-		 * Issuing shutdown() is necessary here to avoid a race
-		 * condition where the last data written may not appear
-		 * in the TCP stream.  See ASTERISK-23548
-		 */
-		shutdown(s->fd, SHUT_RDWR);
-		if (close(s->fd)) {
-			ast_log(LOG_ERROR, "close() failed: %s\n", strerror(errno));
-		}
-		s->fd = -1;
+	if (s->stream) {
+		ast_iostream_close(s->stream);
+		s->stream = NULL;
 	} else {
 		ast_log(LOG_ERROR, "Attempted to close file/file descriptor on mansession without a valid file or file descriptor.\n");
 	}
@@ -7577,17 +7547,20 @@ static void close_mansession_file(struct mansession *s)
 static void process_output(struct mansession *s, struct ast_str **out, struct ast_variable *params, enum output_format format)
 {
 	char *buf;
-	size_t l;
+	off_t l;
+	int fd;
 
-	if (!s->f)
+	if (!s->stream)
 		return;
 
 	/* Ensure buffer is NULL-terminated */
-	fprintf(s->f, "%c", 0);
-	fflush(s->f);
+	ast_iostream_write(s->stream, "", 1);
+
+	fd = ast_iostream_get_fd(s->stream);
 
-	if ((l = ftell(s->f)) > 0) {
-		if (MAP_FAILED == (buf = mmap(NULL, l, PROT_READ | PROT_WRITE, MAP_PRIVATE, s->fd, 0))) {
+	l = lseek(fd, SEEK_CUR, 0);
+	if (l > 0) {
+		if (MAP_FAILED == (buf = mmap(NULL, l, PROT_READ | PROT_WRITE, MAP_PRIVATE, fd, 0))) {
 			ast_log(LOG_WARNING, "mmap failed.  Manager output was not processed\n");
 		} else {
 			if (format == FORMAT_XML || format == FORMAT_HTML) {
@@ -7614,6 +7587,7 @@ static int generic_http_callback(struct ast_tcptls_session_instance *ser,
 	struct mansession s = { .session = NULL, .tcptls_session = ser };
 	struct mansession_session *session = NULL;
 	uint32_t ident;
+	int fd;
 	int blastaway = 0;
 	struct ast_variable *v;
 	struct ast_variable *params = get_params;
@@ -7669,17 +7643,17 @@ static int generic_http_callback(struct ast_tcptls_session_instance *ser,
 	}
 
 	s.session = session;
-	s.fd = mkstemp(template);	/* create a temporary file for command output */
+	fd = mkstemp(template);	/* create a temporary file for command output */
 	unlink(template);
-	if (s.fd <= -1) {
+	if (fd <= -1) {
 		ast_http_error(ser, 500, "Server Error", "Internal Server Error (mkstemp failed)");
 		goto generic_callback_out;
 	}
-	s.f = fdopen(s.fd, "w+");
-	if (!s.f) {
+	s.stream = ast_iostream_from_fd(&fd);
+	if (!s.stream) {
 		ast_log(LOG_WARNING, "HTTP Manager, fdopen failed: %s!\n", strerror(errno));
 		ast_http_error(ser, 500, "Server Error", "Internal Server Error (fdopen failed)");
-		close(s.fd);
+		close(fd);
 		goto generic_callback_out;
 	}
 
@@ -7819,9 +7793,9 @@ generic_callback_out:
 		if (blastaway) {
 			session_destroy(session);
 		} else {
-			if (session->f) {
-				fclose(session->f);
-				session->f = NULL;
+			if (session->stream) {
+				ast_iostream_close(session->stream);
+				session->stream = NULL;
 			}
 			unref_mansession(session);
 		}
@@ -7846,6 +7820,7 @@ static int auth_http_callback(struct ast_tcptls_session_instance *ser,
 	struct message m = { 0 };
 	unsigned int idx;
 	size_t hdrlen;
+	int fd;
 
 	time_t time_now = time(NULL);
 	unsigned long nonce = 0, nc;
@@ -8024,17 +7999,17 @@ static int auth_http_callback(struct ast_tcptls_session_instance *ser,
 
 	ast_mutex_init(&s.lock);
 	s.session = session;
-	s.fd = mkstemp(template);	/* create a temporary file for command output */
+	fd = mkstemp(template);	/* create a temporary file for command output */
 	unlink(template);
-	if (s.fd <= -1) {
+	if (fd <= -1) {
 		ast_http_error(ser, 500, "Server Error", "Internal Server Error (mkstemp failed)");
 		goto auth_callback_out;
 	}
-	s.f = fdopen(s.fd, "w+");
-	if (!s.f) {
+	s.stream = ast_iostream_from_fd(&fd);
+	if (!s.stream) {
 		ast_log(LOG_WARNING, "HTTP Manager, fdopen failed: %s!\n", strerror(errno));
 		ast_http_error(ser, 500, "Server Error", "Internal Server Error (fdopen failed)");
-		close(s.fd);
+		close(fd);
 		goto auth_callback_out;
 	}
 
@@ -8085,7 +8060,7 @@ static int auth_http_callback(struct ast_tcptls_session_instance *ser,
 		m.headers[idx] = NULL;
 	}
 
-	result_size = ftell(s.f); /* Calculate approx. size of result */
+	result_size = lseek(ast_iostream_get_fd(s.stream), SEEK_CUR, 0); /* Calculate approx. size of result */
 
 	http_header = ast_str_create(80);
 	out = ast_str_create(result_size * 2 + 512);
@@ -8137,11 +8112,10 @@ auth_callback_out:
 	ast_free(out);
 
 	ao2_lock(session);
-	if (session->f) {
-		fclose(session->f);
+	if (session->stream) {
+		ast_iostream_close(session->stream);
+		session->stream = NULL;
 	}
-	session->f = NULL;
-	session->fd = -1;
 	ao2_unlock(session);
 
 	if (session->needdestroy) {
diff --git a/main/tcptls.c b/main/tcptls.c
index 262fca074..c8ebab434 100644
--- a/main/tcptls.c
+++ b/main/tcptls.c
@@ -47,506 +47,13 @@
 #include "asterisk/astobj2.h"
 #include "asterisk/pbx.h"
 
-/*! ao2 object used for the FILE stream fopencookie()/funopen() cookie. */
-struct ast_tcptls_stream {
-	/*! SSL state if not NULL */
-	SSL *ssl;
-	/*!
-	 * \brief Start time from when an I/O sequence must complete
-	 * by struct ast_tcptls_stream.timeout.
-	 *
-	 * \note If struct ast_tcptls_stream.start.tv_sec is zero then
-	 * start time is the current I/O request.
-	 */
-	struct timeval start;
-	/*!
-	 * \brief The socket returned by accept().
-	 *
-	 * \note Set to -1 if the stream is closed.
-	 */
-	int fd;
-	/*!
-	 * \brief Timeout in ms relative to struct ast_tcptls_stream.start
-	 * to wait for an event on struct ast_tcptls_stream.fd.
-	 *
-	 * \note Set to -1 to disable timeout.
-	 * \note The socket needs to be set to non-blocking for the timeout
-	 * feature to work correctly.
-	 */
-	int timeout;
-	/*! TRUE if stream can exclusively wait for fd input. */
-	int exclusive_input;
-};
-
-void ast_tcptls_stream_set_timeout_disable(struct ast_tcptls_stream *stream)
-{
-	ast_assert(stream != NULL);
-
-	stream->timeout = -1;
-}
-
-void ast_tcptls_stream_set_timeout_inactivity(struct ast_tcptls_stream *stream, int timeout)
-{
-	ast_assert(stream != NULL);
-
-	stream->start.tv_sec = 0;
-	stream->timeout = timeout;
-}
-
-void ast_tcptls_stream_set_timeout_sequence(struct ast_tcptls_stream *stream, struct timeval start, int timeout)
-{
-	ast_assert(stream != NULL);
-
-	stream->start = start;
-	stream->timeout = timeout;
-}
-
-void ast_tcptls_stream_set_exclusive_input(struct ast_tcptls_stream *stream, int exclusive_input)
-{
-	ast_assert(stream != NULL);
-
-	stream->exclusive_input = exclusive_input;
-}
-
-/*!
- * \internal
- * \brief fopencookie()/funopen() stream read function.
- *
- * \param cookie Stream control data.
- * \param buf Where to put read data.
- * \param size Size of the buffer.
- *
- * \retval number of bytes put into buf.
- * \retval 0 on end of file.
- * \retval -1 on error.
- */
-static HOOK_T tcptls_stream_read(void *cookie, char *buf, LEN_T size)
-{
-	struct ast_tcptls_stream *stream = cookie;
-	struct timeval start;
-	int ms;
-	int res;
-
-	if (!size) {
-		/* You asked for no data you got no data. */
-		return 0;
-	}
-
-	if (!stream || stream->fd == -1) {
-		errno = EBADF;
-		return -1;
-	}
-
-	if (stream->start.tv_sec) {
-		start = stream->start;
-	} else {
-		start = ast_tvnow();
-	}
-
-#if defined(DO_SSL)
-	if (stream->ssl) {
-		for (;;) {
-			res = SSL_read(stream->ssl, buf, size);
-			if (0 < res) {
-				/* We read some payload data. */
-				return res;
-			}
-			switch (SSL_get_error(stream->ssl, res)) {
-			case SSL_ERROR_ZERO_RETURN:
-				/* Report EOF for a shutdown */
-				ast_debug(1, "TLS clean shutdown alert reading data\n");
-				return 0;
-			case SSL_ERROR_WANT_READ:
-				if (!stream->exclusive_input) {
-					/* We cannot wait for data now. */
-					errno = EAGAIN;
-					return -1;
-				}
-				while ((ms = ast_remaining_ms(start, stream->timeout))) {
-					res = ast_wait_for_input(stream->fd, ms);
-					if (0 < res) {
-						/* Socket is ready to be read. */
-						break;
-					}
-					if (res < 0) {
-						if (errno == EINTR || errno == EAGAIN) {
-							/* Try again. */
-							continue;
-						}
-						ast_debug(1, "TLS socket error waiting for read data: %s\n",
-							strerror(errno));
-						return -1;
-					}
-				}
-				break;
-			case SSL_ERROR_WANT_WRITE:
-				while ((ms = ast_remaining_ms(start, stream->timeout))) {
-					res = ast_wait_for_output(stream->fd, ms);
-					if (0 < res) {
-						/* Socket is ready to be written. */
-						break;
-					}
-					if (res < 0) {
-						if (errno == EINTR || errno == EAGAIN) {
-							/* Try again. */
-							continue;
-						}
-						ast_debug(1, "TLS socket error waiting for write space: %s\n",
-							strerror(errno));
-						return -1;
-					}
-				}
-				break;
-			default:
-				/* Report EOF for an undecoded SSL or transport error. */
-				ast_debug(1, "TLS transport or SSL error reading data\n");
-				return 0;
-			}
-			if (!ms) {
-				/* Report EOF for a timeout */
-				ast_debug(1, "TLS timeout reading data\n");
-				return 0;
-			}
-		}
-	}
-#endif	/* defined(DO_SSL) */
-
-	for (;;) {
-		res = read(stream->fd, buf, size);
-		if (0 <= res || !stream->exclusive_input) {
-			/* Got data or we cannot wait for it. */
-			return res;
-		}
-		if (errno != EINTR && errno != EAGAIN) {
-			/* Not a retryable error. */
-			ast_debug(1, "TCP socket error reading data: %s\n",
-				strerror(errno));
-			return -1;
-		}
-		ms = ast_remaining_ms(start, stream->timeout);
-		if (!ms) {
-			/* Report EOF for a timeout */
-			ast_debug(1, "TCP timeout reading data\n");
-			return 0;
-		}
-		ast_wait_for_input(stream->fd, ms);
-	}
-}
-
-/*!
- * \internal
- * \brief fopencookie()/funopen() stream write function.
- *
- * \param cookie Stream control data.
- * \param buf Where to get data to write.
- * \param size Size of the buffer.
- *
- * \retval number of bytes written from buf.
- * \retval -1 on error.
- */
-static HOOK_T tcptls_stream_write(void *cookie, const char *buf, LEN_T size)
-{
-	struct ast_tcptls_stream *stream = cookie;
-	struct timeval start;
-	int ms;
-	int res;
-	int written;
-	int remaining;
-
-	if (!size) {
-		/* You asked to write no data you wrote no data. */
-		return 0;
-	}
-
-	if (!stream || stream->fd == -1) {
-		errno = EBADF;
-		return -1;
-	}
-
-	if (stream->start.tv_sec) {
-		start = stream->start;
-	} else {
-		start = ast_tvnow();
-	}
-
-#if defined(DO_SSL)
-	if (stream->ssl) {
-		written = 0;
-		remaining = size;
-		for (;;) {
-			res = SSL_write(stream->ssl, buf + written, remaining);
-			if (res == remaining) {
-				/* Everything was written. */
-				return size;
-			}
-			if (0 < res) {
-				/* Successfully wrote part of the buffer.  Try to write the rest. */
-				written += res;
-				remaining -= res;
-				continue;
-			}
-			switch (SSL_get_error(stream->ssl, res)) {
-			case SSL_ERROR_ZERO_RETURN:
-				ast_debug(1, "TLS clean shutdown alert writing data\n");
-				if (written) {
-					/* Report partial write. */
-					return written;
-				}
-				errno = EBADF;
-				return -1;
-			case SSL_ERROR_WANT_READ:
-				ms = ast_remaining_ms(start, stream->timeout);
-				if (!ms) {
-					/* Report partial write. */
-					ast_debug(1, "TLS timeout writing data (want read)\n");
-					return written;
-				}
-				ast_wait_for_input(stream->fd, ms);
-				break;
-			case SSL_ERROR_WANT_WRITE:
-				ms = ast_remaining_ms(start, stream->timeout);
-				if (!ms) {
-					/* Report partial write. */
-					ast_debug(1, "TLS timeout writing data (want write)\n");
-					return written;
-				}
-				ast_wait_for_output(stream->fd, ms);
-				break;
-			default:
-				/* Undecoded SSL or transport error. */
-				ast_debug(1, "TLS transport or SSL error writing data\n");
-				if (written) {
-					/* Report partial write. */
-					return written;
-				}
-				errno = EBADF;
-				return -1;
-			}
-		}
-	}
-#endif	/* defined(DO_SSL) */
-
-	written = 0;
-	remaining = size;
-	for (;;) {
-		res = write(stream->fd, buf + written, remaining);
-		if (res == remaining) {
-			/* Yay everything was written. */
-			return size;
-		}
-		if (0 < res) {
-			/* Successfully wrote part of the buffer.  Try to write the rest. */
-			written += res;
-			remaining -= res;
-			continue;
-		}
-		if (errno != EINTR && errno != EAGAIN) {
-			/* Not a retryable error. */
-			ast_debug(1, "TCP socket error writing: %s\n", strerror(errno));
-			if (written) {
-				return written;
-			}
-			return -1;
-		}
-		ms = ast_remaining_ms(start, stream->timeout);
-		if (!ms) {
-			/* Report partial write. */
-			ast_debug(1, "TCP timeout writing data\n");
-			return written;
-		}
-		ast_wait_for_output(stream->fd, ms);
-	}
-}
-
-/*!
- * \internal
- * \brief fopencookie()/funopen() stream close function.
- *
- * \param cookie Stream control data.
- *
- * \retval 0 on success.
- * \retval -1 on error.
- */
-static int tcptls_stream_close(void *cookie)
-{
-	struct ast_tcptls_stream *stream = cookie;
-
-	if (!stream) {
-		errno = EBADF;
-		return -1;
-	}
-
-	if (stream->fd != -1) {
-#if defined(DO_SSL)
-		if (stream->ssl) {
-			int res;
-
-			/*
-			 * According to the TLS standard, it is acceptable for an
-			 * application to only send its shutdown alert and then
-			 * close the underlying connection without waiting for
-			 * the peer's response (this way resources can be saved,
-			 * as the process can already terminate or serve another
-			 * connection).
-			 */
-			res = SSL_shutdown(stream->ssl);
-			if (res < 0) {
-				ast_log(LOG_ERROR, "SSL_shutdown() failed: %d\n",
-					SSL_get_error(stream->ssl, res));
-			}
-
-			if (!stream->ssl->server) {
-				/* For client threads, ensure that the error stack is cleared */
-#if OPENSSL_VERSION_NUMBER >= 0x10000000L
-				ERR_remove_thread_state(NULL);
-#else
-				ERR_remove_state(0);
-#endif	/* OPENSSL_VERSION_NUMBER >= 0x10000000L */
-			}
-
-			SSL_free(stream->ssl);
-			stream->ssl = NULL;
-		}
-#endif	/* defined(DO_SSL) */
-
-		/*
-		 * Issuing shutdown() is necessary here to avoid a race
-		 * condition where the last data written may not appear
-		 * in the TCP stream.  See ASTERISK-23548
-		 */
-		shutdown(stream->fd, SHUT_RDWR);
-		if (close(stream->fd)) {
-			ast_log(LOG_ERROR, "close() failed: %s\n", strerror(errno));
-		}
-		stream->fd = -1;
-	}
-	ao2_t_ref(stream, -1, "Closed tcptls stream cookie");
-
-	return 0;
-}
-
-/*!
- * \internal
- * \brief fopencookie()/funopen() stream destructor function.
- *
- * \param cookie Stream control data.
- *
- * \return Nothing
- */
-static void tcptls_stream_dtor(void *cookie)
-{
-#ifdef AST_DEVMODE
-	/* Since the ast_assert below is the only one using stream,
-	 * and ast_assert is only available with AST_DEVMODE, we
-	 * put this in a conditional to avoid compiler warnings. */
-	struct ast_tcptls_stream *stream = cookie;
-#endif
-
-	ast_assert(stream->fd == -1);
-}
-
-/*!
- * \internal
- * \brief fopencookie()/funopen() stream allocation function.
- *
- * \retval stream_cookie on success.
- * \retval NULL on error.
- */
-static struct ast_tcptls_stream *tcptls_stream_alloc(void)
-{
-	struct ast_tcptls_stream *stream;
-
-	stream = ao2_alloc_options(sizeof(*stream), tcptls_stream_dtor,
-		AO2_ALLOC_OPT_LOCK_NOLOCK);
-	if (stream) {
-		stream->fd = -1;
-		stream->timeout = -1;
-	}
-	return stream;
-}
-
-/*!
- * \internal
- * \brief Open a custom FILE stream for tcptls.
- *
- * \param stream Stream cookie control data.
- * \param ssl SSL state if not NULL.
- * \param fd Socket file descriptor.
- * \param timeout ms to wait for an event on fd. -1 if timeout disabled.
- *
- * \retval fp on success.
- * \retval NULL on error.
- */
-static FILE *tcptls_stream_fopen(struct ast_tcptls_stream *stream, SSL *ssl, int fd, int timeout)
-{
-	FILE *fp;
-
-#if defined(HAVE_FOPENCOOKIE)	/* the glibc/linux interface */
-	static const cookie_io_functions_t cookie_funcs = {
-		tcptls_stream_read,
-		tcptls_stream_write,
-		NULL,
-		tcptls_stream_close
-	};
-#endif	/* defined(HAVE_FOPENCOOKIE) */
-
-	if (fd == -1) {
-		/* Socket not open. */
-		return NULL;
-	}
-
-	stream->ssl = ssl;
-	stream->fd = fd;
-	stream->timeout = timeout;
-	ao2_t_ref(stream, +1, "Opening tcptls stream cookie");
-
-#if defined(HAVE_FUNOPEN)	/* the BSD interface */
-	fp = funopen(stream, tcptls_stream_read, tcptls_stream_write, NULL,
-		tcptls_stream_close);
-#elif defined(HAVE_FOPENCOOKIE)	/* the glibc/linux interface */
-	fp = fopencookie(stream, "w+", cookie_funcs);
-#else
-	/* could add other methods here */
-	ast_debug(2, "No stream FILE methods attempted!\n");
-	fp = NULL;
-#endif
-
-	if (!fp) {
-		stream->fd = -1;
-		ao2_t_ref(stream, -1, "Failed to open tcptls stream cookie");
-	}
-	return fp;
-}
-
-HOOK_T ast_tcptls_server_read(struct ast_tcptls_session_instance *tcptls_session, void *buf, size_t count)
-{
-	if (!tcptls_session->stream_cookie || tcptls_session->stream_cookie->fd == -1) {
-		ast_log(LOG_ERROR, "TCP/TLS read called on invalid stream.\n");
-		errno = EIO;
-		return -1;
-	}
-
-	return tcptls_stream_read(tcptls_session->stream_cookie, buf, count);
-}
-
-HOOK_T ast_tcptls_server_write(struct ast_tcptls_session_instance *tcptls_session, const void *buf, size_t count)
-{
-	if (!tcptls_session->stream_cookie || tcptls_session->stream_cookie->fd == -1) {
-		ast_log(LOG_ERROR, "TCP/TLS write called on invalid stream.\n");
-		errno = EIO;
-		return -1;
-	}
-
-	return tcptls_stream_write(tcptls_session->stream_cookie, buf, count);
-}
-
 static void session_instance_destructor(void *obj)
 {
 	struct ast_tcptls_session_instance *i = obj;
 
-	if (i->stream_cookie) {
-		ao2_t_ref(i->stream_cookie, -1, "Destroying tcptls session instance");
-		i->stream_cookie = NULL;
+	if (i->stream) {
+		ast_iostream_close(i->stream);
+		i->stream = NULL;
 	}
 	ast_free(i->overflow_buf);
 	ao2_cleanup(i->private_data);
@@ -591,9 +98,7 @@ static void *handle_tcptls_connection(void *data)
 {
 	struct ast_tcptls_session_instance *tcptls_session = data;
 #ifdef DO_SSL
-	int (*ssl_setup)(SSL *) = (tcptls_session->client) ? SSL_connect : SSL_accept;
-	int ret;
-	char err[256];
+	SSL *ssl;
 #endif
 
 	/* TCP/TLS connections are associated with external protocols, and
@@ -608,123 +113,94 @@ static void *handle_tcptls_connection(void *data)
 		return NULL;
 	}
 
-	tcptls_session->stream_cookie = tcptls_stream_alloc();
-	if (!tcptls_session->stream_cookie) {
-		ast_tcptls_close_session_file(tcptls_session);
-		ao2_ref(tcptls_session, -1);
-		return NULL;
-	}
+	if (tcptls_session->parent->tls_cfg) {
+#ifdef DO_SSL
+		if (ast_iostream_start_tls(&tcptls_session->stream, tcptls_session->parent->tls_cfg->ssl_ctx, tcptls_session->client) < 0) {
+			ast_tcptls_close_session_file(tcptls_session);
+			ao2_ref(tcptls_session, -1);
+			return NULL;
+		}
 
-	/*
-	* open a FILE * as appropriate.
-	*/
-	if (!tcptls_session->parent->tls_cfg) {
-		tcptls_session->f = tcptls_stream_fopen(tcptls_session->stream_cookie, NULL,
-			tcptls_session->fd, -1);
-		if (tcptls_session->f) {
-			if (setvbuf(tcptls_session->f, NULL, _IONBF, 0)) {
+		ssl = ast_iostream_get_ssl(tcptls_session->stream);
+		if ((tcptls_session->client && !ast_test_flag(&tcptls_session->parent->tls_cfg->flags, AST_SSL_DONT_VERIFY_SERVER))
+			|| (!tcptls_session->client && ast_test_flag(&tcptls_session->parent->tls_cfg->flags, AST_SSL_VERIFY_CLIENT))) {
+			X509 *peer;
+			long res;
+			peer = SSL_get_peer_certificate(ssl);
+			if (!peer) {
+				ast_log(LOG_ERROR, "No peer SSL certificate to verify\n");
 				ast_tcptls_close_session_file(tcptls_session);
+				ao2_ref(tcptls_session, -1);
+				return NULL;
 			}
-		}
-	}
-#ifdef DO_SSL
-	else if ( (tcptls_session->ssl = SSL_new(tcptls_session->parent->tls_cfg->ssl_ctx)) ) {
-		SSL_set_fd(tcptls_session->ssl, tcptls_session->fd);
-		if ((ret = ssl_setup(tcptls_session->ssl)) <= 0) {
-			ast_log(LOG_ERROR, "Problem setting up ssl connection: %s\n", ERR_error_string(ERR_get_error(), err));
-		} else if ((tcptls_session->f = tcptls_stream_fopen(tcptls_session->stream_cookie,
-			tcptls_session->ssl, tcptls_session->fd, -1))) {
-			if ((tcptls_session->client && !ast_test_flag(&tcptls_session->parent->tls_cfg->flags, AST_SSL_DONT_VERIFY_SERVER))
-				|| (!tcptls_session->client && ast_test_flag(&tcptls_session->parent->tls_cfg->flags, AST_SSL_VERIFY_CLIENT))) {
-				X509 *peer;
-				long res;
-				peer = SSL_get_peer_certificate(tcptls_session->ssl);
-				if (!peer) {
-					ast_log(LOG_ERROR, "No peer SSL certificate to verify\n");
-					ast_tcptls_close_session_file(tcptls_session);
-					ao2_ref(tcptls_session, -1);
-					return NULL;
-				}
 
-				res = SSL_get_verify_result(tcptls_session->ssl);
-				if (res != X509_V_OK) {
-					ast_log(LOG_ERROR, "Certificate did not verify: %s\n", X509_verify_cert_error_string(res));
-					X509_free(peer);
-					ast_tcptls_close_session_file(tcptls_session);
-					ao2_ref(tcptls_session, -1);
-					return NULL;
-				}
-				if (!ast_test_flag(&tcptls_session->parent->tls_cfg->flags, AST_SSL_IGNORE_COMMON_NAME)) {
-					ASN1_STRING *str;
-					X509_NAME *name = X509_get_subject_name(peer);
-					STACK_OF(GENERAL_NAME) *alt_names;
-					int pos = -1;
-					int found = 0;
-
-					for (;;) {
-						/* Walk the certificate to check all available "Common Name" */
-						/* XXX Probably should do a gethostbyname on the hostname and compare that as well */
-						pos = X509_NAME_get_index_by_NID(name, NID_commonName, pos);
-						if (pos < 0) {
-							break;
-						}
-
-						str = X509_NAME_ENTRY_get_data(X509_NAME_get_entry(name, pos));
-						if (!check_tcptls_cert_name(str, tcptls_session->parent->hostname, "common name")) {
-							found = 1;
-							break;
-						}
+			res = SSL_get_verify_result(ssl);
+			if (res != X509_V_OK) {
+				ast_log(LOG_ERROR, "Certificate did not verify: %s\n", X509_verify_cert_error_string(res));
+				X509_free(peer);
+				ast_tcptls_close_session_file(tcptls_session);
+				ao2_ref(tcptls_session, -1);
+				return NULL;
+			}
+			if (!ast_test_flag(&tcptls_session->parent->tls_cfg->flags, AST_SSL_IGNORE_COMMON_NAME)) {
+				ASN1_STRING *str;
+				X509_NAME *name = X509_get_subject_name(peer);
+				STACK_OF(GENERAL_NAME) *alt_names;
+				int pos = -1;
+				int found = 0;
+
+				for (;;) {
+					/* Walk the certificate to check all available "Common Name" */
+					/* XXX Probably should do a gethostbyname on the hostname and compare that as well */
+					pos = X509_NAME_get_index_by_NID(name, NID_commonName, pos);
+					if (pos < 0) {
+						break;
 					}
+					str = X509_NAME_ENTRY_get_data(X509_NAME_get_entry(name, pos));
+					if (!check_tcptls_cert_name(str, tcptls_session->parent->hostname, "common name")) {
+						found = 1;
+						break;
+					}
+				}
 
-					if (!found) {
-						alt_names = X509_get_ext_d2i(peer, NID_subject_alt_name, NULL, NULL);
-						if (alt_names != NULL) {
-							int alt_names_count = sk_GENERAL_NAME_num(alt_names);
-
-							for (pos = 0; pos < alt_names_count; pos++) {
-								const GENERAL_NAME *alt_name = sk_GENERAL_NAME_value(alt_names, pos);
+				if (!found) {
+					alt_names = X509_get_ext_d2i(peer, NID_subject_alt_name, NULL, NULL);
+					if (alt_names != NULL) {
+						int alt_names_count = sk_GENERAL_NAME_num(alt_names);
 
-								if (alt_name->type != GEN_DNS) {
-									continue;
-								}
+						for (pos = 0; pos < alt_names_count; pos++) {
+							const GENERAL_NAME *alt_name = sk_GENERAL_NAME_value(alt_names, pos);
 
-								if (!check_tcptls_cert_name(alt_name->d.dNSName, tcptls_session->parent->hostname, "alt name")) {
-									found = 1;
-									break;
-								}
+							if (alt_name->type != GEN_DNS) {
+								continue;
 							}
 
-							sk_GENERAL_NAME_pop_free(alt_names, GENERAL_NAME_free);
+							if (!check_tcptls_cert_name(alt_name->d.dNSName, tcptls_session->parent->hostname, "alt name")) {
+								found = 1;
+								break;
+							}
 						}
-					}
 
-					if (!found) {
-						ast_log(LOG_ERROR, "Certificate common name did not match (%s)\n", tcptls_session->parent->hostname);
-						X509_free(peer);
-						ast_tcptls_close_session_file(tcptls_session);
-						ao2_ref(tcptls_session, -1);
-						return NULL;
+						sk_GENERAL_NAME_pop_free(alt_names, GENERAL_NAME_free);
 					}
 				}
-				X509_free(peer);
+
+				if (!found) {
+					ast_log(LOG_ERROR, "Certificate common name did not match (%s)\n", tcptls_session->parent->hostname);
+					X509_free(peer);
+					ast_tcptls_close_session_file(tcptls_session);
+					ao2_ref(tcptls_session, -1);
+					return NULL;
+				}
 			}
+			X509_free(peer);
 		}
-		if (!tcptls_session->f) {	/* no success opening descriptor stacking */
-			SSL_free(tcptls_session->ssl);
-		}
-	}
-#endif /* DO_SSL */
-
-	if (!tcptls_session->f) {
+#else
+		ast_log(LOG_ERROR, "Attempted a TLS connection without OpenSSL support. This will not work!\n");
 		ast_tcptls_close_session_file(tcptls_session);
-		ast_log(LOG_WARNING, "FILE * open failed!\n");
-#ifndef DO_SSL
-		if (tcptls_session->parent->tls_cfg) {
-			ast_log(LOG_ERROR, "Attempted a TLS connection without OpenSSL support. This will not work!\n");
-		}
-#endif
 		ao2_ref(tcptls_session, -1);
 		return NULL;
+#endif /* DO_SSL */
 	}
 
 	if (tcptls_session->parent->worker_fn) {
@@ -772,7 +248,13 @@ void *ast_tcptls_server_root(void *data)
 		tcptls_session->overflow_buf = ast_str_create(128);
 		flags = fcntl(fd, F_GETFL);
 		fcntl(fd, F_SETFL, flags & ~O_NONBLOCK);
-		tcptls_session->fd = fd;
+
+		tcptls_session->stream = ast_iostream_from_fd(&fd);
+		if (!tcptls_session->stream) {
+			ast_log(LOG_WARNING, "No memory for new session iostream\n");
+			continue;
+		}
+
 		tcptls_session->parent = desc;
 		ast_sockaddr_copy(&tcptls_session->remote_address, &addr);
 
@@ -1036,7 +518,7 @@ client_start_error:
 
 struct ast_tcptls_session_instance *ast_tcptls_client_create(struct ast_tcptls_session_args *desc)
 {
-	int x = 1;
+	int fd, x = 1;
 	struct ast_tcptls_session_instance *tcptls_session = NULL;
 
 	/* Do nothing if nothing has changed */
@@ -1052,8 +534,8 @@ struct ast_tcptls_session_instance *ast_tcptls_client_create(struct ast_tcptls_s
 		close(desc->accept_fd);
 	}
 
-	desc->accept_fd = socket(ast_sockaddr_is_ipv6(&desc->remote_address) ?
-				 AF_INET6 : AF_INET, SOCK_STREAM, IPPROTO_TCP);
+	fd = desc->accept_fd = socket(ast_sockaddr_is_ipv6(&desc->remote_address) ?
+				      AF_INET6 : AF_INET, SOCK_STREAM, IPPROTO_TCP);
 	if (desc->accept_fd < 0) {
 		ast_log(LOG_ERROR, "Unable to allocate socket for %s: %s\n",
 			desc->name, strerror(errno));
@@ -1079,7 +561,11 @@ struct ast_tcptls_session_instance *ast_tcptls_client_create(struct ast_tcptls_s
 
 	tcptls_session->overflow_buf = ast_str_create(128);
 	tcptls_session->client = 1;
-	tcptls_session->fd = desc->accept_fd;
+	tcptls_session->stream = ast_iostream_from_fd(&fd);
+	if (!tcptls_session->stream) {
+		goto error;
+	}
+
 	tcptls_session->parent = desc;
 	tcptls_session->parent->worker_fn = NULL;
 	ast_sockaddr_copy(&tcptls_session->remote_address,
@@ -1170,24 +656,9 @@ error:
 
 void ast_tcptls_close_session_file(struct ast_tcptls_session_instance *tcptls_session)
 {
-	if (tcptls_session->f) {
-		fflush(tcptls_session->f);
-		if (fclose(tcptls_session->f)) {
-			ast_log(LOG_ERROR, "fclose() failed: %s\n", strerror(errno));
-		}
-		tcptls_session->f = NULL;
-		tcptls_session->fd = -1;
-	} else if (tcptls_session->fd != -1) {
-		/*
-		 * Issuing shutdown() is necessary here to avoid a race
-		 * condition where the last data written may not appear
-		 * in the TCP stream.  See ASTERISK-23548
-		 */
-		shutdown(tcptls_session->fd, SHUT_RDWR);
-		if (close(tcptls_session->fd)) {
-			ast_log(LOG_ERROR, "close() failed: %s\n", strerror(errno));
-		}
-		tcptls_session->fd = -1;
+	if (tcptls_session->stream) {
+		ast_iostream_close(tcptls_session->stream);
+		tcptls_session->stream = NULL;
 	} else {
 		ast_log(LOG_ERROR, "ast_tcptls_close_session_file invoked on session instance without file or file descriptor\n");
 	}
diff --git a/main/utils.c b/main/utils.c
index 775fae3af..2c56af3cd 100644
--- a/main/utils.c
+++ b/main/utils.c
@@ -1462,68 +1462,6 @@ int ast_carefulwrite(int fd, char *s, int len, int timeoutms)
 	return res;
 }
 
-int ast_careful_fwrite(FILE *f, int fd, const char *src, size_t len, int timeoutms)
-{
-	struct timeval start = ast_tvnow();
-	int n = 0;
-	int elapsed = 0;
-
-	while (len) {
-		if (wait_for_output(fd, timeoutms - elapsed)) {
-			/* poll returned a fatal error, so bail out immediately. */
-			return -1;
-		}
-
-		/* Clear any errors from a previous write */
-		clearerr(f);
-
-		n = fwrite(src, 1, len, f);
-
-		if (ferror(f) && errno != EINTR && errno != EAGAIN) {
-			/* fatal error from fwrite() */
-			if (!feof(f)) {
-				/* Don't spam the logs if it was just that the connection is closed. */
-				ast_log(LOG_ERROR, "fwrite() returned error: %s\n", strerror(errno));
-			}
-			n = -1;
-			break;
-		}
-
-		/* Update for data already written to the socket */
-		len -= n;
-		src += n;
-
-		elapsed = ast_tvdiff_ms(ast_tvnow(), start);
-		if (elapsed >= timeoutms) {
-			/* We've taken too long to write
-			 * This is only an error condition if we haven't finished writing. */
-			n = len ? -1 : 0;
-			break;
-		}
-	}
-
-	errno = 0;
-	while (fflush(f)) {
-		if (errno == EAGAIN || errno == EINTR) {
-			/* fflush() does not appear to reset errno if it flushes
-			 * and reaches EOF at the same time. It returns EOF with
-			 * the last seen value of errno, causing a possible loop.
-			 * Also usleep() to reduce CPU eating if it does loop */
-			errno = 0;
-			usleep(1);
-			continue;
-		}
-		if (errno && !feof(f)) {
-			/* Don't spam the logs if it was just that the connection is closed. */
-			ast_log(LOG_ERROR, "fflush() returned error: %s\n", strerror(errno));
-		}
-		n = -1;
-		break;
-	}
-
-	return n < 0 ? -1 : 0;
-}
-
 char *ast_strip_quoted(char *s, const char *beg_quotes, const char *end_quotes)
 {
 	char *e;