ssh(1) multiplexing rewrite

Damien Miller djm at mindrot.org
Mon Jan 18 15:26:07 EST 2010


On Thu, 14 Jan 2010, Salvador Fandino wrote:

> Damien Miller wrote:
> > Hi,
> > 
> > At the n2k10 OpenBSD network hackathon, I finally got some time to clean
> > up and rewrite the ssh(1) client multiplexing code. The attached diffs
> > (one for portable OpenSSH, one for OpenBSD) are the result, and they
> > need some testing.
> 
> I have repeatedly run the test suite for my Perl module Net::OpenSSH that
> (ab)uses the multiplexing feature without errors.
> 
> And, while you are at it, I have some feature requests:
> 
> 1) add support for sending signals to the remote processes via mux control
> commands. The bug tracker contains a patch by Darren Tucker implementing the
> signal part of the SSH protocol and later I submitted another patch (now
> obsoleted by your changes) to request sending the signals via mux control
> commands (https://bugzilla.mindrot.org/show_bug.cgi?id=1424).

If/when Darren's diff goes in then adding this should be pretty easy; the
only problem I can think of is identifying which session to signal. I
think we would need something like your friendly-name extension to
sort it out, and that implies adding the ability to query a running mux
master to discover what sessions are open and their names.

This would be post-OpenSSH 5.4, so please file an enhancement request so it
doesn't get lost.

> 2) add support for the new netcat-like feature over mux.

Yes, this was planned and now implemented. See the attached diff; OpenBSD
only for now, I'll do the portability bits when it goes in.

> 3) allow to run the mux server over SSH stdin instead of over a named
> Unix socket. That would be useful to embed ssh inside another program.
>
> For instance, Net::OpenSSH internally starts a new ssh in master mode
> and then sends commands to the remote machine through the mux socket
> running slave ssh processes, one per command.
>
> Handling the named Unix socket is a nuisance because it means
> accessing the file system, looking for a proper location to place the
> socket, checking that permissions are right, avoiding collisions with
> other instances of the module concurrently running and cleaning up.
>
> To add this feature to OpenSSH, besides allowing attaching the mux
> server to stdio, the protocol should also be modified in order to
> allow interleaving requests and responses related to different
> channels over the same mux stream, for instance including some session
> ID.

Yes, that sounds reasonable. I'm not afraid to make incompatible changes
to this protocol and adding interleaved requests isn't so hard now.

Please file a bug for this too, along with any specific suggestions for
how you think the protocol should work (since you have already put more
thought into it than I :)

> It would also require making the mux protocol public so that it could be
> implemented by third party clients.

That's why PROTOCOL.mux exists :)

-d
-------------- next part --------------
Index: channels.c
===================================================================
RCS file: /cvs/src/usr.bin/ssh/channels.c,v
retrieving revision 1.301
diff -u -p -r1.301 channels.c
--- channels.c	11 Jan 2010 01:39:46 -0000	1.301
+++ channels.c	18 Jan 2010 04:20:34 -0000
@@ -235,7 +235,6 @@ channel_register_fds(Channel *c, int rfd
 	c->rfd = rfd;
 	c->wfd = wfd;
 	c->sock = (rfd == wfd) ? rfd : -1;
-	c->ctl_fd = -1; /* XXX: set elsewhere */
 	c->efd = efd;
 	c->extended_usage = extusage;
 
@@ -323,6 +322,9 @@ channel_new(char *ctype, int type, int r
 	c->output_filter = NULL;
 	c->filter_ctx = NULL;
 	c->filter_cleanup = NULL;
+	c->ctl_chan = -1;
+	c->mux_rcb = NULL;
+	c->mux_ctx = NULL;
 	c->delayed = 1;		/* prevent call to channel_post handler */
 	TAILQ_INIT(&c->status_confirms);
 	debug("channel %d: new [%s]", found, remote_name);
@@ -365,11 +367,10 @@ channel_close_fd(int *fdp)
 static void
 channel_close_fds(Channel *c)
 {
-	debug3("channel %d: close_fds r %d w %d e %d c %d",
-	    c->self, c->rfd, c->wfd, c->efd, c->ctl_fd);
+	debug3("channel %d: close_fds r %d w %d e %d",
+	    c->self, c->rfd, c->wfd, c->efd);
 
 	channel_close_fd(&c->sock);
-	channel_close_fd(&c->ctl_fd);
 	channel_close_fd(&c->rfd);
 	channel_close_fd(&c->wfd);
 	channel_close_fd(&c->efd);
@@ -395,8 +396,6 @@ channel_free(Channel *c)
 
 	if (c->sock != -1)
 		shutdown(c->sock, SHUT_RDWR);
-	if (c->ctl_fd != -1)
-		shutdown(c->ctl_fd, SHUT_RDWR);
 	channel_close_fds(c);
 	buffer_free(&c->input);
 	buffer_free(&c->output);
@@ -518,6 +517,7 @@ channel_still_open(void)
 		case SSH_CHANNEL_X11_LISTENER:
 		case SSH_CHANNEL_PORT_LISTENER:
 		case SSH_CHANNEL_RPORT_LISTENER:
+		case SSH_CHANNEL_MUX_LISTENER:
 		case SSH_CHANNEL_CLOSED:
 		case SSH_CHANNEL_AUTH_SOCKET:
 		case SSH_CHANNEL_DYNAMIC:
@@ -531,6 +531,7 @@ channel_still_open(void)
 		case SSH_CHANNEL_OPENING:
 		case SSH_CHANNEL_OPEN:
 		case SSH_CHANNEL_X11_OPEN:
+		case SSH_CHANNEL_MUX_CLIENT:
 			return 1;
 		case SSH_CHANNEL_INPUT_DRAINING:
 		case SSH_CHANNEL_OUTPUT_DRAINING:
@@ -562,6 +563,8 @@ channel_find_open(void)
 		case SSH_CHANNEL_X11_LISTENER:
 		case SSH_CHANNEL_PORT_LISTENER:
 		case SSH_CHANNEL_RPORT_LISTENER:
+		case SSH_CHANNEL_MUX_LISTENER:
+		case SSH_CHANNEL_MUX_CLIENT:
 		case SSH_CHANNEL_OPENING:
 		case SSH_CHANNEL_CONNECTING:
 		case SSH_CHANNEL_ZOMBIE:
@@ -612,6 +615,8 @@ channel_open_message(void)
 		case SSH_CHANNEL_CLOSED:
 		case SSH_CHANNEL_AUTH_SOCKET:
 		case SSH_CHANNEL_ZOMBIE:
+		case SSH_CHANNEL_MUX_CLIENT:
+		case SSH_CHANNEL_MUX_LISTENER:
 			continue;
 		case SSH_CHANNEL_LARVAL:
 		case SSH_CHANNEL_OPENING:
@@ -622,12 +627,12 @@ channel_open_message(void)
 		case SSH_CHANNEL_INPUT_DRAINING:
 		case SSH_CHANNEL_OUTPUT_DRAINING:
 			snprintf(buf, sizeof buf,
-			    "  #%d %.300s (t%d r%d i%d/%d o%d/%d fd %d/%d cfd %d)\r\n",
+			    "  #%d %.300s (t%d r%d i%d/%d o%d/%d fd %d/%d cc %d)\r\n",
 			    c->self, c->remote_name,
 			    c->type, c->remote_id,
 			    c->istate, buffer_len(&c->input),
 			    c->ostate, buffer_len(&c->output),
-			    c->rfd, c->wfd, c->ctl_fd);
+			    c->rfd, c->wfd, c->ctl_chan);
 			buffer_append(&buffer, buf, strlen(buf));
 			continue;
 		default:
@@ -834,9 +839,6 @@ channel_pre_open(Channel *c, fd_set *rea
 			FD_SET(c->efd, readset);
 	}
 	/* XXX: What about efd? races? */
-	if (compat20 && c->ctl_fd != -1 &&
-	    c->istate == CHAN_INPUT_OPEN && c->ostate == CHAN_OUTPUT_OPEN)
-		FD_SET(c->ctl_fd, readset);
 }
 
 /* ARGSUSED */
@@ -981,6 +983,28 @@ channel_pre_x11_open(Channel *c, fd_set 
 	}
 }
 
+static void
+channel_pre_mux_client(Channel *c, fd_set *readset, fd_set *writeset)
+{
+	if (c->istate == CHAN_INPUT_OPEN &&
+	    buffer_check_alloc(&c->input, CHAN_RBUF))
+		FD_SET(c->rfd, readset);
+	if (c->istate == CHAN_INPUT_WAIT_DRAIN) {
+		/* clear buffer immediately (discard any partial packet) */
+		buffer_clear(&c->input);
+		chan_ibuf_empty(c);
+		/* Start output drain. XXX just kill chan? */
+		chan_rcvd_oclose(c);
+	}
+	if (c->ostate == CHAN_OUTPUT_OPEN ||
+	    c->ostate == CHAN_OUTPUT_WAIT_DRAIN) {
+		if (buffer_len(&c->output) > 0)
+			FD_SET(c->wfd, writeset);
+		else if (c->ostate == CHAN_OUTPUT_WAIT_DRAIN)
+			chan_obuf_empty(c);
+	}
+}
+
 /* try to decode a socks4 header */
 /* ARGSUSED */
 static int
@@ -1213,19 +1237,14 @@ channel_decode_socks5(Channel *c, fd_set
 }
 
 Channel *
