/*
 * Amanda, The Advanced Maryland Automatic Network Disk Archiver
 * Copyright (c) 1991-1999 University of Maryland at College Park
 * Copyright (c) 2007-2012 Zmanda, Inc.  All Rights Reserved.
 * Copyright (c) 2013-2016 Carbonite, Inc.  All Rights Reserved.
 * All Rights Reserved.
 *
 * Permission to use, copy, modify, distribute, and sell this software and its
 * documentation for any purpose is hereby granted without fee, provided that
 * the above copyright notice appear in all copies and that both that
 * copyright notice and this permission notice appear in supporting
 * documentation, and that the name of U.M. not be used in advertising or
 * publicity pertaining to distribution of the software without specific,
 * written prior permission.  U.M. makes no representations about the
 * suitability of this software for any purpose.  It is provided "as is"
 * without express or implied warranty.
 *
 * U.M. DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING ALL
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN NO EVENT SHALL U.M.
 * BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION
 * OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 *
 * Authors: the Amanda Development Team.  Its members are listed in a
 * file named AUTHORS, in the root directory of this distribution.
 */
/*
 * $Id: bsd-security.c,v 1.75 2006/07/19 17:41:14 martinea Exp $
 *
 * "BSD" security module
 */

#include "amanda.h"
#include "amutil.h"
#include "clock.h"
#include "dgram.h"
#include "event.h"
#include "packet.h"
#include "security.h"
#include "security-util.h"
#include "sockaddr-util.h"
#include "stream.h"

#ifndef SO_RCVBUF
#undef DUMPER_SOCKET_BUFFERING
#endif

/*
 * Interface functions
 */
static void	bsd_connect(const char *, char *(*)(char *, void *), 
			void (*)(void *, security_handle_t *, security_status_t),
			void *, void *);
static void	bsd_accept(const struct security_driver *,
			char *(*)(char *, void *),
			int, int,
			void (*)(security_handle_t *, pkt_t *),
			void *);
static void	bsd_close(void *);
static void *	bsd_stream_server(void *);
static int	bsd_stream_accept(void *);
static void *	bsd_stream_client(void *, int);
static void	bsd_stream_close(void *);
static void	bsd_stream_close_async(void *s, void (*fn)(void *, ssize_t, void *, ssize_t), void *arg);
static int	bsd_stream_auth(void *);
static int	bsd_stream_id(void *);
static void	bsd_stream_read(void *, void (*)(void *, void *, ssize_t), void *);
static ssize_t	bsd_stream_read_sync(void *, void **);
static void     bsd_stream_read_to_shm_ring(void *s, void (*fn)(void *, void *, ssize_t), shm_ring_t *shm_ring, void *arg);
static void     bsd_stream_read_to_shm_ring_callback(void *arg);
static void	bsd_stream_read_cancel(void *);
static void	bsd_stream_pause(void *);
static void	bsd_stream_resume(void *);

/*
 * This is our interface to the outside world
 */
const security_driver_t bsd_security_driver = {
    "BSD",
    bsd_connect,
    bsd_accept,
    sec_get_authenticated_peer_name_hostname,
    bsd_close,
    udpbsd_sendpkt,
    udp_recvpkt,
    udp_recvpkt_cancel,
    bsd_stream_server,
    bsd_stream_accept,
    bsd_stream_client,
    bsd_stream_close,
    bsd_stream_close_async,
    bsd_stream_auth,
    bsd_stream_id,
    tcp_stream_write,
    tcp_stream_write_async,
    bsd_stream_read,
    bsd_stream_read_sync,
    bsd_stream_read_to_shm_ring,
    bsd_stream_read_cancel,
    bsd_stream_pause,
    bsd_stream_resume,
    sec_close_connection_none,
    NULL,
    NULL,
    generic_data_write,
    generic_data_write_non_blocking,
    generic_data_read
};

