ggate/ggatessh/ggatessh.c

/*-
 * SPDX-License-Identifier: BSD-2-Clause-FreeBSD
 *
 * Copyright (c) 2004 Pawel Jakub Dawidek <pjd@FreeBSD.org>
 * All rights reserved.
 * Copyright 2020 John-Mark Gurney <jmg@FreeBSD.org>
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * $FreeBSD$
 */

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <libutil.h>
#include <paths.h>
#include <pthread.h>
#include <pthread_np.h>
#include <err.h>
#include <errno.h>
#include <assert.h>

#include <sys/param.h>
#include <sys/ioctl.h>
#include <sys/queue.h>
#include <sys/socket.h>
#include <sys/syslog.h>
#include <sys/bio.h>
#include <netdb.h>

#include <semaphore.h>

#include <libssh2.h>
#include <libssh2_sftp.h>

#include <geom/gate/g_gate.h>
#include "ggate.h"

static enum { UNSET, CREATE, DESTROY, LIST, RESCUE } action = UNSET;

struct ggs_connection {
	int			c_fd;
	int			*c_didwork;	/* allocated memory */
	LIBSSH2_SESSION		*c_session;
	LIBSSH2_SFTP		*c_sftp_session;
	LIBSSH2_SFTP_HANDLE	*c_handle;
};

static struct pidfh *pfh;

static const char *username;
static const char *hostname;
static const char *imgpath;
static const char *identityfile;
static const char *pubkeyfile;
static const char *sshport = "22";
static char *ggatessh_pidfile;
static int unit = G_GATE_UNIT_AUTO;
static unsigned flags = 0;
static int force = 0;
static unsigned queue_size = G_GATE_QUEUE_SIZE;
static off_t mediasize;
static unsigned sectorsize = 4096;
static unsigned timeout = G_GATE_TIMEOUT;
static int pushfd, popfd; /* work semaphore */
static pthread_t reqtd, proctd, mediatd;
static unsigned maxconnections = 32;
static struct ggs_connection start_conn;	/* only used once/first */

struct ggs_sess_cache {
	LIBSSH2_SESSION			*sc_ssh_session;
	LIBSSH2_SFTP			*sc_session;
	LIBSSH2_SFTP_HANDLE		*sc_handle;
	TAILQ_ENTRY(ggs_sess_cache)	sc_next;
};

struct ggs_req {
	struct g_gate_ctl_io	r_ggio;
#define r_ssh_session	r_sesscache->sc_ssh_session
#define r_session	r_sesscache->sc_session
#define r_handle	r_sesscache->sc_handle
	struct ggs_sess_cache	*r_sesscache;
	size_t			r_bufoff;
	int			r_didseek;
	TAILQ_ENTRY(ggs_req)	r_next;
};

static TAILQ_HEAD(ggs_reqqueue, ggs_req) procqueue =
	TAILQ_HEAD_INITIALIZER(procqueue);
static TAILQ_HEAD(ggs_sessqueue, ggs_sess_cache) session_cache =
	TAILQ_HEAD_INITIALIZER(session_cache);
static sem_t nconn_sem;
static pthread_mutex_t procqueue_mtx;

static void
usage(void)
{

	fprintf(stderr, "usage: %s create [-v] [-o <ro|wo|rw>] "
	    "[-F pidfile] [-i identifyfile] "
	    "[-l username] [-p port] "
	    "[-q queue_size] [-s sectorsize] [-r nrequests] "
	    "[-t timeout] [-u unit] <host> <path>\n", getprogname());
	fprintf(stderr, "       %s rescue [-v] [-o <ro|wo|rw>] "
	    "[-F pidfile] [-i identifyfile] "
	    "[-l username] [-p port] "
	    "[-r nrequests] <-u unit> <host> <path>\n", getprogname());
	fprintf(stderr, "       %s destroy [-f] <-u unit>\n", getprogname());
	fprintf(stderr, "       %s list [-v] [-u unit]\n", getprogname());
	exit(EXIT_FAILURE);
}

#if 0
static void
libssh2_debug_error(LIBSSH2_SESSION *session, const char *info)
{
	char *errmsg;
	int error;
	int rc;

	error = libssh2_session_last_errno(session);
	rc = libssh2_session_last_error(session, &errmsg, NULL, 0);

	g_gate_log(LOG_DEBUG, "%s: %s(%d)", info, errmsg, error);
}
#endif

static void
libssh2_errorx(LIBSSH2_SESSION *session, const char *info)
{
	char *errmsg;
	int rc;

	rc = libssh2_session_last_error(session, &errmsg, NULL, 0);
	g_gate_xlog("%s: %s", info, errmsg);
}

/*
 * Connect to the service (or port number) on host.
 *
 * Somewhat copied from freebsd/lib/libfetch/fetch_common.c:fetch_connect.
 */