-channel_connect_stdio_fwd(const char *host_to_connect, u_short port_to_connect)
+channel_connect_stdio_fwd(const char *host_to_connect, u_short port_to_connect,
+    int in, int out)
 {
 	Channel *c;
-	int in, out;
 
 	debug("channel_connect_stdio_fwd %s:%d", host_to_connect,
 	    port_to_connect);
 
-	in = dup(STDIN_FILENO);
-	out = dup(STDOUT_FILENO);
-	if (in < 0 || out < 0)
-		fatal("channel_connect_stdio_fwd: dup() in/out failed");
-
 	c = channel_new("stdio-forward", SSH_CHANNEL_OPENING, in, out,
 	    -1, CHAN_TCP_WINDOW_DEFAULT, CHAN_TCP_PACKET_DEFAULT,
 	    0, "stdio-forward", /*nonblock*/0);
@@ -1728,34 +1747,6 @@ channel_handle_efd(Channel *c, fd_set *r
 
 /* ARGSUSED */
 static int
-channel_handle_ctl(Channel *c, fd_set *readset, fd_set *writeset)
-{
-	char buf[16];
-	int len;
-
-	/* Monitor control fd to detect if the slave client exits */
-	if (c->ctl_fd != -1 && FD_ISSET(c->ctl_fd, readset)) {
-		len = read(c->ctl_fd, buf, sizeof(buf));
-		if (len < 0 && (errno == EINTR || errno == EAGAIN))
-			return 1;
-		if (len <= 0) {
-			debug2("channel %d: ctl read<=0", c->self);
-			if (c->type != SSH_CHANNEL_OPEN) {
-				debug2("channel %d: not open", c->self);
-				chan_mark_dead(c);
-				return -1;
-			} else {
-				chan_read_failed(c);
-				chan_write_failed(c);
-			}
-			return -1;
-		} else
-			fatal("%s: unexpected data on ctl fd", __func__);
-	}
-	return 1;
-}
-
-static int
 channel_check_window(Channel *c)
 {
 	if (c->type == SSH_CHANNEL_OPEN &&
@@ -1785,10 +1776,130 @@ channel_post_open(Channel *c, fd_set *re
 	if (!compat20)
 		return;
 	channel_handle_efd(c, readset, writeset);
-	channel_handle_ctl(c, readset, writeset);
 	channel_check_window(c);
 }
 
+static u_int
+read_mux(Channel *c, u_int need)
+{
+	char buf[CHAN_RBUF];
+	int len;
+	u_int rlen;
+
+/*	debug3("%s: channel %d: entering, need %u have %u",
+	    __func__, c->self, need, buffer_len(&c->input)); */
+	if (buffer_len(&c->input) < need) {
+		rlen = need - buffer_len(&c->input);
+		len = read(c->rfd, buf, MIN(rlen, CHAN_RBUF));
+		if (len <= 0) {
+			if (errno != EINTR && errno != EAGAIN) {
+				debug2("channel %d: ctl read<=0 rfd %d len %d",
+				    c->self, c->rfd, len);
+				chan_read_failed(c);
+				return 0;
+			}
+		} else
+			buffer_append(&c->input, buf, len);
+	}
+/*	debug3("%s: channel %d: done, need %u have %u",
+	    __func__, c->self, need, buffer_len(&c->input)); */
+	return buffer_len(&c->input);
+}
+
+static void
+channel_post_mux_client(Channel *c, fd_set *readset, fd_set *writeset)
+{
+	u_int need;
+	ssize_t len;
+
+	if (!compat20)
+		fatal("%s: entered with !compat20", __func__);
+
+	if (c->rfd != -1 && FD_ISSET(c->rfd, readset) &&
+	    (c->istate == CHAN_INPUT_OPEN ||
+	    c->istate == CHAN_INPUT_WAIT_DRAIN)) {
+		/*
+		 * Don't not read past the precise end of packets to
+		 * avoid disrupting fd passing.
+		 */
+		if (read_mux(c, 4) < 4) /* read header */
+			return;
+		need = get_u32(buffer_ptr(&c->input));
+#define CHANNEL_MUX_MAX_PACKET	(256 * 1024)
+		if (need > CHANNEL_MUX_MAX_PACKET) {
+			debug2("channel %d: packet too big %u > %u",
+			    c->self, CHANNEL_MUX_MAX_PACKET, need);
+			chan_rcvd_oclose(c);
+			return;
+		}
+		if (read_mux(c, need + 4) < need + 4) /* read body */
+			return;
+		c->mux_rcb(c, c->mux_ctx);
+	}
+
+	if (c->wfd != -1 && FD_ISSET(c->wfd, writeset) &&
+	    buffer_len(&c->output) > 0) {
+		len = write(c->wfd, buffer_ptr(&c->output),
+		    buffer_len(&c->output));
+		if (len < 0 && (errno == EINTR || errno == EAGAIN))
+			return;
+		if (len <= 0) {
+			chan_mark_dead(c);
+			return;
+		}
+		buffer_consume(&c->output, len);
+	}
+}
+
+static void
+channel_post_mux_listener(Channel *c, fd_set *readset, fd_set *writeset)
+{
+	Channel *nc;
+	struct sockaddr_storage addr;
+	socklen_t addrlen;
+	int newsock;
+	uid_t euid;
+	gid_t egid;
+
+	if (FD_ISSET(c->sock, readset)) {
+		debug("multiplexing control connection");
+
+		/*
+		 * Accept connection on control socket
+		 */
+		memset(&addr, 0, sizeof(addr));
+		addrlen = sizeof(addr);
+		if ((newsock = accept(c->sock, (struct sockaddr*)&addr,
+		    &addrlen)) == -1) {
+			error("%s accept: %s", __func__, strerror(errno));
+			return;
+		}
+	
+		if (getpeereid(newsock, &euid, &egid) < 0) {
+			error("%s getpeereid failed: %s", __func__,
+			    strerror(errno));
+			close(newsock);
+			return;
+		}
+		if ((euid != 0) && (getuid() != euid)) {
+			error("multiplex uid mismatch: peer euid %u != uid %u",
+			    (u_int)euid, (u_int)getuid());
+			close(newsock);
+			return;
+		}
+		nc = channel_new("multiplex client", SSH_CHANNEL_MUX_CLIENT,
+		    newsock, newsock, -1, c->local_window_max,
+		    c->local_maxpacket, 0, "mux-control", 1);
+		nc->mux_rcb = c->mux_rcb;
+		debug3("%s: new mux channel %d fd %d", __func__,
+		    nc->self, nc->sock);
+		/* establish state */
+		nc->mux_rcb(nc, NULL);
+		/* mux state transitions must not elicit protocol messages */
+		nc->flags |= CHAN_LOCAL;
+	}
+}
+
 /* ARGSUSED */
 static void
 channel_post_output_drain_13(Channel *c, fd_set *readset, fd_set *writeset)
@@ -1817,6 +1928,8 @@ channel_handler_init_20(void)
 	channel_pre[SSH_CHANNEL_AUTH_SOCKET] =		&channel_pre_listener;
 	channel_pre[SSH_CHANNEL_CONNECTING] =		&channel_pre_connecting;
 	channel_pre[SSH_CHANNEL_DYNAMIC] =		&channel_pre_dynamic;
+	channel_pre[SSH_CHANNEL_MUX_LISTENER] =		&channel_pre_listener;
+	channel_pre[SSH_CHANNEL_MUX_CLIENT] =		&channel_pre_mux_client;
 
 	channel_post[SSH_CHANNEL_OPEN] =		&channel_post_open;
 	channel_post[SSH_CHANNEL_PORT_LISTENER] =	&channel_post_port_listener;
@@ -1825,6 +1938,8 @@ channel_handler_init_20(void)
 	channel_post[SSH_CHANNEL_AUTH_SOCKET] =		&channel_post_auth_listener;
 	channel_post[SSH_CHANNEL_CONNECTING] =		&channel_post_connecting;
 	channel_post[SSH_CHANNEL_DYNAMIC] =		&channel_post_open;
+	channel_post[SSH_CHANNEL_MUX_LISTENER] =	&channel_post_mux_listener;
+	channel_post[SSH_CHANNEL_MUX_CLIENT] =		&channel_post_mux_client;
 }
 
 static void
Index: channels.h
===================================================================
RCS file: /cvs/src/usr.bin/ssh/channels.h,v
retrieving revision 1.102
diff -u -p -r1.102 channels.h
--- channels.h	11 Jan 2010 01:39:46 -0000	1.102
+++ channels.h	18 Jan 2010 04:20:34 -0000
@@ -53,7 +53,9 @@
 #define SSH_CHANNEL_CONNECTING		12
 #define SSH_CHANNEL_DYNAMIC		13
 #define SSH_CHANNEL_ZOMBIE		14	/* Almost dead. */
-#define SSH_CHANNEL_MAX_TYPE		15
+#define SSH_CHANNEL_MUX_LISTENER	15	/* Listener for mux conn. */
+#define SSH_CHANNEL_MUX_CLIENT		16	/* Conn. to mux slave */
+#define SSH_CHANNEL_MAX_TYPE		17
 
 struct Channel;
 typedef struct Channel Channel;
@@ -81,6 +83,9 @@ struct channel_connect {
 	struct addrinfo *ai, *aitop;
 };
 
+/* Callbacks for mux channels back into client-specific code */
+typedef void mux_callback_fn(struct Channel *, void *);
+
 struct Channel {
 	int     type;		/* channel type/state */
 	int     self;		/* my own channel identifier */
@@ -92,7 +97,7 @@ struct Channel {
 	int     wfd;		/* write fd */
 	int     efd;		/* extended fd */
 	int     sock;		/* sock fd */
-	int     ctl_fd;		/* control fd (client sharing) */
+	int     ctl_chan;	/* control channel (multiplexed connections) */
 	int     isatty;		/* rfd is a tty */
 	int	client_tty;	/* (client) TTY has been requested */
 	int     force_drain;	/* force close on iEOF */
@@ -141,6 +146,10 @@ struct Channel {
 
 	/* non-blocking connect */
 	struct channel_connect	connect_ctx;
+
+	/* multiplexing protocol hook, called for each packet received */
+	mux_callback_fn		*mux_rcb;
+	void			*mux_ctx;
 };
 
 #define CHAN_EXTENDED_IGNORE		0
@@ -171,6 +180,7 @@ struct Channel {
 #define CHAN_CLOSE_RCVD			0x02
 #define CHAN_EOF_SENT			0x04
 #define CHAN_EOF_RCVD			0x08
+#define CHAN_LOCAL			0x10
 
 #define CHAN_RBUF	16*1024
 
@@ -242,7 +252,7 @@ void	 channel_clear_adm_permitted_opens(
 void 	 channel_print_adm_permitted_opens(void);
 int      channel_input_port_forward_request(int, int);
 Channel	*channel_connect_to(const char *, u_short, char *, char *);
-Channel	*channel_connect_stdio_fwd(const char*, u_short);
+Channel	*channel_connect_stdio_fwd(const char*, u_short, int, int);
 Channel	*channel_connect_by_listen_address(u_short, char *, char *);
 int	 channel_request_remote_forwarding(const char *, u_short,
 	     const char *, u_short);
Index: clientloop.c
===================================================================
RCS file: /cvs/src/usr.bin/ssh/clientloop.c,v
retrieving revision 1.216
diff -u -p -r1.216 clientloop.c
--- clientloop.c	9 Jan 2010 05:04:24 -0000	1.216
+++ clientloop.c	18 Jan 2010 04:20:35 -0000
@@ -113,7 +113,7 @@ extern int stdin_null_flag;
 extern int no_shell_flag;
 
 /* Control socket */
-extern int muxserver_sock;
+extern int muxserver_sock; /* XXX use mux_client_cleanup() instead */
 
 /*
  * Name of the host we are connecting to.  This is the name given on the
@@ -138,7 +138,7 @@ static volatile sig_atomic_t received_si
 static int in_non_blocking_mode = 0;
 
 /* Common data for the client loop code. */
-static volatile sig_atomic_t quit_pending; /* Set non-zero to quit the loop. */
+volatile sig_atomic_t quit_pending; /* Set non-zero to quit the loop. */
 static int escape_char1;	/* Escape character. (proto1 only) */
 static int escape_pending1;	/* Last character was an escape (proto1 only) */
 static int last_was_cr;		/* Last character was a newline. */
@@ -556,9 +556,6 @@ client_wait_until_can_do_something(fd_se
 	if (packet_have_data_to_write())
 		FD_SET(connection_out, *writesetp);
 
-	if (muxserver_sock != -1)
-		FD_SET(muxserver_sock, *readsetp);
-
 	/*
 	 * Wait for something to happen.  This will suspend the process until
 	 * some selected descriptor can be read, written, or has some other
@@ -686,7 +683,7 @@ client_status_confirm(int type, Channel 
 
 	/* XXX supress on mux _client_ quietmode */
 	tochan = options.log_level >= SYSLOG_LEVEL_ERROR &&
-	    c->ctl_fd != -1 && c->extended_usage == CHAN_EXTENDED_WRITE;
+	    c->ctl_chan != -1 && c->extended_usage == CHAN_EXTENDED_WRITE;
 
 	if (type == SSH2_MSG_CHANNEL_SUCCESS) {
 		debug2("%s request accepted on channel %d",
@@ -830,6 +827,7 @@ process_cmdline(void)
 	while (isspace(*++s))
 		;
 
+	/* XXX update list of forwards in options */
 	if (delete) {
 		cancel_port = 0;
 		cancel_host = hpdelim(&s);	/* may be NULL */
@@ -927,7 +925,7 @@ process_escapes(Channel *c, Buffer *bin,
 				    escape_char);
 				buffer_append(berr, string, strlen(string));
 
-				if (c && c->ctl_fd != -1) {
+				if (c && c->ctl_chan != -1) {
 					chan_read_failed(c);
 					chan_write_failed(c);
 					return 0;
@@ -937,7 +935,7 @@ process_escapes(Channel *c, Buffer *bin,
 
 			case 'Z' - 64:
 				/* XXX support this for mux clients */
-				if (c && c->ctl_fd != -1) {
+				if (c && c->ctl_chan != -1) {
  noescape:
 					snprintf(string, sizeof string,
 					    "%c%c escape not available to "
@@ -982,7 +980,7 @@ process_escapes(Channel *c, Buffer *bin,
 				continue;
 
 			case '&':
-				if (c && c->ctl_fd != -1)
+				if (c && c->ctl_chan != -1)
 					goto noescape;
 				/*
 				 * Detach the program (continue to serve
@@ -1033,7 +1031,7 @@ process_escapes(Channel *c, Buffer *bin,
 				continue;
 
 			case '?':
-				if (c && c->ctl_fd != -1) {
+				if (c && c->ctl_chan != -1) {
 					snprintf(string, sizeof string,
 "%c?\r\n\
 Supported escape sequences:\r\n\
@@ -1082,7 +1080,7 @@ Supported escape sequences:\r\n\
 				continue;
 
 			case 'C':
-				if (c && c->ctl_fd != -1)
+				if (c && c->ctl_chan != -1)
 					goto noescape;
 				process_cmdline();
 				continue;
@@ -1315,8 +1313,6 @@ client_loop(int have_pty, int escape_cha
 	connection_in = packet_get_connection_in();
 	connection_out = packet_get_connection_out();
 	max_fd = MAX(connection_in, connection_out);
-	if (muxserver_sock != -1)
-		max_fd = MAX(max_fd, muxserver_sock);
 
 	if (!compat20) {
 		/* enable nonblocking unless tty */
@@ -1434,12 +1430,6 @@ client_loop(int have_pty, int escape_cha
 		/* Buffer input from the connection.  */
 		client_process_net_input(readset);
 
-		/* Accept control connections.  */
-		if (muxserver_sock != -1 &&FD_ISSET(muxserver_sock, readset)) {
-			if (muxserver_accept_control())
-				quit_pending = 1;
-		}
-
 		if (quit_pending)
 			break;
 
@@ -1841,9 +1831,8 @@ client_input_channel_req(int type, u_int
 		chan_rcvd_eow(c);
 	} else if (strcmp(rtype, "exit-status") == 0) {
 		exitval = packet_get_int();
-		if (c->ctl_fd != -1) {
-			/* Dispatch to mux client */
-			atomicio(vwrite, c->ctl_fd, &exitval, sizeof(exitval));
+		if (c->ctl_chan != -1) {
+			mux_exit_message(c, exitval);
 			success = 1;
 		} else if (id == session_ident) {
 			/* Record exit value of local session */
Index: clientloop.h
===================================================================
RCS file: /cvs/src/usr.bin/ssh/clientloop.h,v
retrieving revision 1.22
diff -u -p -r1.22 clientloop.h
--- clientloop.h	12 Jun 2008 15:19:17 -0000	1.22
+++ clientloop.h	18 Jan 2010 04:20:35 -0000
@@ -56,18 +56,14 @@ typedef void global_confirm_cb(int, u_in
 void	 client_register_global_confirm(global_confirm_cb *, void *);
 
 /* Multiplexing protocol version */
-#define SSHMUX_VER			2
+#define SSHMUX_VER			4
 
 /* Multiplexing control protocol flags */
 #define SSHMUX_COMMAND_OPEN		1	/* Open new connection */
 #define SSHMUX_COMMAND_ALIVE_CHECK	2	/* Check master is alive */
 #define SSHMUX_COMMAND_TERMINATE	3	/* Ask master to exit */
-
-#define SSHMUX_FLAG_TTY			(1)	/* Request tty on open */
-#define SSHMUX_FLAG_SUBSYS		(1<<1)	/* Subsystem request on open */
-#define SSHMUX_FLAG_X11_FWD		(1<<2)	/* Request X11 forwarding */
-#define SSHMUX_FLAG_AGENT_FWD		(1<<3)	/* Request agent forwarding */
+#define SSHMUX_COMMAND_STDIO_FWD	4	/* Open stdio fwd (ssh -W) */
 
 void	muxserver_listen(void);
-int	muxserver_accept_control(void);
 void	muxclient(const char *);
+void	mux_exit_message(Channel *, int);
Index: mux.c
===================================================================
RCS file: /cvs/src/usr.bin/ssh/mux.c,v
retrieving revision 1.9
diff -u -p -r1.9 mux.c
--- mux.c	9 Jan 2010 05:04:24 -0000	1.9
+++ mux.c	18 Jan 2010 04:20:35 -0000
@@ -17,21 +17,20 @@
 
 /* ssh session multiplexing support */
 
+// XXX signal of slave passed to master
+
 /*
  * TODO:
- *   1. partial reads in muxserver_accept_control (maybe make channels
- *      from accepted connections)
- *   2. Better signalling from master to slave, especially passing of
+ *   - Better signalling from master to slave, especially passing of
  *      error messages
- *   3. Better fall-back from mux slave error to new connection.
- *   3. Add/delete forwardings via slave
- *   4. ExitOnForwardingFailure (after #3 obviously)
- *   5. Maybe extension mechanisms for multi-X11/multi-agent forwarding
- *   6. Document the mux mini-protocol somewhere.
- *   7. Support ~^Z in mux slaves.
- *   8. Inspect or control sessions in master.
- *   9. If we ever support the "signal" channel request, send signals on
- *      sessions in master.
+ *   - Better fall-back from mux slave error to new connection.
+ *   - ExitOnForwardingFailure (after #3 obviously)
+ *   - Maybe extension mechanisms for multi-X11/multi-agent forwarding
+ *   - Document the mux mini-protocol somewhere.
+ *   - Support ~^Z in mux slaves.
+ *   - Inspect or control sessions in master.
+ *   - If we ever support the "signal" channel request, send signals on
+ *     sessions in master.
  */
 
 #include <sys/types.h>
@@ -43,6 +42,7 @@
 
 #include <errno.h>
 #include <fcntl.h>
+#include <poll.h>
 #include <signal.h>
 #include <stdarg.h>
 #include <stddef.h>
@@ -53,6 +53,7 @@
 #include <util.h>
 #include <paths.h>
 
+#include "atomicio.h"
 #include "xmalloc.h"
 #include "log.h"
 #include "ssh.h"
@@ -77,13 +78,16 @@ extern int stdin_null_flag;
 extern char *host;
 extern int subsystem_flag;
 extern Buffer command;
+extern volatile sig_atomic_t quit_pending;
+extern char *stdio_forward_host;
+extern int stdio_forward_port;
 
 /* Context for session open confirmation callback */
 struct mux_session_confirm_ctx {
-	int want_tty;
-	int want_subsys;
-	int want_x_fwd;
-	int want_agent_fwd;
+	u_int want_tty;
+	u_int want_subsys;
+	u_int want_x_fwd;
+	u_int want_agent_fwd;
 	Buffer cmd;
 	char *term;
 	struct termios tio;
@@ -102,268 +106,238 @@ static volatile sig_atomic_t muxclient_t
 /* PID of multiplex server */
 static u_int muxserver_pid = 0;
 
+static Channel *mux_listener_channel = NULL;
 
-/* ** Multiplexing master support */
-
-/* Prepare a mux master to listen on a Unix domain socket. */
-void
-muxserver_listen(void)
-{
-	struct sockaddr_un addr;
-	mode_t old_umask;
-
-	if (options.control_path == NULL ||
-	    options.control_master == SSHCTL_MASTER_NO)
-		return;
-
-	debug("setting up multiplex master socket");
-
-	memset(&addr, '\0', sizeof(addr));
-	addr.sun_family = AF_UNIX;
-	addr.sun_len = offsetof(struct sockaddr_un, sun_path) +
-	    strlen(options.control_path) + 1;
+struct mux_master_state {
+	enum { MUX_HELLO_SEND, MUX_HELLO_WAIT, MUX_UP, MUX_SESSION } conn_state;
+};
 
-	if (strlcpy(addr.sun_path, options.control_path,
-	    sizeof(addr.sun_path)) >= sizeof(addr.sun_path))
-		fatal("ControlPath too long");
+/* mux protocol messages */
+#define MUX_MSG_HELLO           0x00000001
+#define MUX_C_NEW_SESSION       0x10000002
+#define MUX_C_ALIVE_CHECK       0x10000004
+#define MUX_C_TERMINATE         0x10000005
+#define MUX_C_OPEN_FORWARD      0x10000006
+#define MUX_C_CLOSE_FORWARD     0x10000007
+#define MUX_C_NEW_STDIO_FWD	0x10000008
+#define MUX_S_OK                0x80000001
+#define MUX_S_PERMISSION_DENIED 0x80000002
+#define MUX_S_FAILURE           0x80000003
+#define MUX_S_EXIT_MESSAGE      0x80000004
+#define MUX_S_ALIVE             0x80000005
+
+/* type codes for MUX_C_OPEN_FORWARD and MUX_C_CLOSE_FORWARD */
+#define MUX_FWD_LOCAL   1
+#define MUX_FWD_REMOTE  2
+#define MUX_FWD_DYNAMIC 3
+
+static void mux_session_confirm(int, void *);
+
+static int process_mux_master_hello(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+static int process_mux_new_session(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+static int process_mux_alive_check(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+static int process_mux_terminate(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+static int process_mux_open_forward(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+static int process_mux_close_forward(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+static int process_mux_stdio_fwd(struct mux_master_state *, Channel *,
+    Buffer *, Buffer *);
+
+static const struct {
+	u_int type;
+	int (*handler)(struct mux_master_state *, Channel *,
+	    Buffer *, Buffer *);
+} mux_master_handlers[] = {
+	{ MUX_MSG_HELLO, process_mux_master_hello },
+	{ MUX_C_NEW_SESSION, process_mux_new_session },
+	{ MUX_C_ALIVE_CHECK, process_mux_alive_check },
+	{ MUX_C_TERMINATE, process_mux_terminate },
+	{ MUX_C_OPEN_FORWARD, process_mux_open_forward },
+	{ MUX_C_CLOSE_FORWARD, process_mux_close_forward },
+	{ MUX_C_NEW_STDIO_FWD, process_mux_stdio_fwd },
+	{ 0, NULL }
+};
 
-	if ((muxserver_sock = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
-		fatal("%s socket(): %s", __func__, strerror(errno));
+/* Cleanup callback fired on closure of mux slave _session_ channel */
+/* ARGSUSED */
+static void
+mux_master_session_cleanup_cb(int cid, void *unused)
+{
+	Channel *cc, *c = channel_by_id(cid);
 
-	old_umask = umask(0177);
-	if (bind(muxserver_sock, (struct sockaddr *)&addr, addr.sun_len) == -1) {
-		muxserver_sock = -1;
-		if (errno == EINVAL || errno == EADDRINUSE) {
-			error("ControlSocket %s already exists, "
-			    "disabling multiplexing", options.control_path);
-			close(muxserver_sock);
-			muxserver_sock = -1;
-			xfree(options.control_path);
-			options.control_path = NULL;
-			options.control_master = SSHCTL_MASTER_NO;
-			return;
-		} else
-			fatal("%s bind(): %s", __func__, strerror(errno));
+	debug3("%s: entering for channel %d", __func__, cid);
+	if (c == NULL)
+		fatal("%s: channel_by_id(%i) == NULL", __func__, cid);
+	if (c->ctl_chan != -1) {
+		if ((cc = channel_by_id(c->ctl_chan)) == NULL)
+			fatal("%s: channel %d missing control channel %d",
+			    __func__, c->self, c->ctl_chan);
+		c->ctl_chan = -1;
+		cc->remote_id = -1;
+		chan_rcvd_oclose(cc);
 	}
-	umask(old_umask);
-
-	if (listen(muxserver_sock, 64) == -1)
-		fatal("%s listen(): %s", __func__, strerror(errno));
-
-	set_nonblock(muxserver_sock);
+	channel_cancel_cleanup(c->self);
 }
 
-/* Callback on open confirmation in mux master for a mux client session. */
+/* Cleanup callback fired on closure of mux slave _control_ channel */
+/* ARGSUSED */
 static void
-mux_session_confirm(int id, void *arg)
+mux_master_control_cleanup_cb(int cid, void *unused)
 {
-	struct mux_session_confirm_ctx *cctx = arg;
-	const char *display;
-	Channel *c;
-	int i;
-
-	if (cctx == NULL)
-		fatal("%s: cctx == NULL", __func__);
-	if ((c = channel_lookup(id)) == NULL)
-		fatal("%s: no channel for id %d", __func__, id);
-
-	display = getenv("DISPLAY");
-	if (cctx->want_x_fwd && options.forward_x11 && display != NULL) {
-		char *proto, *data;
-		/* Get reasonable local authentication information. */
-		client_x11_get_proto(display, options.xauth_location,
-		    options.forward_x11_trusted, &proto, &data);
-		/* Request forwarding with authentication spoofing. */
-		debug("Requesting X11 forwarding with authentication spoofing.");
-		x11_request_forwarding_with_spoofing(id, display, proto, data);
-		/* XXX wait for reply */
-	}
-
-	if (cctx->want_agent_fwd && options.forward_agent) {
-		debug("Requesting authentication agent forwarding.");
-		channel_request_start(id, "auth-agent-req at openssh.com", 0);
-		packet_send();
-	}
+	Channel *sc, *c = channel_by_id(cid);
 
-	client_session2_setup(id, cctx->want_tty, cctx->want_subsys,
-	    cctx->term, &cctx->tio, c->rfd, &cctx->cmd, cctx->env);
-
-	c->open_confirm_ctx = NULL;
-	buffer_free(&cctx->cmd);
-	xfree(cctx->term);
-	if (cctx->env != NULL) {
-		for (i = 0; cctx->env[i] != NULL; i++)
-			xfree(cctx->env[i]);
-		xfree(cctx->env);
+	debug3("%s: entering for channel %d", __func__, cid);
+	if (c == NULL)
+		fatal("%s: channel_by_id(%i) == NULL", __func__, cid);
+	if (c->remote_id != -1) {
+		if ((sc = channel_by_id(c->remote_id)) == NULL)
+			debug2("%s: channel %d n session channel %d",
+			    __func__, c->self, c->remote_id);
+		c->remote_id = -1;
+		sc->ctl_chan = -1;
+		chan_mark_dead(sc);
 	}
-	xfree(cctx);
+	channel_cancel_cleanup(c->self);
 }
 
-/*
- * Accept a connection on the mux master socket and process the
- * client's request. Returns flag indicating whether mux master should
- * begin graceful close.
- */
-int
-muxserver_accept_control(void)
+/* Check mux client environment variables before passing them to mux master. */
+static int
+env_permitted(char *env)
 {
-	Buffer m;
-	Channel *c;
-	int client_fd, new_fd[3], ver, allowed, window, packetmax;
-	socklen_t addrlen;
-	struct sockaddr_storage addr;
-	struct mux_session_confirm_ctx *cctx;
-	char *cmd;
-	u_int i, j, len, env_len, mux_command, flags, escape_char;
-	uid_t euid;
-	gid_t egid;
-	int start_close = 0;
-
-	/*
-	 * Accept connection on control socket
-	 */
-	memset(&addr, 0, sizeof(addr));
-	addrlen = sizeof(addr);
-	if ((client_fd = accept(muxserver_sock,
-	    (struct sockaddr*)&addr, &addrlen)) == -1) {
-		error("%s accept: %s", __func__, strerror(errno));
-		return 0;
-	}
+	int i, ret;
+	char name[1024], *cp;
 
-	if (getpeereid(client_fd, &euid, &egid) < 0) {
-		error("%s getpeereid failed: %s", __func__, strerror(errno));
-		close(client_fd);
+	if ((cp = strchr(env, '=')) == NULL || cp == env)
 		return 0;
-	}
-	if ((euid != 0) && (getuid() != euid)) {
-		error("control mode uid mismatch: peer euid %u != uid %u",
-		    (u_int) euid, (u_int) getuid());
-		close(client_fd);
+	ret = snprintf(name, sizeof(name), "%.*s", (int)(cp - env), env);
+	if (ret <= 0 || (size_t)ret >= sizeof(name)) {
+		error("env_permitted: name '%.100s...' too long", env);
 		return 0;
 	}
 
-	/* XXX handle asynchronously */
-	unset_nonblock(client_fd);
+	for (i = 0; i < options.num_send_env; i++)
+		if (match_pattern(name, options.send_env[i]))
+			return 1;
 
-	/* Read command */
-	buffer_init(&m);
-	if (ssh_msg_recv(client_fd, &m) == -1) {
-		error("%s: client msg_recv failed", __func__);
-		close(client_fd);
-		buffer_free(&m);
-		return 0;
-	}
-	if ((ver = buffer_get_char(&m)) != SSHMUX_VER) {
-		error("%s: wrong client version %d", __func__, ver);
-		buffer_free(&m);
-		close(client_fd);
-		return 0;
-	}
+	return 0;
+}
 
-	allowed = 1;
-	mux_command = buffer_get_int(&m);
-	flags = buffer_get_int(&m);
+/* Mux master protocol message handlers */
 
-	buffer_clear(&m);
+static int
+process_mux_master_hello(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
+{
+	u_int ver;
 
-	switch (mux_command) {
-	case SSHMUX_COMMAND_OPEN:
-		if (options.control_master == SSHCTL_MASTER_ASK ||
-		    options.control_master == SSHCTL_MASTER_AUTO_ASK)
-			allowed = ask_permission("Allow shared connection "
-			    "to %s? ", host);
-		/* continue below */
-		break;
-	case SSHMUX_COMMAND_TERMINATE:
-		if (options.control_master == SSHCTL_MASTER_ASK ||
-		    options.control_master == SSHCTL_MASTER_AUTO_ASK)
-			allowed = ask_permission("Terminate shared connection "
-			    "to %s? ", host);
-		if (allowed)
-			start_close = 1;
-		/* FALLTHROUGH */
-	case SSHMUX_COMMAND_ALIVE_CHECK:
-		/* Reply for SSHMUX_COMMAND_TERMINATE and ALIVE_CHECK */
-		buffer_clear(&m);
-		buffer_put_int(&m, allowed);
-		buffer_put_int(&m, getpid());
-		if (ssh_msg_send(client_fd, SSHMUX_VER, &m) == -1) {
-			error("%s: client msg_send failed", __func__);
-			close(client_fd);
-			buffer_free(&m);
-			return start_close;
+	if (state->conn_state != MUX_HELLO_WAIT) {
+		error("%s: MUX_MSG_HELLO received in state MUX_UP", __func__);
+		return -1;
+	}
+	if (buffer_get_int_ret(&ver, m) != 0) {
+ malf:
+		error("%s: malformed message", __func__);
+		return -1;
+	}
+	if (ver != SSHMUX_VER) {
+		error("Unsupported multiplexing protocol version %d "
+		    "(expected %d)", ver, SSHMUX_VER);
+		return -1;
+	}
+	debug2("%s: channel %d slave version %u", __func__, c->self, ver);
+
+	/* No extensions are presently defined */
+	while (buffer_len(m) > 0) {
+		char *name = buffer_get_string_ret(m, NULL);
+		char *value = buffer_get_string_ret(m, NULL);
+
+		if (name == NULL || value == NULL) {
+			if (name != NULL)
+				xfree(name);
+			goto malf;
 		}
-		buffer_free(&m);
-		close(client_fd);
-		return start_close;
-	default:
-		error("Unsupported command %d", mux_command);
-		buffer_free(&m);
-		close(client_fd);
-		return 0;
+		debug2("Unrecognised slave extension \"%s\"", name);
+		xfree(name);
+		xfree(value);
 	}
+	state->conn_state = MUX_UP;
+	return 0;
+}
 
-	/* Reply for SSHMUX_COMMAND_OPEN */
-	buffer_clear(&m);
-	buffer_put_int(&m, allowed);
-	buffer_put_int(&m, getpid());
-	if (ssh_msg_send(client_fd, SSHMUX_VER, &m) == -1) {
-		error("%s: client msg_send failed", __func__);
-		close(client_fd);
-		buffer_free(&m);
-		return 0;
+static int
+process_mux_new_session(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
+{
+	Channel *nc;
+	struct mux_session_confirm_ctx *cctx;
+	char *cmd, *cp;
+	u_int i, j, len, env_len, escape_char, window, packetmax;
+	int new_fd[3];
+
+	if (state->conn_state != MUX_UP) {
+		error("%s: incorrect state %u (expected %u)",
+		    __func__, state->conn_state, MUX_UP);
+		return -1;
 	}
 
-	if (!allowed) {
-		error("Refused control connection");
-		close(client_fd);
-		buffer_free(&m);
-		return 0;
+	/* Reply for SSHMUX_COMMAND_OPEN */
+	cctx = xcalloc(1, sizeof(*cctx));
+	cctx->term = NULL;
+	cmd = NULL;
+	if (buffer_get_int_ret(&cctx->want_tty, m) != 0 ||
+	    buffer_get_int_ret(&cctx->want_x_fwd, m) != 0 ||
+	    buffer_get_int_ret(&cctx->want_agent_fwd, m) != 0 ||
+	    buffer_get_int_ret(&cctx->want_subsys, m) != 0 ||
+	    buffer_get_int_ret(&escape_char, m) != 0 ||
+	    (cctx->term = buffer_get_string_ret(m, &len)) == NULL ||
+	    (cmd = buffer_get_string_ret(m, &len)) == NULL) {
+ malf:
+		if (cctx->term != NULL)
+			xfree(cctx->term);
+		error("%s: malformed message", __func__);
+		return -1;
 	}
 
-	buffer_clear(&m);
-	if (ssh_msg_recv(client_fd, &m) == -1) {
-		error("%s: client msg_recv failed", __func__);
-		close(client_fd);
-		buffer_free(&m);
-		return 0;
-	}
-	if ((ver = buffer_get_char(&m)) != SSHMUX_VER) {
-		error("%s: wrong client version %d", __func__, ver);
-		buffer_free(&m);
-		close(client_fd);
-		return 0;
+	cctx->env = NULL;
+	env_len = 0;
+	while (buffer_len(m) > 0) {
+#define MUX_MAX_ENV_VARS	4096
+		if ((cp = buffer_get_string_ret(m, &len)) == NULL) {
+			xfree(cmd);
+			goto malf;
+		}
+		if (!env_permitted(cp)) {
+			xfree(cp);
+			continue;
+		}
+		cctx->env = xrealloc(cctx->env, env_len + 2,
+		    sizeof(*cctx->env));
+		cctx->env[env_len++] = cp;
+		cctx->env[env_len] = NULL;
+		if (env_len > MUX_MAX_ENV_VARS) {
+			error(">%d environment variables received, ignoring "
+			    "additional", MUX_MAX_ENV_VARS);
+			break;
+		}
 	}
 
-	cctx = xcalloc(1, sizeof(*cctx));
-	cctx->want_tty = (flags & SSHMUX_FLAG_TTY) != 0;
-	cctx->want_subsys = (flags & SSHMUX_FLAG_SUBSYS) != 0;
-	cctx->want_x_fwd = (flags & SSHMUX_FLAG_X11_FWD) != 0;
-	cctx->want_agent_fwd = (flags & SSHMUX_FLAG_AGENT_FWD) != 0;
-	cctx->term = buffer_get_string(&m, &len);
-	escape_char = buffer_get_int(&m);
+	debug2("%s: channel %d: request tty %d, X %d, agent %d, subsys %d, "
+	    "term \"%s\", cmd \"%s\", env %u", __func__, c->self,
+	    cctx->want_tty, cctx->want_x_fwd, cctx->want_agent_fwd,
+	    cctx->want_subsys, cctx->term, cmd, env_len);
 
-	cmd = buffer_get_string(&m, &len);
 	buffer_init(&cctx->cmd);
 	buffer_append(&cctx->cmd, cmd, strlen(cmd));
-
-	env_len = buffer_get_int(&m);
-	env_len = MIN(env_len, 4096);
-	debug3("%s: receiving %d env vars", __func__, env_len);
-	if (env_len != 0) {
-		cctx->env = xcalloc(env_len + 1, sizeof(*cctx->env));
-		for (i = 0; i < env_len; i++)
-			cctx->env[i] = buffer_get_string(&m, &len);
-		cctx->env[i] = NULL;
-	}
-
-	debug2("%s: accepted tty %d, subsys %d, cmd %s", __func__,
-	    cctx->want_tty, cctx->want_subsys, cmd);
 	xfree(cmd);
 
 	/* Gather fds from client */
 	for(i = 0; i < 3; i++) {
-		if ((new_fd[i] = mm_receive_fd(client_fd)) == -1) {
+		if ((new_fd[i] = mm_receive_fd(c->sock)) == -1) {
 			error("%s: failed to receive fd %d from slave",
 			    __func__, i);
 			for (j = 0; j < i; j++)
@@ -374,38 +348,44 @@ muxserver_accept_control(void)
 				xfree(cctx->env);
 			xfree(cctx->term);
 			buffer_free(&cctx->cmd);
-			close(client_fd);
 			xfree(cctx);
-			return 0;
+
+			/* prepare reply */
+			buffer_put_int(r, MUX_S_FAILURE);
+			buffer_put_cstring(r,
+			    "did not receive file descriptors");
+			return -1;
 		}
 	}
 
-	debug2("%s: got fds stdin %d, stdout %d, stderr %d", __func__,
+	debug3("%s: got fds stdin %d, stdout %d, stderr %d", __func__,
 	    new_fd[0], new_fd[1], new_fd[2]);
 
+	if (options.control_master == SSHCTL_MASTER_ASK ||
+	    options.control_master == SSHCTL_MASTER_AUTO_ASK) {
+		if (!ask_permission("Allow shared connection to %s? ", host)) {
+			debug2("%s: session refused by user", __func__);
+			close(new_fd[0]);
+			close(new_fd[1]);
+			close(new_fd[2]);
+			xfree(cctx->term);
+			if (env_len != 0) {
+				for (i = 0; i < env_len; i++)
+					xfree(cctx->env[i]);
+				xfree(cctx->env);
+			}
+			buffer_free(&cctx->cmd);
+			/* prepare reply */
+			buffer_put_int(r, MUX_S_PERMISSION_DENIED);
+			buffer_put_cstring(r, "Permission denied");
+			return 0;
+		}
+	}
+
 	/* Try to pick up ttymodes from client before it goes raw */
 	if (cctx->want_tty && tcgetattr(new_fd[0], &cctx->tio) == -1)
 		error("%s: tcgetattr: %s", __func__, strerror(errno));
 
-	/* This roundtrip is just for synchronisation of ttymodes */
-	buffer_clear(&m);
-	if (ssh_msg_send(client_fd, SSHMUX_VER, &m) == -1) {
-		error("%s: client msg_send failed", __func__);
-		close(client_fd);
-		close(new_fd[0]);
-		close(new_fd[1]);
-		close(new_fd[2]);
-		buffer_free(&m);
-		xfree(cctx->term);
-		if (env_len != 0) {
-			for (i = 0; i < env_len; i++)
-				xfree(cctx->env[i]);
-			xfree(cctx->env);
-		}
-		return 0;
-	}
-	buffer_free(&m);
-
 	/* enable nonblocking unless tty */
 	if (!isatty(new_fd[0]))
 		set_nonblock(new_fd[0]);
@@ -414,257 +394,1048 @@ muxserver_accept_control(void)
 	if (!isatty(new_fd[2]))
 		set_nonblock(new_fd[2]);
 
-	set_nonblock(client_fd);
-
 	window = CHAN_SES_WINDOW_DEFAULT;
 	packetmax = CHAN_SES_PACKET_DEFAULT;
 	if (cctx->want_tty) {
 		window >>= 1;
 		packetmax >>= 1;
 	}
-	
-	c = channel_new("session", SSH_CHANNEL_OPENING,
+
+	nc = channel_new("session", SSH_CHANNEL_OPENING,
 	    new_fd[0], new_fd[1], new_fd[2], window, packetmax,
 	    CHAN_EXTENDED_WRITE, "client-session", /*nonblock*/0);
 
-	c->ctl_fd = client_fd;
+	nc->ctl_chan = c->self;		/* link session -> control channel */
+	c->remote_id = nc->self; 	/* link control -> session channel */
+
 	if (cctx->want_tty && escape_char != 0xffffffff) {
-		channel_register_filter(c->self,
+		channel_register_filter(nc->self,
 		    client_simple_escape_filter, NULL,
 		    client_filter_cleanup,
 		    client_new_escape_filter_ctx((int)escape_char));
 	}
 
-	debug3("%s: channel_new: %d", __func__, c->self);
-
-	channel_send_open(c->self);
-	channel_register_open_confirm(c->self, mux_session_confirm, cctx);
-	return 0;
-}
+	debug2("%s: channel_new: %d linked to control channel %d",
+	    __func__, nc->self, nc->ctl_chan);
 
-/* ** Multiplexing client support */
+	channel_send_open(nc->self);
+	channel_register_open_confirm(nc->self, mux_session_confirm, cctx);
+	channel_register_cleanup(nc->self, mux_master_session_cleanup_cb, 0);
+
+	/* prepare reply */
+	/* XXX defer until mux_session_confirm() fires */
+	buffer_put_int(r, MUX_S_OK);
+	state->conn_state = MUX_SESSION;
 
-/* Exit signal handler */
-static void
-control_client_sighandler(int signo)
-{
-	muxclient_terminate = signo;
+	return 0;
 }
 
-/*
- * Relay signal handler - used to pass some signals from mux client to
- * mux master.
- */
-static void
-control_client_sigrelay(int signo)
+static int
+process_mux_alive_check(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
 {
-	int save_errno = errno;
+	debug2("%s: channel %d: alive check", __func__, c->self);
 
-	if (muxserver_pid > 1)
-		kill(muxserver_pid, signo);
+	/* prepare reply */
+	buffer_put_int(r, MUX_S_ALIVE);
+	buffer_put_int(r, (u_int)getpid());
 
-	errno = save_errno;
+	return 0;
 }
 
-/* Check mux client environment variables before passing them to mux master. */
 static int
-env_permitted(char *env)
+process_mux_terminate(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
 {
-	int i, ret;
-	char name[1024], *cp;
-
-	if ((cp = strchr(env, '=')) == NULL || cp == env)
-		return (0);
-	ret = snprintf(name, sizeof(name), "%.*s", (int)(cp - env), env);
-	if (ret <= 0 || (size_t)ret >= sizeof(name))
-		fatal("env_permitted: name '%.100s...' too long", env);
+	debug2("%s: channel %d: terminate request", __func__, c->self);
 
-	for (i = 0; i < options.num_send_env; i++)
-		if (match_pattern(name, options.send_env[i]))
-			return (1);
+	if (options.control_master == SSHCTL_MASTER_ASK ||
+	    options.control_master == SSHCTL_MASTER_AUTO_ASK) {
+		if (!ask_permission("Terminate shared connection to %s? ",
+		    host)) {
+			debug2("%s: termination refused by user", __func__);
+			buffer_put_int(r, MUX_S_PERMISSION_DENIED);
+			buffer_put_cstring(r, "Permission denied");
+			return 0;
+		}
+	}
 
-	return (0);
+	quit_pending = 1;
+	buffer_put_int(r, MUX_S_OK);
+	/* XXX exit happens too soon - message never makes it to client */
+	return 0;
 }
 
-/* Multiplex client main loop. */
-void
-muxclient(const char *path)
+static char *
+format_forward(u_int ftype, Forward *fwd)
 {
-	struct sockaddr_un addr;
-	int i, r, fd, sock, exitval[2], num_env;
-	Buffer m;
-	char *term;
-	extern char **environ;
-	u_int allowed, flags;
-
-	if (muxclient_command == 0)
-		muxclient_command = SSHMUX_COMMAND_OPEN;
+	char *ret;
 
-	switch (options.control_master) {
-	case SSHCTL_MASTER_AUTO:
-	case SSHCTL_MASTER_AUTO_ASK:
-		debug("auto-mux: Trying existing master");
-		/* FALLTHROUGH */
-	case SSHCTL_MASTER_NO:
+	switch (ftype) {
+	case MUX_FWD_LOCAL:
+		xasprintf(&ret, "local forward %.200s:%d -> %.200s:%d",
+		    (fwd->listen_host == NULL) ?
+		    (options.gateway_ports ? "*" : "LOCALHOST") :
+		    fwd->listen_host, fwd->listen_port,
+		    fwd->connect_host, fwd->connect_port);
+		break;
+	case MUX_FWD_DYNAMIC:
+		xasprintf(&ret, "dynamic forward %.200s:%d -> *",
+		    (fwd->listen_host == NULL) ?
+		    (options.gateway_ports ? "*" : "LOCALHOST") :
+		     fwd->listen_host, fwd->listen_port);
+		break;
+	case MUX_FWD_REMOTE:
+		xasprintf(&ret, "remote forward %.200s:%d -> %.200s:%d",
+		    (fwd->listen_host == NULL) ?
+		    "LOCALHOST" : fwd->listen_host,
+		    fwd->listen_port,
+		    fwd->connect_host, fwd->connect_port);
 		break;
 	default:
-		return;
+		fatal("%s: unknown forward type %u", __func__, ftype);
 	}
+	return ret;
+}
 
-	memset(&addr, '\0', sizeof(addr));
-	addr.sun_family = AF_UNIX;
-	addr.sun_len = offsetof(struct sockaddr_un, sun_path) +
-	    strlen(path) + 1;
+static int
+compare_host(const char *a, const char *b)
+{
+	if (a == NULL && b == NULL)
+		return 1;
+	if (a == NULL || b == NULL)
+		return 0;
+	return strcmp(a, b) == 0;
+}
 
-	if (strlcpy(addr.sun_path, path,
-	    sizeof(addr.sun_path)) >= sizeof(addr.sun_path))
-		fatal("ControlPath too long");
+static int
+compare_forward(Forward *a, Forward *b)
+{
+	if (!compare_host(a->listen_host, b->listen_host))
+		return 0;
+	if (a->listen_port != b->listen_port)
+		return 0;
+	if (!compare_host(a->connect_host, b->connect_host))
+		return 0;
+	if (a->connect_port != b->connect_port)
+		return 0;
 
-	if ((sock = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
-		fatal("%s socket(): %s", __func__, strerror(errno));
+	return 1;
+}
 
-	if (connect(sock, (struct sockaddr *)&addr, addr.sun_len) == -1) {
-		if (muxclient_command != SSHMUX_COMMAND_OPEN) {
-			fatal("Control socket connect(%.100s): %s", path,
-			    strerror(errno));
-		}
-		if (errno == ENOENT)
-			debug("Control socket \"%.100s\" does not exist", path);
-		else {
-			error("Control socket connect(%.100s): %s", path,
-			    strerror(errno));
+static int
+process_mux_open_forward(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
+{
+	Forward fwd;
+	char *fwd_desc = NULL;
+	u_int ftype;
+	int i, ret = 0, freefwd = 1;
+
+	fwd.listen_host = fwd.connect_host = NULL;
+	if (buffer_get_int_ret(&ftype, m) != 0 ||
+	    (fwd.listen_host = buffer_get_string_ret(m, NULL)) == NULL ||
+	    buffer_get_int_ret(&fwd.listen_port, m) != 0 ||
+	    (fwd.connect_host = buffer_get_string_ret(m, NULL)) == NULL ||
+	    buffer_get_int_ret(&fwd.connect_port, m) != 0) {
+		error("%s: malformed message", __func__);
+		ret = -1;
+		goto out;
+	}
+
+	if (*fwd.listen_host == '\0') {
+		xfree(fwd.listen_host);
+		fwd.listen_host = NULL;
+	}
+	if (*fwd.connect_host == '\0') {
+		xfree(fwd.connect_host);
+		fwd.connect_host = NULL;
+	}
+
+	debug2("%s: channel %d: request %s", __func__, c->self,
+	    (fwd_desc = format_forward(ftype, &fwd)));
+
+	if (ftype != MUX_FWD_LOCAL && ftype != MUX_FWD_REMOTE &&
+	    ftype != MUX_FWD_DYNAMIC) {
+		logit("%s: invalid forwarding type %u", __func__, ftype);
+ invalid:
+		xfree(fwd.listen_host);
+		xfree(fwd.connect_host);
+		buffer_put_int(r, MUX_S_FAILURE);
+		buffer_put_cstring(r, "Invalid forwarding request");
+		return 0;
+	}
+	/* XXX support rport0 forwarding with reply of port assigned */
+	if (fwd.listen_port == 0 || fwd.listen_port >= 65536) {
+		logit("%s: invalid listen port %u", __func__,
+		    fwd.listen_port);
+		goto invalid;
+	}
+	if (fwd.connect_port >= 65536 || (ftype != MUX_FWD_DYNAMIC &&
+	    ftype != MUX_FWD_REMOTE && fwd.connect_port == 0)) {
+		logit("%s: invalid connect port %u", __func__,
+		    fwd.connect_port);
+		goto invalid;
+	}
+	if (ftype != MUX_FWD_DYNAMIC && fwd.connect_host == NULL) {
+		logit("%s: missing connect host", __func__);
+		goto invalid;
+	}
+
+	/* Skip forwards that have already been requested */
+	switch (ftype) {
+	case MUX_FWD_LOCAL:
+	case MUX_FWD_DYNAMIC:
+		for (i = 0; i < options.num_local_forwards; i++) {
+			if (compare_forward(&fwd,
+			    options.local_forwards + i)) {
+ exists:
+				debug2("%s: found existing forwarding",
+				    __func__);
+				buffer_put_int(r, MUX_S_OK);
+				goto out;
+			}
 		}
-		close(sock);
+		break;
+	case MUX_FWD_REMOTE:
+		for (i = 0; i < options.num_remote_forwards; i++) {
+			if (compare_forward(&fwd,
+			    options.remote_forwards + i))
+				goto exists;
+		}
+		break;
+	}
+
+	if (options.control_master == SSHCTL_MASTER_ASK ||
+	    options.control_master == SSHCTL_MASTER_AUTO_ASK) {
+		if (!ask_permission("Open %s on %s?", fwd_desc, host)) {
+			debug2("%s: forwarding refused by user", __func__);
+			buffer_put_int(r, MUX_S_PERMISSION_DENIED);
+			buffer_put_cstring(r, "Permission denied");
+			goto out;
+		}
+	}
+
+	if (ftype == MUX_FWD_LOCAL || ftype == MUX_FWD_DYNAMIC) {
+		if (options.num_local_forwards + 1 >=
+		    SSH_MAX_FORWARDS_PER_DIRECTION ||
+		    channel_setup_local_fwd_listener(fwd.listen_host,
+		    fwd.listen_port, fwd.connect_host, fwd.connect_port,
+		    options.gateway_ports) < 0) {
+ fail:
+			logit("slave-requested %s failed", fwd_desc);
+			buffer_put_int(r, MUX_S_FAILURE);
+			buffer_put_cstring(r, "Port forwarding failed");
+			goto out;
+		}
+		add_local_forward(&options, &fwd);
+		freefwd = 0;
+	} else {
+		/* XXX wait for remote to confirm */
+		if (options.num_remote_forwards + 1 >=
+		    SSH_MAX_FORWARDS_PER_DIRECTION ||
+		    channel_request_remote_forwarding(fwd.listen_host,
+		    fwd.listen_port, fwd.connect_host, fwd.connect_port) < 0)
+			goto fail;
+		add_remote_forward(&options, &fwd);
+		freefwd = 0;
+	}
+	buffer_put_int(r, MUX_S_OK);
+ out:
+	if (fwd_desc != NULL)
+		xfree(fwd_desc);
+	if (freefwd) {
+		if (fwd.listen_host != NULL)
+			xfree(fwd.listen_host);
+		if (fwd.connect_host != NULL)
+			xfree(fwd.connect_host);
+	}
+	return ret;
+}
+
+static int
+process_mux_close_forward(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
+{
+	Forward fwd;
+	char *fwd_desc = NULL;
+	u_int ftype;
+	int ret = 0;
+
+	fwd.listen_host = fwd.connect_host = NULL;
+	if (buffer_get_int_ret(&ftype, m) != 0 ||
+	    (fwd.listen_host = buffer_get_string_ret(m, NULL)) == NULL ||
+	    buffer_get_int_ret(&fwd.listen_port, m) != 0 ||
+	    (fwd.connect_host = buffer_get_string_ret(m, NULL)) == NULL ||
+	    buffer_get_int_ret(&fwd.connect_port, m) != 0) {
+		error("%s: malformed message", __func__);
+		ret = -1;
+		goto out;
+	}
+
+	if (*fwd.listen_host == '\0') {
+		xfree(fwd.listen_host);
+		fwd.listen_host = NULL;
+	}
+	if (*fwd.connect_host == '\0') {
+		xfree(fwd.connect_host);
+		fwd.connect_host = NULL;
+	}
+
+	debug2("%s: channel %d: request %s", __func__, c->self,
+	    (fwd_desc = format_forward(ftype, &fwd)));
+
+	/* XXX implement this */
+	buffer_put_int(r, MUX_S_FAILURE);
+	buffer_put_cstring(r, "unimplemented");
+
+ out:
+	if (fwd_desc != NULL)
+		xfree(fwd_desc);
+	if (fwd.listen_host != NULL)
+		xfree(fwd.listen_host);
+	if (fwd.connect_host != NULL)
+		xfree(fwd.connect_host);
+
+	return ret;
+}
+
+static int
+process_mux_stdio_fwd(struct mux_master_state *state, Channel *c,
+    Buffer *m, Buffer *r)
+{
+	Channel *nc;
+	char *chost;
+	u_int cport, i, j;
+	int new_fd[2];
+
+	if (state->conn_state != MUX_UP) {
+		error("%s: incorrect state %u (expected %u)",
+		    __func__, state->conn_state, MUX_UP);
+		return -1;
+	}
+
+	if ((chost = buffer_get_string_ret(m, NULL)) == NULL ||
+	    buffer_get_int_ret(&cport, m) != 0) {
+		if (chost != NULL)
+			xfree(chost);
+		error("%s: malformed message", __func__);
+		return -1;
+	}
+
+	debug2("%s: channel %d: request stdio fwd to %s:%u",
+	    __func__, c->self, chost, cport);
+
+	/* Gather fds from client */
+	for(i = 0; i < 2; i++) {
+		if ((new_fd[i] = mm_receive_fd(c->sock)) == -1) {
+			error("%s: failed to receive fd %d from slave",
+			    __func__, i);
+			for (j = 0; j < i; j++)
+				close(new_fd[j]);
+			xfree(chost);
+
+			/* prepare reply */
+			buffer_put_int(r, MUX_S_FAILURE);
+			buffer_put_cstring(r,
+			    "did not receive file descriptors");
+			return -1;
+		}
+	}
+
+	debug3("%s: got fds stdin %d, stdout %d", __func__,
+	    new_fd[0], new_fd[1]);
+
+	if (options.control_master == SSHCTL_MASTER_ASK ||
+	    options.control_master == SSHCTL_MASTER_AUTO_ASK) {
+		if (!ask_permission("Allow forward to to %s:%u? ",
+		    chost, cport)) {
+			debug2("%s: stdio fwd refused by user", __func__);
+			close(new_fd[0]);
+			close(new_fd[1]);
+			xfree(chost);
+			/* prepare reply */
+			buffer_put_int(r, MUX_S_PERMISSION_DENIED);
+			buffer_put_cstring(r, "Permission denied");
+			return 0;
+		}
+	}
+
+	/* enable nonblocking unless tty */
+	if (!isatty(new_fd[0]))
+		set_nonblock(new_fd[0]);
+	if (!isatty(new_fd[1]))
+		set_nonblock(new_fd[1]);
+
+	nc = channel_connect_stdio_fwd(chost, cport, new_fd[0], new_fd[1]);
+
+	nc->ctl_chan = c->self;		/* link session -> control channel */
+	c->remote_id = nc->self; 	/* link control -> session channel */
+
+	debug2("%s: channel_new: %d linked to control channel %d",
+	    __func__, nc->self, nc->ctl_chan);
+
+	channel_register_cleanup(nc->self, mux_master_session_cleanup_cb, 0);
+
+	/* prepare reply */
+	/* XXX defer until channel confirmed */
+	buffer_put_int(r, MUX_S_OK);
+	state->conn_state = MUX_SESSION;
+
+	return 0;
+}
+
+/* Channel callbacks fired on read/write from mux slave fd */
+static void
+mux_master_read_cb(Channel *c, void *ctx)
+{
+	struct mux_master_state *state = (struct mux_master_state *)ctx;
+	Buffer in, out;
+	void *ptr;
+	u_int type, have, i;
+	int ret = -1;
+
+	/* Complete setup of channel */
+	if (ctx == NULL) {
+		state = xcalloc(1, sizeof(state));
+		state->conn_state = MUX_HELLO_SEND;
+		c->mux_ctx = ctx = state;
+		channel_register_cleanup(c->self,
+		    mux_master_control_cleanup_cb, 0);
+	}
+
+/* 	debug3("%s: enter channel %d ibuf len %u obuf len %u state %d",
+	    __func__, c->self, buffer_len(&c->input), buffer_len(&c->output),
+	    state->conn_state); */
+
+	switch (state->conn_state) {
+	case MUX_HELLO_SEND:
+		buffer_init(&out);
+		buffer_put_int(&out, MUX_MSG_HELLO);
+		buffer_put_int(&out, SSHMUX_VER);
+		/* no extensions */
+		buffer_put_string(&c->output, buffer_ptr(&out),
+		    buffer_len(&out));
+		buffer_free(&out);
+		state->conn_state = MUX_HELLO_WAIT;
+		debug3("%s: channel %d: hello sent", __func__, c->self);
+		ret = 0;
+		break;
+	case MUX_HELLO_WAIT:
+	case MUX_UP:
+	case MUX_SESSION:
+		buffer_init(&in);
+		buffer_init(&out);
+
+		/* Channel code ensures that we receive whole packets */
+		if ((ptr = buffer_get_string_ptr_ret(&c->input,
+		    &have)) == NULL) {
+ malf:
+			error("%s: malformed message", __func__);
+			goto out;
+		}
+		buffer_append(&in, ptr, have);
+
+		if (buffer_get_int_ret(&type, &in))
+			goto malf;
+		debug3("%s: channel %d packet type 0x%08x len %u",
+		    __func__, c->self, type, buffer_len(&in));
+
+		if (state->conn_state == MUX_HELLO_WAIT &&
+		    type != MUX_MSG_HELLO) {
+			error("%s: expected MUX_MSG_HELLO(0x%08x), "
+			    "received 0x%08x", __func__, MUX_MSG_HELLO, type);
+			goto out;
+		}
+
+		for (i = 0; mux_master_handlers[i].handler != NULL; i++) {
+			if (type == mux_master_handlers[i].type) {
+				ret = mux_master_handlers[i].handler(state,
+				    c, &in, &out);
+				break;
+			}
+		}
+		if (mux_master_handlers[i].handler == NULL) {
+			error("%s: unsupported mux message 0x%08x",
+			    __func__, type);
+			buffer_put_int(&out, MUX_S_FAILURE);
+			buffer_put_cstring(&out, "unsupported request");
+			ret = 0;
+		}
+		/* Enqueue reply packet */
+		if (buffer_len(&out) != 0) {
+			buffer_put_string(&c->output, buffer_ptr(&out),
+			    buffer_len(&out));
+		}
+ out:
+/*		debug3("%s: reply channel %d ibuf len %u obuf len %u state %d",
+		    __func__, c->self, buffer_len(&c->input),
+		    buffer_len(&c->output), state->conn_state); */
+
+		buffer_free(&in);
+		buffer_free(&out);
+		break;
+	default:
+		fatal("%s: unknown state %d", __func__, state->conn_state);
+	}
+}
+
+void
+mux_exit_message(Channel *c, int exitval)
+{
+	Buffer m;
+	Channel *mux_chan;
+
+	debug3("%s: channel %d: exit message, evitval %d", __func__, c->self,
+	    exitval);
+
+	if ((mux_chan = channel_by_id(c->ctl_chan)) == NULL)
+		fatal("%s: channel %d missing mux channel %d",
+		    __func__, c->self, c->ctl_chan);
+
+	/* Append exit message packet to control socket output queue */
+	buffer_init(&m);
+	buffer_put_int(&m, MUX_S_EXIT_MESSAGE);
+	buffer_put_int(&m, exitval);
+
+	buffer_put_string(&mux_chan->output, buffer_ptr(&m), buffer_len(&m));
+	buffer_free(&m);
+}
+
+/* Prepare a mux master to listen on a Unix domain socket. */
+void
+muxserver_listen(void)
+{
+	struct sockaddr_un addr;
+	mode_t old_umask;
+
+	if (options.control_path == NULL ||
+	    options.control_master == SSHCTL_MASTER_NO)
 		return;
+
+	debug("setting up multiplex master socket");
+
+	memset(&addr, '\0', sizeof(addr));
+	addr.sun_family = AF_UNIX;
+	addr.sun_len = offsetof(struct sockaddr_un, sun_path) +
+	    strlen(options.control_path) + 1;
+
+	if (strlcpy(addr.sun_path, options.control_path,
+	    sizeof(addr.sun_path)) >= sizeof(addr.sun_path))
+		fatal("ControlPath too long");
+
+	if ((muxserver_sock = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
+		fatal("%s socket(): %s", __func__, strerror(errno));
+
+	old_umask = umask(0177);
+	if (bind(muxserver_sock, (struct sockaddr *)&addr, addr.sun_len) == -1) {
+		muxserver_sock = -1;
+		if (errno == EINVAL || errno == EADDRINUSE) {
+			error("ControlSocket %s already exists, "
+			    "disabling multiplexing", options.control_path);
+			close(muxserver_sock);
+			muxserver_sock = -1;
+			xfree(options.control_path);
+			options.control_path = NULL;
+			options.control_master = SSHCTL_MASTER_NO;
+			return;
+		} else
+			fatal("%s bind(): %s", __func__, strerror(errno));
 	}
+	umask(old_umask);
 
-	if (stdin_null_flag) {
-		if ((fd = open(_PATH_DEVNULL, O_RDONLY)) == -1)
-			fatal("open(/dev/null): %s", strerror(errno));
-		if (dup2(fd, STDIN_FILENO) == -1)
-			fatal("dup2: %s", strerror(errno));
-		if (fd > STDERR_FILENO)
-			close(fd);
+	if (listen(muxserver_sock, 64) == -1)
+		fatal("%s listen(): %s", __func__, strerror(errno));
+
+	set_nonblock(muxserver_sock);
+
+	mux_listener_channel = channel_new("mux listener",
+	    SSH_CHANNEL_MUX_LISTENER, muxserver_sock, muxserver_sock, -1,
+	    CHAN_TCP_WINDOW_DEFAULT, CHAN_TCP_PACKET_DEFAULT,
+	    0, addr.sun_path, 1);
+	mux_listener_channel->mux_rcb = mux_master_read_cb;
+	debug3("%s: mux listener channel %d fd %d", __func__,
+	    mux_listener_channel->self, mux_listener_channel->sock);
+}
+
+/* Callback on open confirmation in mux master for a mux client session. */
+static void
+mux_session_confirm(int id, void *arg)
+{
+	struct mux_session_confirm_ctx *cctx = arg;
+	const char *display;
+	Channel *c;
+	int i;
+
+	if (cctx == NULL)
+		fatal("%s: cctx == NULL", __func__);
+	if ((c = channel_by_id(id)) == NULL)
+		fatal("%s: no channel for id %d", __func__, id);
+
+	display = getenv("DISPLAY");
+	if (cctx->want_x_fwd && options.forward_x11 && display != NULL) {
+		char *proto, *data;
+		/* Get reasonable local authentication information. */
+		client_x11_get_proto(display, options.xauth_location,
+		    options.forward_x11_trusted, &proto, &data);
+		/* Request forwarding with authentication spoofing. */
+		debug("Requesting X11 forwarding with authentication spoofing.");
+		x11_request_forwarding_with_spoofing(id, display, proto, data);
+		/* XXX wait for reply */
 	}
 
-	term = getenv("TERM");
+	if (cctx->want_agent_fwd && options.forward_agent) {
+		debug("Requesting authentication agent forwarding.");
+		channel_request_start(id, "auth-agent-req at openssh.com", 0);
+		packet_send();
+	}
 
-	flags = 0;
-	if (tty_flag)
-		flags |= SSHMUX_FLAG_TTY;
-	if (subsystem_flag)
-		flags |= SSHMUX_FLAG_SUBSYS;
-	if (options.forward_x11)
-		flags |= SSHMUX_FLAG_X11_FWD;
-	if (options.forward_agent)
-		flags |= SSHMUX_FLAG_AGENT_FWD;
+	client_session2_setup(id, cctx->want_tty, cctx->want_subsys,
+	    cctx->term, &cctx->tio, c->rfd, &cctx->cmd, cctx->env);
 
-	signal(SIGPIPE, SIG_IGN);
+	c->open_confirm_ctx = NULL;
+	buffer_free(&cctx->cmd);
+	xfree(cctx->term);
+	if (cctx->env != NULL) {
+		for (i = 0; cctx->env[i] != NULL; i++)
+			xfree(cctx->env[i]);
+		xfree(cctx->env);
+	}
+	xfree(cctx);
+}
+
+/* ** Multiplexing client support */
+
+/* Exit signal handler */
+static void
+control_client_sighandler(int signo)
+{
+	muxclient_terminate = signo;
+}
+
+/*
+ * Relay signal handler - used to pass some signals from mux client to
+ * mux master.
+ */
+static void
+control_client_sigrelay(int signo)
+{
+	int save_errno = errno;
+
+	if (muxserver_pid > 1)
+		kill(muxserver_pid, signo);
+
+	errno = save_errno;
+}
+
+static int
+mux_client_read(int fd, Buffer *b, u_int need)
+{
+	u_int have;
+	ssize_t len;
+	u_char *p;
+	struct pollfd pfd;
+
+	pfd.fd = fd;
+	pfd.events = POLLIN;
+	p = buffer_append_space(b, need);
+	for (have = 0; have < need; ) {
+		if (muxclient_terminate) {
+			errno = EINTR;
+			return -1;
+		}
+		len = read(fd, p + have, need - have);
+		if (len < 0) {
+			switch (errno) {
+			case EAGAIN:
+				(void)poll(&pfd, 1, -1);
+				/* FALLTHROUGH */
+			case EINTR:
+				continue;
+			default:
+				return -1;
+			}
+		}
+		if (len == 0) {
+			errno = EPIPE;
+			return -1;
+		}
+		have += (u_int)len;
+	}
+	return 0;
+}
+
+static int
+mux_client_write_packet(int fd, Buffer *m)
+{
+	Buffer queue;
+	u_int have, need;
+	int oerrno, len;
+	u_char *ptr;
+	struct pollfd pfd;
+
+	pfd.fd = fd;
+	pfd.events = POLLOUT;
+	buffer_init(&queue);
+	buffer_put_string(&queue, buffer_ptr(m), buffer_len(m));
+
+	need = buffer_len(&queue);
+	ptr = buffer_ptr(&queue);
+
+	for (have = 0; have < need; ) {
+		if (muxclient_terminate) {
+			buffer_free(&queue);
+			errno = EINTR;
+			return -1;
+		}
+		len = write(fd, ptr + have, need - have);
+		if (len < 0) {
+			switch (errno) {
+			case EAGAIN:
+				(void)poll(&pfd, 1, -1);
+				/* FALLTHROUGH */
+			case EINTR:
+				continue;
+			default:
+				oerrno = errno;
+				buffer_free(&queue);
+				errno = oerrno;
+				return -1;
+			}
+		}
+		if (len == 0) {
+			buffer_free(&queue);
+			errno = EPIPE;
+			return -1;
+		}
+		have += (u_int)len;
+	}
+	buffer_free(&queue);
+	return 0;
+}
+
+static int
+mux_client_read_packet(int fd, Buffer *m)
+{
+	Buffer queue;
+	u_int need, have;
+	void *ptr;
+	int oerrno;
+
+	buffer_init(&queue);
+	if (mux_client_read(fd, &queue, 4) != 0) {
+		if ((oerrno = errno) == EPIPE)
+		debug3("%s: read header failed: %s", __func__, strerror(errno));
+		errno = oerrno;
+		return -1;
+	}
+	need = get_u32(buffer_ptr(&queue));
+	if (mux_client_read(fd, &queue, need) != 0) {
+		oerrno = errno;
+		debug3("%s: read body failed: %s", __func__, strerror(errno));
+		errno = oerrno;
+		return -1;
+	}
+	ptr = buffer_get_string_ptr(&queue, &have);
+	buffer_append(m, ptr, have);
+	buffer_free(&queue);
+	return 0;
+}
+
+static int
+mux_client_hello_exchange(int fd)
+{
+	Buffer m;
+	u_int type, ver;
 
 	buffer_init(&m);
+	buffer_put_int(&m, MUX_MSG_HELLO);
+	buffer_put_int(&m, SSHMUX_VER);
+	/* no extensions */
 
-	/* Send our command to server */
-	buffer_put_int(&m, muxclient_command);
-	buffer_put_int(&m, flags);
-	if (ssh_msg_send(sock, SSHMUX_VER, &m) == -1) {
-		error("%s: msg_send", __func__);
- muxerr:
-		close(sock);
+	if (mux_client_write_packet(fd, &m) != 0)
+		fatal("%s: write packet: %s", __func__, strerror(errno));
+
+	buffer_clear(&m);
+
+	/* Read their HELLO */
+	if (mux_client_read_packet(fd, &m) != 0) {
 		buffer_free(&m);
-		if (muxclient_command != SSHMUX_COMMAND_OPEN)
-			cleanup_exit(255);
-		logit("Falling back to non-multiplexed connection");
-		xfree(options.control_path);
-		options.control_path = NULL;
-		options.control_master = SSHCTL_MASTER_NO;
-		return;
+		return -1;
 	}
+
+	type = buffer_get_int(&m);
+	if (type != MUX_MSG_HELLO)
+		fatal("%s: expected HELLO (%u) received %u",
+		    __func__, MUX_MSG_HELLO, type);
+	ver = buffer_get_int(&m);
+	if (ver != SSHMUX_VER)
+		fatal("Unsupported multiplexing protocol version %d "
+		    "(expected %d)", ver, SSHMUX_VER);
+	debug2("%s: master version %u", __func__, ver);
+	/* No extensions are presently defined */
+	while (buffer_len(&m) > 0) {
+		char *name = buffer_get_string(&m, NULL);
+		char *value = buffer_get_string(&m, NULL);
+
+		debug2("Unrecognised master extension \"%s\"", name);
+		xfree(name);
+		xfree(value);
+	}
+	buffer_free(&m);
+	return 0;
+}
+
+static u_int
+mux_client_request_alive(int fd)
+{
+	Buffer m;
+	char *e;
+	u_int pid, type;
+
+	debug3("%s: entering", __func__);
+
+	buffer_init(&m);
+	buffer_put_int(&m, MUX_C_ALIVE_CHECK);
+
+	if (mux_client_write_packet(fd, &m) != 0)
+		fatal("%s: write packet: %s", __func__, strerror(errno));
+
 	buffer_clear(&m);
 
-	/* Get authorisation status and PID of controlee */
-	if (ssh_msg_recv(sock, &m) == -1) {
-		error("%s: Did not receive reply from master", __func__);
-		goto muxerr;
-	}
-	if (buffer_get_char(&m) != SSHMUX_VER) {
-		error("%s: Master replied with wrong version", __func__);
-		goto muxerr;
-	}
-	if (buffer_get_int_ret(&allowed, &m) != 0) {
-		error("%s: bad server reply", __func__);
-		goto muxerr;
-	}
-	if (allowed != 1) {
-		error("Connection to master denied");
-		goto muxerr;
+	/* Read their reply */
+	if (mux_client_read_packet(fd, &m) != 0) {
+		buffer_free(&m);
+		return 0;
+	}
+
+	type = buffer_get_int(&m);
+	if (type != MUX_S_ALIVE) {
+		e = buffer_get_string(&m, NULL);
+		fatal("%s: master returned error: %s", __func__, e);
 	}
-	muxserver_pid = buffer_get_int(&m);
+
+	pid = buffer_get_int(&m);
+	buffer_free(&m);
+
+	debug3("%s: done pid = %u", __func__, pid);
+
+	return pid;
+}
+
+static void
+mux_client_request_terminate(int fd)
+{
+	Buffer m;
+	char *e;
+	u_int type;
+
+	debug3("%s: entering", __func__);
+
+	buffer_init(&m);
+	buffer_put_int(&m, MUX_C_TERMINATE);
+
+	if (mux_client_write_packet(fd, &m) != 0)
+		fatal("%s: write packet: %s", __func__, strerror(errno));
 
 	buffer_clear(&m);
 
-	switch (muxclient_command) {
-	case SSHMUX_COMMAND_ALIVE_CHECK:
-		fprintf(stderr, "Master running (pid=%d)\r\n",
-		    muxserver_pid);
-		exit(0);
-	case SSHMUX_COMMAND_TERMINATE:
-		fprintf(stderr, "Exit request sent.\r\n");
-		exit(0);
-	case SSHMUX_COMMAND_OPEN:
-		buffer_put_cstring(&m, term ? term : "");
-		if (options.escape_char == SSH_ESCAPECHAR_NONE)
-			buffer_put_int(&m, 0xffffffff);
-		else
-			buffer_put_int(&m, options.escape_char);
-		buffer_append(&command, "\0", 1);
-		buffer_put_cstring(&m, buffer_ptr(&command));
-
-		if (options.num_send_env == 0 || environ == NULL) {
-			buffer_put_int(&m, 0);
-		} else {
-			/* Pass environment */
-			num_env = 0;
-			for (i = 0; environ[i] != NULL; i++) {
-				if (env_permitted(environ[i]))
-					num_env++; /* Count */
-			}
-			buffer_put_int(&m, num_env);
-		for (i = 0; environ[i] != NULL && num_env >= 0; i++) {
-				if (env_permitted(environ[i])) {
-					num_env--;
-					buffer_put_cstring(&m, environ[i]);
-				}
-			}
+	/* Read their reply */
+	if (mux_client_read_packet(fd, &m) != 0) {
+		/* Remote end exited already */
+		if (errno == EPIPE) {
+			buffer_free(&m);
+			return;
 		}
+		fatal("%s: read from master failed: %s",
+		    __func__, strerror(errno));
+	}
+
+	type = buffer_get_int(&m);
+	switch (type) {
+	case MUX_S_OK:
 		break;
+	case MUX_S_PERMISSION_DENIED:
+		e = buffer_get_string(&m, NULL);
+		fatal("Master refused termination request: %s", e);
+	case MUX_S_FAILURE:
+		e = buffer_get_string(&m, NULL);
+		fatal("%s: termination request failed: %s", __func__, e);
 	default:
-		fatal("unrecognised muxclient_command %d", muxclient_command);
+		fatal("%s: unexpected response from master 0x%08x",
+		    __func__, type);
+	}
+	buffer_free(&m);
+}
+
+static int
+mux_client_request_forward(int fd, u_int ftype, Forward *fwd)
+{
+	Buffer m;
+	char *e, *fwd_desc;
+	u_int type;
+
+	fwd_desc = format_forward(ftype, fwd);
+	debug("Requesting %s", fwd_desc);
+	xfree(fwd_desc);
+
+	buffer_init(&m);
+	buffer_put_int(&m, MUX_C_OPEN_FORWARD);
+	buffer_put_int(&m, ftype);
+	buffer_put_cstring(&m,
+	    fwd->listen_host == NULL ? "" : fwd->listen_host);
+	buffer_put_int(&m, fwd->listen_port);
+	buffer_put_cstring(&m,
+	    fwd->connect_host == NULL ? "" : fwd->connect_host);
+	buffer_put_int(&m, fwd->connect_port);
+
+	if (mux_client_write_packet(fd, &m) != 0)
+		fatal("%s: write packet: %s", __func__, strerror(errno));
+
+	buffer_clear(&m);
+
+	/* Read their reply */
+	if (mux_client_read_packet(fd, &m) != 0) {
+		buffer_free(&m);
+		return -1;
 	}
 
-	if (ssh_msg_send(sock, SSHMUX_VER, &m) == -1) {
-		error("%s: msg_send", __func__);
-		goto muxerr;
+	type = buffer_get_int(&m);
+	switch (type) {
+	case MUX_S_OK:
+		break;
+	case MUX_S_PERMISSION_DENIED:
+		e = buffer_get_string(&m, NULL);
+		buffer_free(&m);
+		error("Master refused forwarding request: %s", e);
+		return -1;
+	case MUX_S_FAILURE:
+		e = buffer_get_string(&m, NULL);
+		buffer_free(&m);
+		error("%s: session request failed: %s", __func__, e);
+		return -1;
+	default:
+		fatal("%s: unexpected response from master 0x%08x",
+		    __func__, type);
 	}
+	buffer_free(&m);
 
-	if (mm_send_fd(sock, STDIN_FILENO) == -1 ||
-	    mm_send_fd(sock, STDOUT_FILENO) == -1 ||
-	    mm_send_fd(sock, STDERR_FILENO) == -1) {
-		error("%s: send fds failed", __func__);
-		goto muxerr;
+	return 0;
+}
+
+static int
+mux_client_request_forwards(int fd)
+{
+	int i;
+
+	debug3("%s: requesting forwardings: %d local, %d remote", __func__,
+	    options.num_local_forwards, options.num_remote_forwards);
+
+	/* XXX ExitOnForwardingFailure */
+	for (i = 0; i < options.num_local_forwards; i++) {
+		if (mux_client_request_forward(fd,
+		    options.local_forwards[i].connect_port == 0 ?
+		    MUX_FWD_DYNAMIC : MUX_FWD_LOCAL,
+		    options.local_forwards + i) != 0)
+			return -1;
+	}
+	for (i = 0; i < options.num_remote_forwards; i++) {
+		if (mux_client_request_forward(fd, MUX_FWD_REMOTE,
+		    options.remote_forwards + i) != 0)
+			return -1;
 	}
+	return 0;
+}
 
-	/*
-	 * Mux errors are non-recoverable from this point as the master
-	 * has ownership of the session now.
-	 */
+static int
+mux_client_request_session(int fd)
+{
+	Buffer m;
+	char *e, *term;
+	u_int i, exitval, type, exitval_seen;
+	extern char **environ;
+	int devnull;
+
+	debug3("%s: entering", __func__);
+
+	if ((muxserver_pid = mux_client_request_alive(fd)) == 0) {
+		error("%s: master alive request failed", __func__);
+		return -1;
+	}
 
-	/* Wait for reply, so master has a chance to gather ttymodes */
+	signal(SIGPIPE, SIG_IGN);
+
+	if (stdin_null_flag) {
+		if ((devnull = open(_PATH_DEVNULL, O_RDONLY)) == -1)
+			fatal("open(/dev/null): %s", strerror(errno));
+		if (dup2(devnull, STDIN_FILENO) == -1)
+			fatal("dup2: %s", strerror(errno));
+		if (devnull > STDERR_FILENO)
+			close(devnull);
+	}
+
+	term = getenv("TERM");
+
+	buffer_init(&m);
+	buffer_put_int(&m, MUX_C_NEW_SESSION);
+	buffer_put_int(&m, tty_flag);
+	buffer_put_int(&m, options.forward_x11);
+	buffer_put_int(&m, options.forward_agent);
+	buffer_put_int(&m, subsystem_flag);
+	buffer_put_int(&m, options.escape_char == SSH_ESCAPECHAR_NONE ?
+	    0xffffffff : (u_int)options.escape_char);
+	buffer_put_cstring(&m, term == NULL ? "" : term);
+	buffer_put_string(&m, buffer_ptr(&command), buffer_len(&command));
+
+	if (options.num_send_env > 0 && environ != NULL) {
+		/* Pass environment */
+		for (i = 0; environ[i] != NULL; i++) {
+			if (env_permitted(environ[i])) {
+				buffer_put_cstring(&m, environ[i]);
+			}
+		}
+	}
+
+	if (mux_client_write_packet(fd, &m) != 0)
+		fatal("%s: write packet: %s", __func__, strerror(errno));
+
+	/* Send the stdio file descriptors */
+	if (mm_send_fd(fd, STDIN_FILENO) == -1 ||
+	    mm_send_fd(fd, STDOUT_FILENO) == -1 ||
+	    mm_send_fd(fd, STDERR_FILENO) == -1)
+		fatal("%s: send fds failed", __func__);
+
+	debug3("%s: session request sent", __func__);
+
+	/* Read their reply */
 	buffer_clear(&m);
-	if (ssh_msg_recv(sock, &m) == -1)
-		fatal("%s: msg_recv", __func__);
-	if (buffer_get_char(&m) != SSHMUX_VER)
-		fatal("%s: wrong version", __func__);
-	buffer_free(&m);
+	if (mux_client_read_packet(fd, &m) != 0) {
+		error("%s: read from master failed: %s",
+		    __func__, strerror(errno));
+		buffer_free(&m);
+		return -1;
+	}
+
+	type = buffer_get_int(&m);
+	switch (type) {
+	case MUX_S_OK:
+		break;
+	case MUX_S_PERMISSION_DENIED:
+		e = buffer_get_string(&m, NULL);
+		buffer_free(&m);
+		error("Master refused forwarding request: %s", e);
+		return -1;
+	case MUX_S_FAILURE:
+		e = buffer_get_string(&m, NULL);
+		buffer_free(&m);
+		error("%s: forwarding request failed: %s", __func__, e);
+		return -1;
+	default:
+		buffer_free(&m);
+		error("%s: unexpected response from master 0x%08x",
+		    __func__, type);
+		return -1;
+	}
 
 	signal(SIGHUP, control_client_sighandler);
 	signal(SIGINT, control_client_sighandler);
@@ -676,42 +1447,224 @@ muxclient(const char *path)
 
 	/*
 	 * Stick around until the controlee closes the client_fd.
-	 * Before it does, it is expected to write this process' exit
-	 * value (one int). This process must read the value and wait for
-	 * the closure of the client_fd; if this one closes early, the 
-	 * multiplex master will terminate early too (possibly losing data).
+	 * Before it does, it is expected to write an exit message.
+	 * This process must read the value and wait for the closure of
+	 * the client_fd; if this one closes early, the multiplex master will
+	 * terminate early too (possibly losing data).
 	 */
-	exitval[0] = 0;
-	for (i = 0; !muxclient_terminate && i < (int)sizeof(exitval);) {
-		r = read(sock, (char *)exitval + i, sizeof(exitval) - i);
-		if (r == 0) {
-			debug2("Received EOF from master");
+	for (exitval = 255, exitval_seen = 0;;) {
+		buffer_clear(&m);
+		if (mux_client_read_packet(fd, &m) != 0)
 			break;
+		type = buffer_get_int(&m);
+		if (type != MUX_S_EXIT_MESSAGE) {
+			e = buffer_get_string(&m, NULL);
+			fatal("%s: master returned error: %s", __func__, e);
 		}
-		if (r == -1) {
-			if (errno == EINTR)
-				continue;
-			fatal("%s: read %s", __func__, strerror(errno));
-		}
-		i += r;
+		if (exitval_seen)
+			fatal("%s: exitval sent twice", __func__);
+		exitval = buffer_get_int(&m);
+		exitval_seen = 1;
 	}
 
-	close(sock);
+	close(fd);
 	leave_raw_mode(force_tty_flag);
-	if (i > (int)sizeof(int))
-		fatal("%s: master returned too much data (%d > %lu)",
-		    __func__, i, (u_long)sizeof(int));
+
 	if (muxclient_terminate) {
 		debug2("Exiting on signal %d", muxclient_terminate);
-		exitval[0] = 255;
-	} else if (i < (int)sizeof(int)) {
+		exitval = 255;
+	} else if (!exitval_seen) {
 		debug2("Control master terminated unexpectedly");
-		exitval[0] = 255;
+		exitval = 255;
 	} else
-		debug2("Received exit status from master %d", exitval[0]);
+		debug2("Received exit status from master %d", exitval);
 
 	if (tty_flag && options.log_level != SYSLOG_LEVEL_QUIET)
 		fprintf(stderr, "Shared connection to %s closed.\r\n", host);
 
-	exit(exitval[0]);
+	exit(exitval);
+}
+
+static int
+mux_client_request_stdio_fwd(int fd)
+{
+	Buffer m;
+	char *e;
+	u_int type;
+	int devnull;
+
+	debug3("%s: entering", __func__);
+
+	if ((muxserver_pid = mux_client_request_alive(fd)) == 0) {
+		error("%s: master alive request failed", __func__);
+		return -1;
+	}
+
+	signal(SIGPIPE, SIG_IGN);
+
+	if (stdin_null_flag) {
+		if ((devnull = open(_PATH_DEVNULL, O_RDONLY)) == -1)
+			fatal("open(/dev/null): %s", strerror(errno));
+		if (dup2(devnull, STDIN_FILENO) == -1)
+			fatal("dup2: %s", strerror(errno));
+		if (devnull > STDERR_FILENO)
+			close(devnull);
+	}
+
+	buffer_init(&m);
+	buffer_put_int(&m, MUX_C_NEW_STDIO_FWD);
+	buffer_put_cstring(&m, stdio_forward_host);
+	buffer_put_int(&m, stdio_forward_port);
+
+	if (mux_client_write_packet(fd, &m) != 0)
+		fatal("%s: write packet: %s", __func__, strerror(errno));
+
+	/* Send the stdio file descriptors */
+	if (mm_send_fd(fd, STDIN_FILENO) == -1 ||
+	    mm_send_fd(fd, STDOUT_FILENO) == -1)
+		fatal("%s: send fds failed", __func__);
+
+	debug3("%s: stdio forward request sent", __func__);
+
+	/* Read their reply */
+	buffer_clear(&m);
+
+	if (mux_client_read_packet(fd, &m) != 0) {
+		error("%s: read from master failed: %s",
+		    __func__, strerror(errno));
+		buffer_free(&m);
+		return -1;
+	}
+
+	type = buffer_get_int(&m);
+	switch (type) {
+	case MUX_S_OK:
+		break;
+	case MUX_S_PERMISSION_DENIED:
+		e = buffer_get_string(&m, NULL);
+		buffer_free(&m);
+		fatal("Master refused forwarding request: %s", e);
+	case MUX_S_FAILURE:
+		e = buffer_get_string(&m, NULL);
+		buffer_free(&m);
+		fatal("%s: stdio forwarding request failed: %s", __func__, e);
+	default:
+		buffer_free(&m);
+		error("%s: unexpected response from master 0x%08x",
+		    __func__, type);
+		return -1;
+	}
+
+	signal(SIGHUP, control_client_sighandler);
+	signal(SIGINT, control_client_sighandler);
+	signal(SIGTERM, control_client_sighandler);
+	signal(SIGWINCH, control_client_sigrelay);
+
+	/*
+	 * Stick around until the controlee closes the client_fd.
+	 */
+		buffer_clear(&m);
+	if (mux_client_read_packet(fd, &m) != 0) {
+		if (errno == EPIPE ||
+		    (errno == EINTR && muxclient_terminate != 0))
+			return 0;
+		fatal("%s: mux_client_read_packet: %s",
+		    __func__, strerror(errno));
+	}
+	type = buffer_get_int(&m);
+	if (type == MUX_S_FAILURE || type == MUX_S_PERMISSION_DENIED) {
+		e = buffer_get_string(&m, NULL);
+		fatal("%s: master returned error: %s", __func__, e);
+	} else
+		fatal("%s: master returned unexpected message %u",
+		    __func__, type);
+}
+
+/* Multiplex client main loop. */
+void
+muxclient(const char *path)
+{
+	struct sockaddr_un addr;
+	int sock;
+	u_int pid;
+
+	if (muxclient_command == 0) {
+		if (stdio_forward_host != NULL)
+			muxclient_command = SSHMUX_COMMAND_STDIO_FWD;
+		else
+			muxclient_command = SSHMUX_COMMAND_OPEN;
+	}
+
+	switch (options.control_master) {
+	case SSHCTL_MASTER_AUTO:
+	case SSHCTL_MASTER_AUTO_ASK:
+		debug("auto-mux: Trying existing master");
+		/* FALLTHROUGH */
+	case SSHCTL_MASTER_NO:
+		break;
+	default:
+		return;
+	}
+
+	memset(&addr, '\0', sizeof(addr));
+	addr.sun_family = AF_UNIX;
+	addr.sun_len = offsetof(struct sockaddr_un, sun_path) +
+	    strlen(path) + 1;
+
+	if (strlcpy(addr.sun_path, path,
+	    sizeof(addr.sun_path)) >= sizeof(addr.sun_path))
+		fatal("ControlPath too long");
+
+	if ((sock = socket(PF_UNIX, SOCK_STREAM, 0)) < 0)
+		fatal("%s socket(): %s", __func__, strerror(errno));
+
+	if (connect(sock, (struct sockaddr *)&addr, addr.sun_len) == -1) {
+		switch (muxclient_command) {
+		case SSHMUX_COMMAND_OPEN:
+		case SSHMUX_COMMAND_STDIO_FWD:
+			break;
+		default:
+			fatal("Control socket connect(%.100s): %s", path,
+			    strerror(errno));
+		}
+		if (errno == ENOENT)
+			debug("Control socket \"%.100s\" does not exist", path);
+		else {
+			error("Control socket connect(%.100s): %s", path,
+			    strerror(errno));
+		}
+		close(sock);
+		return;
+	}
+	set_nonblock(sock);
+
+	if (mux_client_hello_exchange(sock) != 0) {
+		error("%s: master hello exchange failed", __func__);
+		close(sock);
+		return;
+	}
+
+	switch (muxclient_command) {
+	case SSHMUX_COMMAND_ALIVE_CHECK:
+		if ((pid = mux_client_request_alive(sock)) == 0)
+			fatal("%s: master alive check failed", __func__);
+		fprintf(stderr, "Master running (pid=%d)\r\n", pid);
+		exit(0);
+	case SSHMUX_COMMAND_TERMINATE:
+		mux_client_request_terminate(sock);
+		fprintf(stderr, "Exit request sent.\r\n");
+		exit(0);
+	case SSHMUX_COMMAND_OPEN:
+		if (mux_client_request_forwards(sock) != 0) {
+			error("%s: master forward request failed", __func__);
+			return;
+		}
+		mux_client_request_session(sock);
+		return;
+	case SSHMUX_COMMAND_STDIO_FWD:
+		mux_client_request_stdio_fwd(sock);
+		exit(0);
+	default:
+		fatal("unrecognised muxclient_command %d", muxclient_command);
+	}
 }
Index: nchan.c
===================================================================
RCS file: /cvs/src/usr.bin/ssh/nchan.c,v
retrieving revision 1.62
diff -u -p -r1.62 nchan.c
--- nchan.c	7 Nov 2008 18:50:18 -0000	1.62
+++ nchan.c	18 Jan 2010 04:20:35 -0000
@@ -159,7 +159,7 @@ chan_ibuf_empty(Channel *c)
 	switch (c->istate) {
 	case CHAN_INPUT_WAIT_DRAIN:
 		if (compat20) {
-			if (!(c->flags & CHAN_CLOSE_SENT))
+			if (!(c->flags & (CHAN_CLOSE_SENT|CHAN_LOCAL)))
 				chan_send_eof2(c);
 			chan_set_istate(c, CHAN_INPUT_CLOSED);
 		} else {
@@ -276,9 +276,12 @@ static void
 chan_rcvd_close2(Channel *c)
 {
 	debug2("channel %d: rcvd close", c->self);
-	if (c->flags & CHAN_CLOSE_RCVD)
-		error("channel %d: protocol error: close rcvd twice", c->self);
-	c->flags |= CHAN_CLOSE_RCVD;
+	if (!(c->flags & CHAN_LOCAL)) {
+		if (c->flags & CHAN_CLOSE_RCVD)
+			error("channel %d: protocol error: close rcvd twice",
+			    c->self);
+		c->flags |= CHAN_CLOSE_RCVD;
+	}
 	if (c->type == SSH_CHANNEL_LARVAL) {
 		/* tear down larval channels immediately */
 		chan_set_ostate(c, CHAN_OUTPUT_CLOSED);
@@ -300,11 +303,13 @@ chan_rcvd_close2(Channel *c)
 		chan_set_istate(c, CHAN_INPUT_CLOSED);
 		break;
 	case CHAN_INPUT_WAIT_DRAIN:
-		chan_send_eof2(c);
+		if (!(c->flags & CHAN_LOCAL))
+			chan_send_eof2(c);
 		chan_set_istate(c, CHAN_INPUT_CLOSED);
 		break;
 	}
 }
+
 void
 chan_rcvd_eow(Channel *c)
 {
@@ -452,6 +457,10 @@ chan_is_dead(Channel *c, int do_send)
 		    c->self, c->efd, buffer_len(&c->extended));
 		return 0;
 	}
+	if (c->flags & CHAN_LOCAL) {
+		debug2("channel %d: is dead (local)", c->self);
+		return 1;
+	}		
 	if (!(c->flags & CHAN_CLOSE_SENT)) {
 		if (do_send) {
 			chan_send_close2(c);
Index: ssh.c
===================================================================
RCS file: /cvs/src/usr.bin/ssh/ssh.c,v
retrieving revision 1.331
diff -u -p -r1.331 ssh.c
--- ssh.c	11 Jan 2010 01:39:46 -0000	1.331
+++ ssh.c	18 Jan 2010 04:20:35 -0000
@@ -306,6 +306,11 @@ main(int ac, char **av)
 			options.gateway_ports = 1;
 			break;
 		case 'O':
+			if (stdio_forward_host != NULL)
+				fatal("Cannot specify multiplexing "
+				    "command with -W");
+			else if (muxclient_command != 0)
+				fatal("Multiplexing command already specified");
 			if (strcmp(optarg, "check") == 0)
 				muxclient_command = SSHMUX_COMMAND_ALIVE_CHECK;
 			else if (strcmp(optarg, "exit") == 0)
@@ -382,6 +387,10 @@ main(int ac, char **av)
 			}
 			break;
 		case 'W':
+			if (stdio_forward_host != NULL)
+				fatal("stdio forward already specified");
+			if (muxclient_command != 0)
+				fatal("Cannot specify stdio forward with -O");
 			if (parse_forward(&fwd, optarg, 1, 0)) {
 				stdio_forward_host = fwd.listen_host;
 				stdio_forward_port = fwd.listen_port;
@@ -883,11 +892,18 @@ static int
 client_setup_stdio_fwd(const char *host_to_connect, u_short port_to_connect)
 {
 	Channel *c;
+	int in, out;
 
 	debug3("client_setup_stdio_fwd %s:%d", host_to_connect,
 	    port_to_connect);
-	if ((c = channel_connect_stdio_fwd(host_to_connect, port_to_connect))
-	    == NULL)
+
+	in = dup(STDIN_FILENO);
+	out = dup(STDOUT_FILENO);
+	if (in < 0 || out < 0)
+		fatal("channel_connect_stdio_fwd: dup() in/out failed");
+
+	if ((c = channel_connect_stdio_fwd(host_to_connect, port_to_connect,
+	    in, out)) == NULL)
 		return 0;
 	channel_register_cleanup(c->self, client_cleanup_stdio_fwd, 0);
 	return 1;
Index: PROTOCOL.mux
===================================================================
RCS file: PROTOCOL.mux
diff -N PROTOCOL.mux
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ PROTOCOL.mux	18 Jan 2010 04:20:36 -0000
@@ -0,0 +1,168 @@
+This document describes the multiplexing protocol used by ssh(1)'s
+ControlMaster connection-sharing.
+
+1. Connection setup
+
+When a multiplexing connection is made to a ssh(1) operating as a
+ControlMaster from a ssh(1) in multiplex slave mode, the first
+action of each is to exchange hello messages:
+
+	uint32	MUX_MSG_HELLO
+	uint32  protocol version
+	string  extension name [optional]
+	string  extension value [optional]
+	...
+
+The current version of the mux protocol is 4. A slave should refuse
+to connect to a master that speaks an unsupported protocol version.
+Following the version identifier are zero or more extensions
+represented as a name/value pair. No extensions are currently
+defined.
+
+2. Opening sessions
+
+To open a new multiplexed session, a client may send the following
+request:
+
+	uint32	MUX_C_MSG_NEW_SESSION
+	bool	want tty flag
+	bool	want X11 forwarding flag
+	bool	want agent flag
+	bool	subsystem flag
+	uint32	escape char
+	string	terminal type
+	string	command
+	string	environment string 0 [optional]
+	...
+
+To disable the use of an escape character, "escape char" may be set
+to 0xffffffff. "terminal type" is generally set to the value of
+$TERM. zero or more environment strings may follow the command.
+
+The client then sends its standard input, output and error file
+descriptors (in that order) using Unix domain socket control messages.
+
+The server will then reply with MUX_S_OK, MUX_S_PERMISSION_DENIED
+or MUX_S_FAILURE.
+
+Once the server has received the fds, it will respond with MUX_S_OK
+indicating that the session is up. The client now waits for the
+session to end. When it does, the server will send an exit status
+message:
+
+	uint32	MUX_S_EXIT_MESSAGE
+	uint32	exit value
+
+The client should exit with this value to mimic the behaviour of a
+non-multiplexed ssh(1) connection. Two additional cases that the
+client must cope with are it receiving a signal itself and the
+server disconnecting without sending an exit message.
+
+3. Health checks
+
+The client may request a health check/PID report from a server:
+
+	uint32	MUX_C_ALIVE_CHECK
+
+The server replies with:
+
+	uint32	MUX_S_ALIVE
+	uint32	server pid
+
+4. Remotely terminating a master
+
+A client may request that a master terminate immediately:
+
+	uint32	MUX_C_TERMINATE
+
+The server will reply with one of MUX_S_OK or MUX_S_PERMISSION_DENIED.
+
+5. Requesting establishment of port forwards
+
+A client may request the master to establish a port forward:
+
+	uint32	MUX_C_OPEN_FORWARD
+	uint32	forwarding type
+	string	listen host
+	string	listen port
+	string	connect host
+	string	connect port
+
+forwarding type may be MUX_FWD_LOCAL, MUX_FWD_REMOTE, MUX_FWD_DYNAMIC.
+
+A server may reply with a MUX_S_OK, a MUX_S_PERMISSION_DENIED or a
+MUX_S_FAILURE.
+
+5. Requesting closure of port forwards
+
+A client may request the master to establish a port forward:
+
+	uint32	MUX_C_OPEN_FORWARD
+	uint32	forwarding type
+	string	listen host
+	string	listen port
+	string	connect host
+	string	connect port
+
+forwarding type may be MUX_FWD_LOCAL, MUX_FWD_REMOTE, MUX_FWD_DYNAMIC.
+
+A server may reply with a MUX_S_OK, a MUX_S_PERMISSION_DENIED or a
+MUX_S_FAILURE.
+
+6. Requesting stdio forwarding
+
+A client may request the master to establish a stdio forwarding:
+
+	uint32	MUX_C_NEW_STDIO_FWD
+	string	connect host
+	string	connect port
+
+The client then sends its standard input and output file descriptors
+(in that order) using Unix domain socket control messages.
+
+A server may reply with a MUX_S_OK, a MUX_S_PERMISSION_DENIED or a
+MUX_S_FAILURE.
+
+7. Status messages
+
+The MUX_S_OK message is empty:
+
+	uint32	MUX_S_OK
+
+The MUX_S_PERMISSION_DENIED and MUX_S_FAILURE include a reason:
+
+	uint32	MUX_S_PERMISSION_DENIED
+	string	reason
+
+	uint32	MUX_S_FAILURE
+	string	reason
+
+7. Protocol numbers
+
+#define MUX_MSG_HELLO		0x00000001
+#define MUX_C_NEW_SESSION	0x10000002
+#define MUX_C_ALIVE_CHECK	0x10000004
+#define MUX_C_TERMINATE		0x10000005
+#define MUX_C_OPEN_FORWARD	0x10000006
+#define MUX_C_CLOSE_FORWARD	0x10000007
+#define MUX_S_OK		0x80000001
+#define MUX_S_PERMISSION_DENIED	0x80000002
+#define MUX_S_FAILURE		0x80000003
+#define MUX_S_EXIT_MESSAGE	0x80000004
+#define MUX_S_ALIVE		0x80000005
+
+#define MUX_FWD_LOCAL	1
+#define MUX_FWD_REMOTE	2
+#define MUX_FWD_DYNAMIC	3
+
+XXX TODO
+XXX extended status (e.g. report open channels / forwards)
+XXX graceful close (delete listening socket, but keep existing sessions active)
+XXX lock (maybe)
+XXX watch in/out traffic (pre/post crypto)
+XXX inject packet (what about replies)
+XXX server->client error/warning notifications
+XXX port0 rfwd (need custom response message)
+XXX send signals via mux
+
+$OpenBSD$


More information about the openssh-unix-dev mailing list