/*
 * This is data local to the datagram socket.  We have one datagram
 * per process, so it is global.
 */
static udp_handle_t netfd4;
static udp_handle_t netfd6;
static int not_init4 = 1;
static int not_init6 = 1;

/* generate new handles from here */
static int newhandle = 0;

/*
 * These are the internal helper functions
 */
static void	stream_read_callback(void *);
static void	stream_read_sync_callback(void *);

/*
 * Setup and return a handle outgoing to a client
 */

static void
bsd_connect(
    const char *	hostname,
    char *		(*conf_fn)(char *, void *),
    void		(*fn)(void *, security_handle_t *, security_status_t),
    void *		arg,
    void *		datap)
{
    struct sec_handle *bh;
    in_port_t port = 0;
    struct timeval sequence_time;
    int sequence;
    char *handle;
    int result;
    struct addrinfo *res, *res_addr;
    char *canonname;
    int result_bind;
    char *service;
    char *bind_msg = NULL;

    assert(hostname != NULL);

    (void)conf_fn;	/* Quiet unused parameter warning */
    (void)datap;        /* Quiet unused parameter warning */

    bh = g_new0(struct sec_handle, 1);
    bh->proto_handle=NULL;
    bh->dle_hostname = g_strdup(hostname);
    security_handleinit(&bh->sech, &bsd_security_driver);

    result = resolve_hostname(hostname, SOCK_DGRAM, &res, &canonname);
    if(result != 0) {
	dbprintf(_("resolve_hostname(%s): %s\n"), hostname, gai_strerror(result));
	security_seterror(&bh->sech, _("resolve_hostname(%s): %s"), hostname,
			  gai_strerror(result));
	(*fn)(arg, &bh->sech, S_ERROR);
	return;
    }
    if (canonname == NULL) {
	dbprintf(_("resolve_hostname(%s) did not return a canonical name\n"), hostname);
	security_seterror(&bh->sech,
	        _("resolve_hostname(%s) did not return a canonical name"), hostname);
	(*fn)(arg, &bh->sech, S_ERROR);
	if (res) freeaddrinfo(res);
	return;
    }
    if (res == NULL) {
	dbprintf(_("resolve_hostname(%s): no results\n"), hostname);
	security_seterror(&bh->sech,
	        _("resolve_hostname(%s): no results"), hostname);
	(*fn)(arg, &bh->sech, S_ERROR);
	amfree(canonname);
	return;
    }

    for (res_addr = res; res_addr != NULL; res_addr = res_addr->ai_next) {
#ifdef WORKING_IPV6
	/* IPv6 socket already bound */
	if (res_addr->ai_addr->sa_family == AF_INET6 && not_init6 == 0) {
	    break;
	}
	/*
	 * Only init the IPv6 socket once
	 */
	if (res_addr->ai_addr->sa_family == AF_INET6 && not_init6 == 1) {
	    dgram_zero(&netfd6.dgram);

	    result_bind = dgram_bind(&netfd6.dgram,
				     res_addr->ai_addr->sa_family, &port, 1, &bind_msg);
	    if (bind_msg) {
		continue;
	    }
	    if (result_bind != 0) {
		continue;
	    }
	    netfd6.handle = NULL;
	    netfd6.pkt.body = NULL;
	    netfd6.recv_security_ok = &bsd_recv_security_ok;
	    netfd6.prefix_packet = &bsd_prefix_packet;
	    netfd6.need_priv_port = 1;
	    /*
	     * We must have a reserved port.  Bomb if we didn't get one.
	     */
	    if (port >= IPPORT_RESERVED) {
		security_seterror(&bh->sech,
		    _("unable to bind to a reserved port (got port %u)"),
		    (unsigned int)port);
		(*fn)(arg, &bh->sech, S_ERROR);
		freeaddrinfo(res);
		amfree(canonname);
		return;
	    }
	    not_init6 = 0;
	    bh->udp = &netfd6;
	    break;
	}
#endif

	/* IPv4 socket already bound */
	if (res_addr->ai_addr->sa_family == AF_INET && not_init4 == 0) {
	    break;
	}

	/*
	 * Only init the IPv4 socket once
	 */
	if (res_addr->ai_addr->sa_family == AF_INET && not_init4 == 1) {
	    dgram_zero(&netfd4.dgram);

	    result_bind = dgram_bind(&netfd4.dgram,
				     res_addr->ai_addr->sa_family, &port, 1, &bind_msg);
	    if (bind_msg) {
		continue;
	    }
	    if (result_bind != 0) {
		continue;
	    }
	    netfd4.handle = NULL;
	    netfd4.pkt.body = NULL;
	    netfd4.recv_security_ok = &bsd_recv_security_ok;
	    netfd4.prefix_packet = &bsd_prefix_packet;
	    netfd4.need_priv_port = 1;
	    /*
	     * We must have a reserved port.  Bomb if we didn't get one.
	     */
	    if (port >= IPPORT_RESERVED) {
		security_seterror(&bh->sech,
		    "unable to bind to a reserved port (got port %u)",
		    (unsigned int)port);
		(*fn)(arg, &bh->sech, S_ERROR);
		freeaddrinfo(res);
		amfree(canonname);
		return;
	    }
	    not_init4 = 0;
	    bh->udp = &netfd4;
	    break;
	}
    }

    if (bind_msg) {
	g_debug("%s", bind_msg);
	security_seterror(&bh->sech,
	        "%s", bind_msg);
	g_free(bind_msg);
	(*fn)(arg, &bh->sech, S_ERROR);
	amfree(canonname);
	freeaddrinfo(res);
	return;
    }
    if (res_addr == NULL) {
	dbprintf(_("Can't bind a socket to connect to %s\n"), hostname);
	security_seterror(&bh->sech,
	        _("Can't bind a socket to connect to %s"), hostname);
	(*fn)(arg, &bh->sech, S_ERROR);
	amfree(canonname);
	freeaddrinfo(res);
	return;
    }

#ifdef WORKING_IPV6
    if (res_addr->ai_addr->sa_family == AF_INET6)
	bh->udp = &netfd6;
    else
#endif
	bh->udp = &netfd4;

    auth_debug(1, _("Resolved hostname=%s\n"), canonname);

    if (conf_fn) {
        service = conf_fn("client_port", datap);
        if (!service || strlen(service) <= 1)
            service = AMANDA_SERVICE_NAME;
    } else {
        service = AMANDA_SERVICE_NAME;
    }
    port = find_port_for_service(service, "udp");
    if (port == 0) {
        security_seterror(&bh->sech, _("%s/udp unknown protocol"), service);
	(*fn)(arg, &bh->sech, S_ERROR);
        amfree(canonname);
	freeaddrinfo(res);
	return;
    }

    amanda_gettimeofday(&sequence_time);
    sequence = (int)sequence_time.tv_sec ^ (int)sequence_time.tv_usec;
    handle=g_malloc(15);
    g_snprintf(handle, 14, "000-%08x",  (unsigned)newhandle++);
    if (udp_inithandle(bh->udp, bh, canonname,
	(sockaddr_union *)res_addr->ai_addr, port, handle, sequence) < 0) {
	(*fn)(arg, &bh->sech, S_ERROR);
	amfree(bh->hostname);
	amfree(bh);
    }
    else {
	(*fn)(arg, &bh->sech, S_OK);
    }
    amfree(handle);
    amfree(canonname);

    freeaddrinfo(res);
}