static int
tcp_connect(const char *host, const char *service, int af)
{
	struct addrinfo hints, *sai, *sai0;
	int err, sd;

	hints = (struct addrinfo){
		.ai_family = af,
		.ai_socktype = SOCK_STREAM,
		.ai_flags = AI_ADDRCONFIG,
	};

	/* resolve server address */
	if (getaddrinfo(host, service, &hints, &sai0) == -1)
		return -1;

	if (sai0 == NULL) {
		errno = ENOENT;
		return -1;
	}

	sd = -1;
	err = -1;
	/* try each server address in turn */
	for (sai = sai0; sai != NULL; sai = sai->ai_next) {
		/* open socket */
		if ((sd = socket(sai->ai_family, sai->ai_socktype,
		    sai->ai_protocol)) == -1)
			break;

		/* attempt to connect to server address */
		if ((err = connect(sd, sai->ai_addr, sai->ai_addrlen)) == 0)
			break;

		/* clean up before next attempt */
		close(sd);
		sd = -1;
	}

	err = errno;

	fflush(stdout);

	freeaddrinfo(sai0);

	/* Fully close if it was opened; otherwise just don't leak the fd. */
	if (err == -1 && sd >= 0)
		close(sd);

	errno = err;

	return sd;
}

static int
get_open_flags()
{

	switch (flags) {
	case G_GATE_FLAG_READONLY:
		return LIBSSH2_FXF_READ;

	case G_GATE_FLAG_WRITEONLY:
		return LIBSSH2_FXF_WRITE;

	default:
		return LIBSSH2_FXF_READ|LIBSSH2_FXF_WRITE;
	}
}

static struct ggs_connection
make_connection(void)
{
	LIBSSH2_SESSION *session;
	LIBSSH2_SFTP *sftp_session;
	LIBSSH2_SFTP_HANDLE *handle;
	char *tmp;
	int *didworkp;
	int sockfd;
	int rc;

	sockfd = tcp_connect(hostname, sshport, 0);
	if (sockfd == -1) {
		if (errno == ENOENT)
			g_gate_xlog("tcp_connect: failed to lookup %s",
			    hostname);
		g_gate_xlog("tcp_connect: %s.", strerror(errno));
	}

	didworkp = malloc(sizeof *didworkp);
	if (didworkp == NULL)
		g_gate_xlog("malloc failed.");

	/* session = libssh2_session_init(); */
	session = libssh2_session_init_ex(NULL, NULL, NULL, didworkp);
	if (session == NULL)
		libssh2_errorx(session, "libssh2_session_init");

	if (g_gate_verbose)
		libssh2_trace(session, LIBSSH2_TRACE_SOCKET|LIBSSH2_TRACE_KEX|
		    LIBSSH2_TRACE_AUTH|LIBSSH2_TRACE_CONN|LIBSSH2_TRACE_SFTP|
		    LIBSSH2_TRACE_ERROR|LIBSSH2_TRACE_PUBLICKEY);

	/* XXX - libssh2_session_flag to enable compression */

	rc = libssh2_session_handshake(session, sockfd);
	if (rc)
		libssh2_errorx(session, "libssh2_session_handshake");

	libssh2_session_set_blocking(session, 1);

	/* XXX - known hosts handling */

	if (identityfile == NULL) {
		tmp = NULL;
		asprintf(&tmp, "%s/.ssh/id_rsa", getenv("HOME"));
		identityfile = tmp;
		tmp = NULL;
	}

	asprintf(&tmp, "%s.pub", identityfile);
	pubkeyfile = tmp;
	tmp = NULL;

	g_gate_log(LOG_DEBUG, "trying identity file: %s", identityfile);

	rc = libssh2_userauth_publickey_fromfile(session, username, pubkeyfile,
	    identityfile, NULL);
	//rc = libssh2_userauth_password(session, "freebsd", "freebsd");
	if (rc) {
		g_gate_log(LOG_ERR, "identity file: %s", identityfile);
		libssh2_errorx(session, "libssh2_userauth_publickey_fromfile");
	}

	/* always need at least one */
	sftp_session = libssh2_sftp_init(session);

	if (sftp_session == NULL)
		g_gate_xlog("libssh2_sftp_init");

	handle = libssh2_sftp_open(sftp_session, imgpath, get_open_flags(), 0);
	if (handle == NULL) {
		g_gate_log(LOG_ERR, "image file: %s", imgpath);
		libssh2_errorx(session, "libssh2_sftp_open");
	}

	return (struct ggs_connection){
		.c_fd = sockfd,
		.c_didwork = didworkp,
		.c_session = session,
		.c_sftp_session = sftp_session,
		.c_handle = handle,
	};
}

/*
 * Resize is required to run in a thread because the resize requires
 * that I/O is able to complete before it can return.
 */
static void *
mediachg(void *arg __unused)
{
	struct g_gate_ctl_modify ggiom;

	/* update mediasize, it may have changed */
	ggiom = (struct g_gate_ctl_modify){
		.gctl_version = G_GATE_VERSION,
		.gctl_unit = unit,
		.gctl_modify = GG_MODIFY_MEDIASIZE,
		.gctl_mediasize = mediasize,
	};
	g_gate_ioctl(G_GATE_CMD_MODIFY, &ggiom);
	g_gate_log(LOG_DEBUG, "updated ggate%d mediasize to %zd", unit,
	    mediasize);

	return NULL;
}

static void *
req_thread(void *arg __unused)
{
	struct ggs_req *greq;
	static char *buf;
	int buflen = 1024*1024;
	int error;

	g_gate_log(LOG_NOTICE, "%s: started!", __func__);

	greq = NULL;

	for (;;) {
		if (greq == NULL)
			greq = malloc(sizeof *greq);

		if (buf == NULL)
			buf = malloc(buflen);

		if (greq == NULL || buf == NULL) {
			/* XXX */
			g_gate_log(LOG_ERR, "Unable to allocate memory.");
			exit(1);
		}

		*greq = (struct ggs_req){
			.r_ggio = (struct g_gate_ctl_io){
				.gctl_version = G_GATE_VERSION,
				.gctl_unit = unit,
				.gctl_data = buf,
				.gctl_length = buflen,
				.gctl_error = 0,
			},
		};

		//g_gate_log(LOG_DEBUG, "waiting for ioctl");
		g_gate_ioctl(G_GATE_CMD_START, &greq->r_ggio);
		//g_gate_log(LOG_DEBUG, "got ioctl");

		error = greq->r_ggio.gctl_error;
		switch (error) {
		case 0:
			break;
		case ECANCELED:
			/* Exit gracefully. */
			g_gate_close_device();
			exit(EXIT_SUCCESS);
		case ENXIO:
		default:
			g_gate_xlog("ioctl(/dev/%s): %s.", G_GATE_CTL_NAME,
			    strerror(error));
		}

		g_gate_log(LOG_DEBUG, "ggio(%p), ver: %u, unit: %d, seq: %llu, "
		    "cmd: %u, offset: %llu, len: %llu", greq,
		    greq->r_ggio.gctl_version, greq->r_ggio.gctl_unit,
		    greq->r_ggio.gctl_seq, greq->r_ggio.gctl_cmd,
		    greq->r_ggio.gctl_offset, greq->r_ggio.gctl_length);

		switch (greq->r_ggio.gctl_cmd) {
		case BIO_READ:
			/* use a correctly sized allocation */
			greq->r_ggio.gctl_data =
			    malloc(greq->r_ggio.gctl_length);
			break;

		case BIO_WRITE:
			/* r_ggio takes ownership of buf now */
			buf = NULL;
			break;

		case BIO_DELETE:
		case BIO_FLUSH:
			greq->r_ggio.gctl_data = NULL;
			break;

		default:
			greq->r_ggio.gctl_error = EOPNOTSUPP;
			g_gate_ioctl(G_GATE_CMD_DONE, &greq->r_ggio);
			continue;	/* return EOPNOTSUPP */
			break;
		}

		//g_gate_log(LOG_DEBUG, "waiting for slot");
		sem_wait(&nconn_sem);
#if 0
		int semval;
		sem_getvalue(&nconn_sem, &semval);
		g_gate_log(LOG_DEBUG, "slots: %d", semval);
#endif

		error = pthread_mutex_lock(&procqueue_mtx);
		assert(error == 0);
		TAILQ_INSERT_TAIL(&procqueue, greq, r_next);
		error = pthread_mutex_unlock(&procqueue_mtx);
		assert(error == 0);

		/* notify processing thread a request is waiting */
		error = write(pushfd, "T", 1);
		if (error != 1)
			g_gate_xlog("write pushfd: %d, error: %s.", error,
			    strerror(error));

		/* pass ownership */
		greq = NULL;
	}
	g_gate_log(LOG_DEBUG, "%s: Died.", __func__);
	return (NULL);
}

static const char *
sftperrno_str(int err)
{
	const char *strs[] = {
		[0] = "ok",
		[1] = "eof",
		[2] = "no such file",
		[3] = "permission denied",
		[4] = "failure",
		[5] = "bad message",
		[6] = "no connection",
		[7] = "connection lost",
		[8] = "op unsupported",
	};

	if (err < 0 || err >= (int)nitems(strs))
		return "invalid errno";

	return strs[err];
}

static int
process_pending(struct ggs_reqqueue *req_pending,
    struct ggs_sessqueue *sessqueue)
{
	struct ggs_req *greq, *greq2;
	char *errmsg;
	int rc;
	int sftperrno;
	int didwork;