/*
 * Setup to accept new incoming connections
 */
static void
bsd_accept(
    const struct security_driver *	driver,
    char       *(*conf_fn)(char *, void *),
    int		in,
    int		out,
    void	(*fn)(security_handle_t *, pkt_t *),
    void       *datap)
{
    struct stat sbuf;

    assert(in >= 0 && out >= 0);
    assert(fn != NULL);

    (void)out;	/* Quiet unused parameter warning */
    (void)driver; /* Quiet unused parameter warning */
    (void)conf_fn;
    (void)datap;

    /*
     * We assume in and out point to the same socket, and just use
     * in.
     */
    dgram_socket(&netfd4.dgram, in);
    dgram_socket(&netfd6.dgram, in);

    /*
     * Assign the function and return.  When they call recvpkt later,
     * the recvpkt callback will call this function when it discovers
     * new incoming connections
     */
    netfd4.accept_fn = fn;
    netfd4.recv_security_ok = &bsd_recv_security_ok;
    netfd4.prefix_packet = &bsd_prefix_packet;
    netfd4.need_priv_port = 1;
    netfd4.driver = &bsd_security_driver;

    /* check if in is a socket */
    if (fstat(in, &sbuf) == -1) {
	g_warning("Can't fstat file descriptor; cannot use BSD auth");
    } else if (S_ISSOCK(sbuf.st_mode)) {
	udp_addref(&netfd4, &udp_netfd_read_callback);
    } else {
	g_warning("input file descriptor is not a socket; cannot use BSD auth");
    }
}