	didwork = 0;

	/* Work on each pending request */
	TAILQ_FOREACH_SAFE(greq, req_pending, r_next, greq2) {
again:
		switch (greq->r_ggio.gctl_cmd) {
		case BIO_READ:
			g_gate_log(LOG_DEBUG, "sftp_read(%p): %d(%d), rem: %d",
			    greq, greq->r_ggio.gctl_offset,
			    greq->r_ggio.gctl_length,
			    greq->r_ggio.gctl_length - greq->r_bufoff);
			if (greq->r_didseek == 0) {
				libssh2_sftp_seek64(greq->r_handle,
				    greq->r_ggio.gctl_offset);
				greq->r_didseek = 1;
			}
			rc = libssh2_sftp_read(greq->r_handle,
			    (char *)greq->r_ggio.gctl_data + greq->r_bufoff,
			    greq->r_ggio.gctl_length - greq->r_bufoff);
			g_gate_log(LOG_DEBUG, "sftp_read ret: %d", rc);
			if (rc < 0 && rc != LIBSSH2_ERROR_EAGAIN)
				g_gate_log(LOG_ERR, "libssh2_sftp_read");
			break;

		case BIO_WRITE:
			g_gate_log(LOG_DEBUG, "sftp_write(%p): %d(%d), rem: %d",
			    greq, greq->r_ggio.gctl_offset,
			    greq->r_ggio.gctl_length,
			    greq->r_ggio.gctl_length - greq->r_bufoff);
			if (greq->r_didseek == 0) {
				libssh2_sftp_seek64(greq->r_handle,
				    greq->r_ggio.gctl_offset);
				greq->r_didseek = 1;
			}
			rc = libssh2_sftp_write(greq->r_handle,
			    (char *)greq->r_ggio.gctl_data + greq->r_bufoff,
			    greq->r_ggio.gctl_length - greq->r_bufoff);
			g_gate_log(LOG_DEBUG, "sftp_write ret: %d", rc);
			if (rc < 0 && rc != LIBSSH2_ERROR_EAGAIN)
				libssh2_errorx(greq->r_ssh_session,
				    "libssh2_sftp_write");
			break;

		case BIO_FLUSH:
			g_gate_log(LOG_DEBUG, "sftp_flush(%p)", greq);
			rc = libssh2_sftp_fsync(greq->r_handle);

			didwork = 1;	/* assume this always does work */
			switch (rc) {
			case LIBSSH2_ERROR_SFTP_PROTOCOL:
				greq->r_ggio.gctl_error = EOPNOTSUPP;
				goto completeio;

			case LIBSSH2_ERROR_EAGAIN:
				continue;

			case 0:	/* success */
				goto completeio;

			default:
				libssh2_session_last_error(greq->r_ssh_session,
				    &errmsg, NULL, 0);
				g_gate_log(LOG_ERR, "sftp_flush(%p) ret %d: %s",
				    greq, rc, errmsg);
				greq->r_ggio.gctl_error = EIO;
				goto completeio;
			}
			/* NOTREACHABLE */
			break;

		case BIO_DELETE:
			g_gate_log(LOG_DEBUG, "sftp_punchhole(%p)", greq);
			rc = libssh2_sftp_punchhole(greq->r_handle,
			    greq->r_ggio.gctl_offset, greq->r_ggio.gctl_length);

			didwork = 1;	/* assume this always does work */
			switch (rc) {
			case LIBSSH2_ERROR_SFTP_PROTOCOL:
				greq->r_ggio.gctl_error = EOPNOTSUPP;
				sftperrno = libssh2_sftp_last_error(
				    greq->r_session);
				g_gate_log(LOG_DEBUG, "sftp_punchhole(%p) errno: %s(%d)", greq,
				    sftperrno_str(sftperrno), sftperrno);
				goto completeio;

			case LIBSSH2_ERROR_EAGAIN:
				continue;

			case 0:	/* success */
				goto completeio;

			default:
				libssh2_session_last_error(greq->r_ssh_session,
				    &errmsg, NULL, 0);
				g_gate_log(LOG_ERR, "sftp_punchhole(%p) ret %d: %s",
				    greq, rc, errmsg);
				greq->r_ggio.gctl_error = EIO;
				goto completeio;
			}
			/* NOTREACHABLE */
			break;

		default:
			rc = 0;
			g_gate_log(LOG_ERR, "unhandled op: %d",
			    greq->r_ggio.gctl_cmd);
			continue;
		}

		if (rc > 0) {
			didwork = 1;
			greq->r_bufoff += rc;

			/*
			 * try again on partial read/write,
			 * might have more data pending
			 */
			if ((off_t)greq->r_bufoff != greq->r_ggio.gctl_length)
				goto again;
		}

		if ((off_t)greq->r_bufoff == greq->r_ggio.gctl_length) {
			/* complete */
completeio:
			g_gate_log(LOG_DEBUG, "cmd complete: seq: %d, cmd: %d",
			    greq->r_ggio.gctl_seq, greq->r_ggio.gctl_cmd);
			g_gate_ioctl(G_GATE_CMD_DONE, &greq->r_ggio);
			TAILQ_REMOVE(req_pending, greq, r_next);
			TAILQ_INSERT_HEAD(sessqueue, greq->r_sesscache,
			    sc_next);
			free(greq->r_ggio.gctl_data);
			free(greq);

			/* release this slot */
			sem_post(&nconn_sem);
		}
	}