/*
 * Frees a handle allocated by the above
 */
static void
bsd_close(
    void *	cookie)
{
    struct sec_handle *bh = cookie;

    if(bh->proto_handle == NULL) {
	return;
    }

    auth_debug(1, _("bsd: close handle '%s'\n"), bh->proto_handle);

    udp_recvpkt_cancel(bh);
    if(bh->next) {
	bh->next->prev = bh->prev;
    }
    else {
	if (!not_init6 && netfd6.bh_last == bh)
	    netfd6.bh_last = bh->prev;
	else
	    netfd4.bh_last = bh->prev;
    }
    if(bh->prev) {
	bh->prev->next = bh->next;
    }
    else {
	if (!not_init6 && netfd6.bh_first == bh)
	    netfd6.bh_first = bh->next;
	else
	    netfd4.bh_first = bh->next;
    }

    amfree(bh->proto_handle);
    amfree(bh->hostname);
    amfree(bh);
}

/*
 * Create the server end of a stream.  For bsd, this means setup a tcp
 * socket for receiving a connection.
 */
static void *
bsd_stream_server(
    void *	h)
{
    struct sec_stream *bs = NULL;
    struct sec_handle *bh = h;

    assert(bh != NULL);

    bs = g_new0(struct sec_stream, 1);
    security_streaminit(&bs->secstr, &bsd_security_driver);
    bs->socket = stream_server(SU_GET_FAMILY(&bh->udp->peer), &bs->port,
			       (size_t)STREAM_BUFSIZE, (size_t)STREAM_BUFSIZE,
			       0);
    if (bs->socket < 0) {
	security_seterror(&bh->sech,
	    _("can't create server stream: %s"), strerror(errno));
	amfree(bs->secstr.error);
	amfree(bs);
	return (NULL);
    }
    bs->fd = -1;
    bs->ev_read = NULL;
    return (bs);
}

/*
 * Accepts a new connection on unconnected streams.  Assumes it is ok to
 * block on accept()
 */
static int
bsd_stream_accept(
    void *	s)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);
    assert(bs->socket != -1);
    assert(bs->fd < 0);

    bs->fd = stream_accept(bs->socket, 30, STREAM_BUFSIZE, STREAM_BUFSIZE);
    if (bs->fd < 0) {
	security_stream_seterror(&bs->secstr,
	    _("can't accept new stream connection: %s"), strerror(errno));
	return (-1);
    }
    return (0);
}

/*
 * Return a connected stream
 */
static void *
bsd_stream_client(
    void *	h,
    int		id)
{
    struct sec_stream *bs = NULL;
    struct sec_handle *bh = h;
#ifdef DUMPER_SOCKET_BUFFERING
    int rcvbuf = sizeof(bs->databuf) * 2;
#endif
    char *stream_msg = NULL;

    assert(bh != NULL);

    bs = g_new0(struct sec_stream, 1);
    security_streaminit(&bs->secstr, &bsd_security_driver);
    bs->fd = stream_client(NULL, bh->hostname, (in_port_t)id,
	STREAM_BUFSIZE, STREAM_BUFSIZE, &bs->port, 0, &stream_msg);
    if (stream_msg) {
	security_seterror(&bh->sech, "can't connect stream to %s port %d: %s",
			  bh->hostname, id, stream_msg);
	amfree(bs->secstr.error);
	amfree(bs);
	g_free(stream_msg);
	return (NULL);
    }
    if (bs->fd < 0) {
	security_seterror(&bh->sech, "can't connect stream to %s port %d: %s",
			  bh->hostname, id, strerror(errno));
	amfree(bs->secstr.error);
	amfree(bs);
	return (NULL);
    }
    bs->socket = -1;	/* we're a client */
    bs->ev_read = NULL;
#ifdef DUMPER_SOCKET_BUFFERING
    setsockopt(bs->fd, SOL_SOCKET, SO_RCVBUF, (void *)&rcvbuf, sizeof(rcvbuf));
#endif
    return (bs);
}

/*
 * Close and unallocate resources for a stream
 */
static void
bsd_stream_close(
    void *	s)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);

    if (bs->fd != -1)
	aclose(bs->fd);
    if (bs->socket != -1)
	aclose(bs->socket);
    bsd_stream_read_cancel(bs);
    amfree(bs);
}

/*
 * Close and unallocate resources for a stream
 */
static void
bsd_stream_close_async(
    void *	s,
    void (*fn)(void *, ssize_t, void *, ssize_t),
    void *arg)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);

    if (bs->fd != -1)
	aclose(bs->fd);
    if (bs->socket != -1)
	aclose(bs->socket);
    bsd_stream_read_cancel(bs);
    (*fn)(arg, 0, NULL, 0);
    amfree(bs);
}

/*
 * Authenticate a stream.  bsd streams have no authentication
 */
static int
bsd_stream_auth(
    void *	s)
{
    (void)s;		/* Quiet unused parameter warning */

    return (0);	/* success */
}

/*
 * Returns the stream id for this stream.  This is just the local port.
 */
static int
bsd_stream_id(
    void *	s)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);

    return ((int)bs->port);
}

/*
 * Submit a request to read some data.  Calls back with the given function
 * and arg when completed.
 */
static void
bsd_stream_read(
    void *	s,
    void	(*fn)(void *, void *, ssize_t),
    void *	arg)
{
    struct sec_stream *bs = s;

    /*
     * Only one read request can be active per stream.
     */
    if (bs->ev_read != NULL)
	event_release(bs->ev_read);

    bs->ev_read = event_create((event_id_t)bs->fd, EV_READFD, stream_read_callback, bs);
    bs->fn = fn;
    bs->arg = arg;
    event_activate(bs->ev_read);
}

/* buffer for bsd_stream_read_sync function */
static ssize_t  sync_pktlen;
static void    *sync_pkt;

/*
 * Read a chunk of data to a stream.  Blocks until completion.
 */
static ssize_t
bsd_stream_read_sync(
    void *	s,
    void **	buf)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);

    /*
     * Only one read request can be active per stream.
     */
    if(bs->ev_read != NULL) {
        return -1;
    }
    sync_pktlen = 0;
    sync_pkt = NULL;
    bs->ev_read = event_create((event_id_t)bs->fd, EV_READFD,
			stream_read_sync_callback, bs);
    event_activate(bs->ev_read);
    event_wait(bs->ev_read);
    *buf = sync_pkt;
    return (sync_pktlen);
}


/*
 * Callback for bsd_stream_read_sync
 */