	return didwork;
}

/*
 * XXX - expose this, this is to try to silence the loop that can
 * happen when there is data waiting (oob or window info), but no
 * outstanding requests are pending.
 */
int _libssh2_transport_read(LIBSSH2_SESSION *session);

/*
 * libssh2 does not have a good way to handle detection when it's
 * truly time to sleep.  Writing is an easy case, as if there's space
 * to write, and the _BLOCK_OUTBOUND flag is set, select will return
 * and the data will get processed.
 *
 * In the case of reading, it is more complicated, as a later request
 * could read data for a previous request, and there is no known way
 * to know when this is done.  The solution I came up with is to wrap
 * the recv function, and continue to iterate through the pending
 * requests until no more data is read.
 */

static ssize_t
ggatessh_recv_libssh2_hack(libssh2_socket_t fd, void *buf,
    size_t len, int recv_flags, void **abstract)
{
	int *didworkp = *abstract;
	ssize_t ret;

	ret = recv(fd, buf, len, recv_flags);

	if (ret > 0)
		*didworkp = 1;

	if (ret == -1 && errno == EAGAIN)
		return -EAGAIN;

	return ret;
}

/*
 * sftp session management is a bit tricky.
 * if there is an entry in sessioncache, use that one.
 * if we are waiting for a new session (gsc_pend != NULL),
 *  establish session, then open handle
 * when the new session completes, process the work queue
 */
static void *
proc_thread(void *arg __unused)
{
	char scratch[32];
	struct ggs_reqqueue req_pending;
	struct timeval to;
	struct ggs_sess_cache *gsc, *gsc_pending;
	struct ggs_req *greq;
	LIBSSH2_SESSION *session;
	int *didworkp;	/* was any reads done, rescan work */
	fd_set fdread;
	fd_set fdwrite;
	fd_set fdexcep;
	int sockfd;
	int maxfd;
	int error;
	int dir;
	int rc;

	g_gate_log(LOG_NOTICE, "%s: started!", __func__);

	TAILQ_INIT(&req_pending);

	/* make sure we don't block on reading */
	fcntl(popfd, F_SETFL, O_NONBLOCK);

	sockfd = start_conn.c_fd;
	didworkp = start_conn.c_didwork;
	session = start_conn.c_session;

	gsc = malloc(sizeof *gsc);
	*gsc = (struct ggs_sess_cache){
		.sc_ssh_session = start_conn.c_session,
		.sc_session = start_conn.c_sftp_session,
		.sc_handle = start_conn.c_handle,
	};

	TAILQ_INSERT_HEAD(&session_cache, gsc, sc_next);
	gsc = NULL;
	gsc_pending = NULL;

	*didworkp = 0;

	libssh2_session_set_blocking(session, 0);
	libssh2_session_callback_set(session, LIBSSH2_CALLBACK_RECV,
	    ggatessh_recv_libssh2_hack);

	for (;;) {
		//g_gate_log(LOG_DEBUG, "looping");

		if (!*didworkp) {
			/* setup polling loop */
			maxfd = -1;
			FD_ZERO(&fdread);
			FD_ZERO(&fdwrite);
			FD_ZERO(&fdexcep);

			dir = libssh2_session_block_directions(session);
			if (dir & LIBSSH2_SESSION_BLOCK_INBOUND ||
			    gsc_pending != NULL)
				FD_SET(sockfd, &fdread);
			if (dir & LIBSSH2_SESSION_BLOCK_OUTBOUND)
				FD_SET(sockfd, &fdwrite);

			/* add in the pop descriptor */
			FD_SET(popfd, &fdread);
			maxfd = MAX(popfd, sockfd);

			g_gate_log(LOG_DEBUG, "selecting: %s %s, " \
			    "read: sockfd: %d, popfd: %d, write: sockfd: %d",
			    (dir & LIBSSH2_SESSION_BLOCK_INBOUND) ? "inbound" :
			    "", (dir & LIBSSH2_SESSION_BLOCK_OUTBOUND) ?
			    "outbound" : "", FD_ISSET(sockfd, &fdread),
			    FD_ISSET(popfd, &fdread),
			    FD_ISSET(sockfd, &fdwrite));
			to = (struct timeval){ .tv_sec = 1, .tv_usec = 1000 };
			(void)to;
			rc = select(maxfd + 1, &fdread, &fdwrite, &fdexcep,
			    NULL);
			switch (rc) {
			case -1:
				g_gate_log(LOG_ERR, "%s: select failed: %s",
				    __func__, strerror(errno));
				break;
			case 0:
			default:
				g_gate_log(LOG_DEBUG, "select: %d, " \
				    "read: sockfd: %d, popfd: %d, " \
				    "write: sockfd: %d", rc,
				    FD_ISSET(sockfd, &fdread),
				    FD_ISSET(popfd, &fdread),
				    FD_ISSET(sockfd, &fdwrite));
				break;
			}
		}

		_libssh2_transport_read(session);

		*didworkp = 0;

		/* process pending, so any completed can be reused */
		process_pending(&req_pending, &session_cache);

		if (FD_ISSET(popfd, &fdread)) {
			/* read off the tokens */
			g_gate_log(LOG_DEBUG, "popping");
			read(popfd, scratch, sizeof scratch);

			for (;;) {
procreq:
				/* get the request */
				error = pthread_mutex_lock(&procqueue_mtx);
				assert(error == 0);
				greq = TAILQ_FIRST(&procqueue);
				g_gate_log(LOG_DEBUG, "greq: %p", greq);
				if (greq != NULL)
					TAILQ_REMOVE(&procqueue, greq, r_next);
				error = pthread_mutex_unlock(&procqueue_mtx);
				assert(error == 0);

				/* no more to process */
				if (greq == NULL)
					break;

				gsc = TAILQ_FIRST(&session_cache);
				if (gsc == NULL) {
					if (gsc_pending == NULL) {
						/* need new session */
						g_gate_log(LOG_DEBUG,
						    "need new session");
						gsc_pending =
						    malloc(sizeof *gsc);
						gsc_pending->sc_ssh_session =
						    session;
						gsc_pending->sc_session = NULL;
						gsc_pending->sc_handle = NULL;
					}

					/* put back request */
					error =
					    pthread_mutex_lock(&procqueue_mtx);
					assert(error == 0);
					TAILQ_INSERT_HEAD(&procqueue, greq,
					    r_next);
					error = pthread_mutex_unlock(
					    &procqueue_mtx);
					assert(error == 0);

					break;
				} else {
					/* process request */
					TAILQ_REMOVE(&session_cache, gsc,
					    sc_next);
					greq->r_sesscache = gsc;
					gsc = NULL;

					greq->r_bufoff = 0;

					TAILQ_INSERT_TAIL(&req_pending, greq,
					    r_next);

					greq = NULL;
				}
			}
		}

		if (gsc_pending != NULL) {
			/* we are creating a new session */
			if (gsc_pending->sc_session == NULL) {
				*didworkp = 1;
				gsc_pending->sc_session =
				    libssh2_sftp_init(session);
			}

			if (gsc_pending->sc_session != NULL) {
				*didworkp = 1;
				gsc_pending->sc_handle = libssh2_sftp_open(
				    gsc_pending->sc_session, imgpath,
				    get_open_flags(), 0);

				if (gsc_pending->sc_handle == NULL) {
					error = libssh2_session_last_errno(
					    session);
					if (error != LIBSSH2_ERROR_EAGAIN)
						libssh2_errorx(session,
						    "sftp_open");
				}
			}

			g_gate_log(LOG_DEBUG,
			    "pending: session: %p, handle: %p",
			    gsc_pending->sc_session, gsc_pending->sc_handle);

			/* we have a fully initalized entry, use it */
			if (gsc_pending->sc_handle != NULL) {
				g_gate_log(LOG_DEBUG, "new session created");
				TAILQ_INSERT_HEAD(&session_cache, gsc_pending,
				    sc_next);
				gsc_pending = NULL;
				*didworkp = 1;
				goto procreq;
			}
		}

		/* kick of any queued requests from above */
		process_pending(&req_pending, &session_cache);
	}

	g_gate_log(LOG_DEBUG, "%s: Died.", __func__);
	pthread_exit(NULL);
}

static void
ggatessh_makepidfile(void)
{
	pid_t otherpid;

	if (!g_gate_verbose) {
		if (ggatessh_pidfile == NULL) {
			asprintf(&ggatessh_pidfile,
			    _PATH_VARRUN "/ggatessh.ggate%d.pid", unit);
			if (ggatessh_pidfile == NULL)
				err(EXIT_FAILURE,
				    "Cannot allocate memory for pidfile");
		}
		pfh = pidfile_open(ggatessh_pidfile, 0600, &otherpid);
		if (pfh == NULL) {
			if (errno == EEXIST) {
				errx(EXIT_FAILURE,
				    "Daemon already running, pid: %jd.",
				    (intmax_t)otherpid);
			}
			err(EXIT_FAILURE, "Cannot open/create pidfile");
		}
	}
}