static void
stream_read_sync_callback(
    void *	s)
{
    struct sec_stream *bs = s;
    ssize_t n;

    assert(bs != NULL);

    auth_debug(1, _("bsd: stream_read_callback_sync: fd %d\n"), bs->fd);

    /*
     * Remove the event first, in case they reschedule it in the callback.
     */
    bsd_stream_read_cancel(bs);
    do {
	n = read(bs->fd, bs->databuf, sizeof(bs->databuf));
    } while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));
    if (n < 0)
        security_stream_seterror(&bs->secstr, "%s", strerror(errno));
    bs->len = n;
    sync_pktlen = bs->len;
    if (sync_pktlen > 0) {
	sync_pkt = malloc(sync_pktlen);
	memcpy(sync_pkt, bs->databuf, sync_pktlen);
    } else {
	sync_pkt = NULL;
    }
}

static void
bsd_stream_read_to_shm_ring_callback(
    void *arg)
{
    struct sec_stream *bs = arg;
    ssize_t n = -1;
    size_t to_read;
    uint64_t write_offset;
    uint64_t readx;
    uint64_t written;
    uint64_t shm_ring_size;
    char     *buf = NULL;


    assert(bs != NULL);

    if (!bs->ring_init) {
	shm_ring_producer_set_size(bs->shm_ring, NETWORK_BLOCK_BYTES*8, NETWORK_BLOCK_BYTES);
	bs->ring_init = TRUE;
    }
    to_read = NETWORK_BLOCK_BYTES;
    write_offset = bs->shm_ring->mc->write_offset;
    written = bs->shm_ring->mc->written;
    shm_ring_size = bs->shm_ring->mc->ring_size;

    /* wait for enough space in the ring to read the packet */
    while (!bs->shm_ring->mc->cancelled) {
	readx = bs->shm_ring->mc->readx;
	if (shm_ring_size - (written - readx) > to_read)
	    break;
	if (shm_ring_sem_wait(bs->shm_ring, bs->shm_ring->sem_write) != 0) {
	    auth_debug(1, _("bsd_stream_read_to_shm_ring_callback: A return(-1)\n"));
	    goto ring_failed;
	}
    }
    if (bs->shm_ring->mc->cancelled) {
	auth_debug(1, _("bsd_stream_read_to_shm_ring_callback: B return(-1)\n"));
	goto ring_failed;
    }

    buf = bs->shm_ring->data + write_offset;
    if (write_offset + to_read <= shm_ring_size) {
	do {
	    n = read(bs->fd, bs->shm_ring->data + write_offset,
			     to_read);
	} while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));
    } else {
	do {
	    n = read(bs->fd, bs->shm_ring->data + write_offset,
			     shm_ring_size - write_offset);
	} while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));
	if (n == (int64_t)(shm_ring_size - write_offset)) {
	    ssize_t rval1;
	    do {
		rval1 = read(bs->fd, bs->shm_ring->data,
				to_read -(shm_ring_size - write_offset));
	    } while ((rval1 < 0) && ((errno == EINTR) || (errno == EAGAIN)));
	    if (rval1 > 0) {
		n += rval1;
	    }
	}
    }