static void
mydaemon(void)
{

	if (g_gate_verbose > 0)
		return;
	if (daemon(0, 0) == 0)
		return;
	if (action == CREATE)
		g_gate_destroy(unit, 1);
	err(EXIT_FAILURE, "Cannot daemonize");
}

static int
g_gatessh_connect(void)
{
	struct ggs_connection conn;
	LIBSSH2_SFTP_ATTRIBUTES attrs;
	int rc;

	/* get the remote's size */
	conn = make_connection();

	rc = libssh2_sftp_fstat(conn.c_handle, &attrs);

	/* only allow regular and char devices */
	if (!(LIBSSH2_SFTP_S_ISREG(attrs.flags) || 
	    !LIBSSH2_SFTP_S_ISCHR(attrs.flags))) {
		g_gate_xlog("remote file not a regular file");
	}

	mediasize = attrs.filesize;
	g_gate_log(LOG_DEBUG, "got mediasize: %zd", mediasize);

	start_conn = conn;	/* cache to use later */

	return 1;
}

static void
g_gatessh_start(void)
{
	int filedes[2];
	int error;

	pipe(filedes);
	pushfd = filedes[1];
	popfd = filedes[0];

	error = pthread_mutex_init(&procqueue_mtx, NULL);
	if (error != 0) {
		g_gate_xlog("pthread_mutex_init(inqueue_mtx): %s.",
		    strerror(error));
	}

	sem_init(&nconn_sem, 0, maxconnections);

	error = pthread_create(&proctd, NULL, proc_thread, NULL);
	if (error != 0) {
		g_gate_destroy(unit, 1); /* XXX - remove */
		g_gate_xlog("pthread_create(proc_thread): %s.",
		    strerror(error));
	}
	pthread_set_name_np(proctd, "proc");

	reqtd = pthread_self();
	pthread_set_name_np(reqtd, "req");
	req_thread(NULL);

	/* Disconnected. */
	close(pushfd);
	close(popfd);
}

static void
signop(int sig __unused)
{

	/* Do nothing. */
}

static void
g_gatessh_loop(void)
{
	struct g_gate_ctl_cancel ggioc;

	signal(SIGUSR1, signop);
	for (;;) {
		g_gatessh_start();
		g_gate_log(LOG_NOTICE, "Disconnected [%s@%s:%s]. Connecting...",
		    username, hostname, imgpath);

		ggioc.gctl_version = G_GATE_VERSION;
		ggioc.gctl_unit = unit;
		ggioc.gctl_seq = 0;
		g_gate_ioctl(G_GATE_CMD_CANCEL, &ggioc);
	}
}

static void
g_gatessh_create(void)
{
	struct g_gate_ctl_create ggioc;

	if (!g_gatessh_connect())
		g_gate_xlog("Cannot connect: %s.", strerror(errno));

	/*
	 * Ok, got both sockets, time to create provider.
	 */
	memset(&ggioc, 0, sizeof(ggioc));
	ggioc.gctl_version = G_GATE_VERSION;
	ggioc.gctl_mediasize = mediasize;
	ggioc.gctl_sectorsize = sectorsize;
	ggioc.gctl_flags = flags;
	ggioc.gctl_maxcount = queue_size;
	ggioc.gctl_timeout = timeout;
	ggioc.gctl_unit = unit;
	snprintf(ggioc.gctl_info, sizeof(ggioc.gctl_info), "%s@%s:%s",
	    username, hostname, imgpath);
	g_gate_ioctl(G_GATE_CMD_CREATE, &ggioc);
	if (unit == -1) {
		printf("%s%u\n", G_GATE_PROVIDER_NAME, ggioc.gctl_unit);
		fflush(stdout);
	}
	unit = ggioc.gctl_unit;

	ggatessh_makepidfile();

	mydaemon();
	if (pfh != NULL)
		pidfile_write(pfh);
	g_gatessh_loop();
}

static void
g_gatessh_rescue(void)
{
	struct g_gate_ctl_cancel ggioc;
	int error;

	if (!g_gatessh_connect())
		g_gate_xlog("Cannot connect: %s.", strerror(errno));

	ggioc = (struct g_gate_ctl_cancel){
		.gctl_version = G_GATE_VERSION,
		.gctl_unit = unit,
		.gctl_seq = 0,
	};
	g_gate_ioctl(G_GATE_CMD_CANCEL, &ggioc);

	ggatessh_makepidfile();

	mydaemon();
	pidfile_write(pfh);

	error = pthread_create(&mediatd, NULL, mediachg, NULL);
	if (error != 0)
		g_gate_xlog("unable to create mediasize change thread",
		    strerror(errno));

	g_gatessh_loop();
}