ring_failed:
    if (n < 0) {
	security_stream_seterror(&bs->secstr, "%s", strerror(errno));
	bsd_stream_read_cancel(bs);
	bs->shm_ring->mc->cancelled = TRUE;
	bs->shm_ring->mc->eof_flag = TRUE;
	sem_post(bs->shm_ring->sem_read);
	sem_post(bs->shm_ring->sem_read);
	sem_post(bs->shm_ring->sem_write);
	auth_debug(1, _("bsd_stream_read_to_shm_ring_callback: C return(-1)\n"));
    } else if (n == 0) {
	bsd_stream_read_cancel(bs);
	bs->shm_ring->mc->eof_flag = TRUE;
	sem_post(bs->shm_ring->sem_read);
	sem_post(bs->shm_ring->sem_read);
    } else {
	if (bs->shm_ring->mc->written == 0 && bs->shm_ring->mc->need_sem_ready) {
	    sem_post(bs->shm_ring->sem_ready);
	    if (shm_ring_sem_wait(bs->shm_ring, bs->shm_ring->sem_start) != 0) {
		security_stream_seterror(&bs->secstr, "%s", strerror(errno));
		bsd_stream_read_cancel(bs);
		bs->shm_ring->mc->cancelled = TRUE;
		bs->shm_ring->mc->eof_flag = TRUE;
		sem_post(bs->shm_ring->sem_read);
		sem_post(bs->shm_ring->sem_read);
		sem_post(bs->shm_ring->sem_write);
		auth_debug(1, _("bsd_stream_read_to_shm_ring_callback: D return(-1)\n"));
		goto shm_failed;
	    }
	}
	write_offset += n;
	if (write_offset >= shm_ring_size) {
	    write_offset -= shm_ring_size;
	}
	bs->shm_ring->mc->write_offset = write_offset;
	bs->shm_ring->mc->written += n;
	sem_post(bs->shm_ring->sem_read);
    }

shm_failed:
    (*bs->fn)(bs->arg, buf, n);
}

static void
bsd_stream_read_to_shm_ring(
    void *s,
    void (*fn)(void *, void *, ssize_t),
    shm_ring_t *shm_ring,
    void *arg)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);
    auth_debug(1, _("bsd: stream_read_to_shm_ring: fd %d\n"), bs->fd);

    /*
     * Only one read request can be active per stream.
     */
    if (bs->ev_read != NULL)
        event_release(bs->ev_read);

    bs->r_callback.handle = bs->handle;
    bs->r_callback.s = bs;
    bs->r_callback.callback = bsd_stream_read_to_shm_ring_callback;

    bs->ev_read = event_create((event_id_t)bs->fd, EV_READFD, bsd_stream_read_to_shm_ring_callback, bs);
    bs->fn = fn;
    bs->arg = arg;
    bs->shm_ring = shm_ring;
    bs->ring_init = FALSE;
    event_activate(bs->ev_read);
}

/*
 * Cancel a previous stream read request.  It's ok if we didn't
 * have a read scheduled.
 */
static void
bsd_stream_read_cancel(
    void *	s)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);
    if (bs->ev_read != NULL) {
	event_release(bs->ev_read);
	bs->ev_read = NULL;
    }
}

/*
 * Pause a previous stream read request.  It's ok if we didn't
 * have a read scheduled.
 */
static void
bsd_stream_pause(
    void *	s)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);
    if (bs->paused) {
	return;
    }
    if (!bs->ev_read) {
	return;
    }
    bsd_stream_read_cancel(s);
    bs->paused = TRUE;
}

/*
 * Resume a previous stream read request.  It's ok if we didn't
 * have a read scheduled.
 */
static void
bsd_stream_resume(
    void *	s)
{
    struct sec_stream *bs = s;

    assert(bs != NULL);
    if (bs->ev_read) {
	return;
    }
    if (!bs->paused) {
	return;
    }

    bs->ev_read = event_create((event_id_t)bs->fd, EV_READFD, stream_read_callback, bs);
    event_activate(bs->ev_read);
    bs->paused = FALSE;
}

/*
 * Callback for bsd_stream_read
 */
static void
stream_read_callback(
    void *	arg)
{
    struct sec_stream *bs = arg;
    ssize_t n;

    assert(bs != NULL);

    do {
	n = read(bs->fd, bs->databuf, sizeof(bs->databuf));
    } while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));

    if (n <= 0)
	bsd_stream_read_cancel(bs);
    if (n < 0)
	security_stream_seterror(&bs->secstr, "%s", strerror(errno));

    (*bs->fn)(bs->arg, bs->databuf, n);
}