/*
 * handle two methods of specifying things.
 * if only one arg, in the future split it how ssh does:
 * [user[:password]@]<host>:<img>
 * and URI:
 * sftp://[user[:password]@]<host>[:<port>]/<img>
 *
 * If both are specified, it's the same as above, but split
 * on the :.
 */
static void
handle_params(int argc, char *argv[])
{

	if (username == NULL) {
		username = getenv("USER");
		if (username == NULL) {
			err(EXIT_FAILURE,
			    "USER environment variable not present, set, "
			    "or specify via -l argument.");
		}
	}

	if (argc != 2)
		usage();

	hostname = argv[0];
	imgpath = argv[1];
}

int
main(int argc, char *argv[])
{
	int rc;

	if (argc < 2)
		usage();
	if (strcasecmp(argv[1], "create") == 0)
		action = CREATE;
	else if (strcasecmp(argv[1], "destroy") == 0)
		action = DESTROY;
	else if (strcasecmp(argv[1], "list") == 0)
		action = LIST;
	else if (strcasecmp(argv[1], "rescue") == 0)
		action = RESCUE;
	else
		usage();

	argc -= 1;
	argv += 1;
	for (;;) {
		int ch;

		ch = getopt(argc, argv, "fF:i:l:o:p:q:r:s:t:u:v");
		if (ch == -1)
			break;
		switch (ch) {
		case 'f':
			if (action != DESTROY)
				usage();
			force = 1;
			break;

		case 'F':
			ggatessh_pidfile = optarg;
			break;

		case 'i':
			identityfile = optarg;
			break;

		case 'l':
			username = optarg;
			break;

		case 'o':
			if (action != CREATE && action != RESCUE)
				usage();
			if (strcasecmp("ro", optarg) == 0)
				flags = G_GATE_FLAG_READONLY;
			else if (strcasecmp("wo", optarg) == 0)
				flags = G_GATE_FLAG_WRITEONLY;
			else if (strcasecmp("rw", optarg) == 0)
				flags = 0;
			else {
				errx(EXIT_FAILURE,
				    "Invalid argument for '-o' option.");
			}
			break;

		case 'p':
			sshport = optarg;
			break;

		case 'q':
			if (action != CREATE)
				usage();
			errno = 0;
			queue_size = strtoul(optarg, NULL, 10);
			if (queue_size == 0 && errno != 0)
				errx(EXIT_FAILURE, "Invalid queue_size.");
			break;
		case 'r':
			if (action != CREATE && action != RESCUE)
				usage();
			errno = 0;
			maxconnections = strtoul(optarg, NULL, 10);
			if (maxconnections == 0 && errno != 0)
				errx(EXIT_FAILURE, "Invalid queue_size.");
			break;
		case 's':
			if (action != CREATE)
				usage();
			errno = 0;
			sectorsize = strtoul(optarg, NULL, 10);
			if (sectorsize == 0 && errno != 0)
				errx(EXIT_FAILURE, "Invalid sectorsize.");
			break;
		case 't':
			if (action != CREATE)
				usage();
			errno = 0;
			timeout = strtoul(optarg, NULL, 10);
			if (timeout == 0 && errno != 0)
				errx(EXIT_FAILURE, "Invalid timeout.");
			break;
		case 'u':
			errno = 0;
			unit = strtol(optarg, NULL, 10);
			if (unit == 0 && errno != 0)
				errx(EXIT_FAILURE, "Invalid unit number.");
			break;
		case 'v':
			if (action == DESTROY)
				usage();
			g_gate_verbose++;
			break;
		default:
			usage();
		}
	}
	argc -= optind;
	argv += optind;

	g_gate_log(LOG_DEBUG, "libssh2_init");
	rc = libssh2_init(0);
	if (rc != 0) {
		fprintf(stderr, "libssh2 initialization failed (%d)\n", rc);
		return 1;
	}

	switch (action) {
	case CREATE:
		if (argc < 1 || argc > 2)
			usage();
		handle_params(argc, argv);
		g_gate_load_module();
		g_gate_open_device();
		g_gatessh_create();
		break;
	case DESTROY:
		if (argc != 0)
			usage();
		if (unit == -1) {
			fprintf(stderr, "Required unit number.\n");
			usage();
		}
		g_gate_verbose = 1;
		g_gate_open_device();
		g_gate_destroy(unit, force);
		break;
	case LIST:
		g_gate_list(unit, g_gate_verbose);
		break;
	case RESCUE:
		if (argc < 1 || argc > 2)
			usage();
		if (unit == -1) {
			fprintf(stderr, "Required unit number.\n");
			usage();
		}
		handle_params(argc, argv);
		g_gate_open_device();
		g_gatessh_rescue();
		break;
	case UNSET:
	default:
		usage();
	}
	g_gate_close_device();
	exit(EXIT_SUCCESS);
}