conmux/conmux

#!/usr/bin/perl
#
# conmux -- the main console multiplexor daemon
# 
# Main console multiplexor daemon.  There is one of these daemons for
# each open console supported in the system.  Clients are directed to
# this daemon via the conmux-registry deamon.
#
# (C) Copyright IBM Corp. 2004, 2005, 2006
# Author: Andy Whitcroft <andyw@uk.ibm.com>
#
# The Console Multiplexor is released under the GNU Public License V2
#
use strict;

use FindBin;
use Symbol qw(gensym);
use IO::Socket;
use IO::Multiplex;
use IPC::Open3;
use URI::Escape;
use Net::Domain;

# Find our internal libraries.
use lib $FindBin::Bin;
use lib "$FindBin::Bin/../lib/";
use lib "$FindBin::Bin/lib/";
use Conmux;

our $P = 'conmux';
our $debug = 0;

$SIG{'CHLD'} = "IGNORE";

$| = 1;

#
# CALLBACK: this class is used to provide a timed callback.  The multiplexor
# libarary allows us to set a timeout on any open file we have registered.
# So, we open a new file descriptor to /dev/null and set a timeout on that.
#
package Callback;

sub new {
	my ($class, $mux, $who, $time) = @_;
	my $self = bless { 'who' => $who }, $class;

	my ($fh);

	print "Callback::new [$self] mux<$mux> who<$who> time<$time>\n"
		if ($main::debug);

	# Open a file handle to nothing, we need this to hang the timeout
	# on in the multiplexor.  It will fail with a mux_eof, which we ignore.
	open($fh, "</dev/null") || die "$P: /dev/null: open failed - $!\n";

	$mux->add($fh);
	$mux->set_callback_object($self, $fh);

	$mux->set_timeout($fh, $time);

	$self;
}
sub mux_timeout {
	my ($self, $mux, $fh) = @_;

	print "Callback::mux_timeout [$self] mux<$mux> fh<$fh>\n"
		if ($main::debug);

	$self->{'who'}->callback_timeout();

	$mux->close($fh);
}
sub DESTROY {
	my ($self) = @_;
	print "Callback::DESTROY [$self]\n" if ($main::debug);
}

#
# LISTENER SOCKET: creates an intenet listener for new clients and
# connects them to the junction provided.
#
package ListenerSocket;

sub new {
	my ($class, $mux, $port) = @_;
	my $self = bless { 'mux' => $mux }, $class;

	print "ListenerSocket::new [$self] mux<$mux> port<$port>\n"
		if ($main::debug);

	$self->initialise($port);

	$self;
}

sub initialise {
	my ($self, $port) = @_;
	my ($sock);

	print "ListenerSocket::initialise [$self] port<$port> "
		if ($main::debug);

	# Create a listening socket and add it to the multiplexor.
	my $sock = new IO::Socket::INET(Proto     => 'tcp',
					LocalPort => $port,
					Listen    => 4,
					ReuseAddr => 1)
		or die "socket: $@";

	print "  adding $self $sock\n" if ($main::debug);
	$self->mux->listen($sock);
	$self->mux->set_callback_object($self, $sock);
	$self->listener($sock);
}

# DATA accessors.
sub mux {
	my $self = shift;
	if (@_) { $self->{'mux'} = shift }
	return $self->{'mux'};
}
sub listener {
	my $self = shift;
	if (@_) { $self->{'listener'} = shift }
	return $self->{'listener'};
}

sub address {
	my ($self) = @_;
	Net::Domain::hostfqdn() . ':' . $self->{'listener'}->sockport();
}

# JUNCTION: callbacks.
##sub junctionInput {
##}
##sub junctionEOF {
##	my ($self) = @_;
##	
##	$self->{'junction'}->junctionRemove($self, 'console-client');
##	$self->{'mux'}->close($self->{'listener'});
##}

# Handle new connections by instantiating a new client class.
sub mux_connection {
	my ($self, $mux, $fh) = @_;
	my ($client);

	print "ListenerSocket::mux_connection [$self] mux<$mux> fh<$fh>\n"
		if ($main::debug);

	# Make a new client connection.
	$client = ClientCmd->new($mux, $fh);
	print "  new connection $self $client\n" if ($main::debug);
}

sub DESTROY {
	my ($self) = @_;

	print "ListenerSocket::DESTROY [$self]\n" if ($main::debug);

	close($self->listener);
}

#
# JUNCTION: generic junction box object, connects names groups of objects
# to other named groups.
#
# Expects the following callbacks to be defined on each object registered:
#   junctionInput($from, $data)
#   junctionEOF($from, $to)
# 
package Junction;

sub new {
	my ($class) = @_;
	my $self = bless { }, $class;

	print "Junction::new [$self]\n" if ($main::debug);

	$self;
}

sub junctionAdd {
	my ($self, $client) = @_;

	print "Junction::junctionAdd [$self] client<$client>\n"
		if ($main::debug);

	# Add ourselves to the list of recipients.
	$self->{$client} = $client;
}

sub junctionInput {
	my ($self, $client, $data) = @_;
	my ($c);

	print "Junction::junctionInput [$self] client<$client> " .
		"data<$data>\n" if ($main::debug);

	# Send this data on to the clients listed in the output list.
	for $c (values %{$self}) {
		print "  sending to $c\n" if ($main::debug);
		$c->junctionInput($client, $data);
	}
}
sub junctionEOF {
	my ($self, $client) = @_;
	my ($c);

	print "Junction::junctionEOF [$self] client<$client>\n"
		if ($main::debug);

	# Send this eof on to the clients listed in the output list.
	for $c (values %{$self}) {
		print "  sending to $c\n" if ($main::debug);
		$c->junctionEOF($client);
	}
}
sub junctionRemove {
	my ($self, $client) = @_;

	print "Junction::junctionRemove [$self] client<$client>\n"
		if ($main::debug);

	# Drop this client from our lists.
	delete $self->{$client};
}

#
# PAYLOAD: generic payload object, connects itself to the requisite junction.
#
package Payload;

my %payloads = ();
my $payloads = 0;

sub lookup {
	my ($class, $name) = @_;

	$payloads{$name};
}

sub found {
	my ($class, $name, $self) = @_;

	print "Payloads::found name<$name> self<$self>\n" if ($main::debug);

	$payloads{$name} = $self;
	$payloads++;
}
sub lost {
	my ($class, $name, $self) = @_;

	print "Payloads::lost name<$name> self<$self>\n" if ($main::debug);

	undef $payloads{$name};
	if (--$payloads == 0) {
		exit(0);
	}
}

sub new {
	my ($class, $name, $title, $mux, @a) = @_;
	my $self = bless { }, $class;

	print "Payload::new [$self] name<$name> title<$title> mux<$mux>\n"
		if ($main::debug);

	Payload->found($name, $self);

	$self->name($name);
	$self->title($title);
	$self->mux($mux);
	$self->enabled(1);

	$self->cin(Junction->new);
	$self->cout(Junction->new);

	$self->initialise(@a);

	$self;
}

# Data accessors.
sub name {
	my $self = shift;
	if (@_) { $self->{'name'} = shift }
	return $self->{'name'};
}
sub title {
	my $self = shift;
	if (@_) { $self->{'title'} = shift }
	return $self->{'title'};
}
sub mux {
	my $self = shift;
	if (@_) { $self->{'mux'} = shift }
	return $self->{'mux'};
}
sub cin {
	my $self = shift;
	if (@_) { $self->{'cin'} = shift }
	return $self->{'cin'};
}
sub cout {
	my $self = shift;
	if (@_) { $self->{'cout'} = shift }
	return $self->{'cout'};
}
sub enabled {
	my $self = shift;
	if (@_) { $self->{'enabled'} = shift }
	return $self->{'enabled'};
}
sub connected {
	my $self = shift;
	if (@_) { $self->{'connected'} = shift }
	$self->transition();
	return $self->{'connected'};
}
sub transition {
	my $self = shift;
	my $time = time;
	if (($time - $self->{'trans_minor'}) > 30) {
		$self->{'trans_major'} = $time;
	}
	$self->{'trans_minor'} = $time;
}
sub retry_timeout {
	my $self = shift;
	my $time = time - $self->{'trans_major'};

	if ($time < 60) {
		return 1;
	} elsif ($time < 120) {
		return 10;
	} else {
		return 30;
	}
}
sub state {
	my $self = shift;
	my $ctime = $self->{'connected'};
	my $ttime = $self->{'trans_major'};
	my $time = time;

	if ($ctime && ($time - $ctime) > 30) {
		"connected";
	} elsif ($ttime && ($time - $ttime) < 60) {
		"transition";
	} else {
		"disconnected";
	}
}

sub initialise {
	my ($self) = @_;
	my ($sock);

	print "Payload::initialise [$self]\n" if ($main::debug);

	# Ensure we recieve client input.
	$self->cin->junctionAdd($self);

	$self->connected(time);
}

# Telnet constants.
my $TN_IAC      = sprintf("%c", 255);
my $TN_DONT     = sprintf("%c", 254);
my $TN_DO       = sprintf("%c", 253);
my $TN_WONT     = sprintf("%c", 252);
my $TN_WILL     = sprintf("%c", 251);
my $TN_SB       = sprintf("%c", 250);
my $TN_SE       = sprintf("%c", 240);
my $TN_BREAK    = sprintf("%c", 243);

my $TNOPT_ECHO	= sprintf("%c", 1);
my $TNOPT_SGA	= sprintf("%c", 3);

#
# If we get here then we have accumulated a complete telnet
# negotiation string.
#
# Telnet negotiation protocol - RFC#854:
#
# DO       We are being asked to DO an option
# DONT     We are being asked to NOT DO an option
# WILL     We are being told they will DO an option
# WONT     We are being told they will NOT DO an option
#
# DO/DONT requests indicate we should {en,dis}able a mode.
# We are expected to respond with WILL or WONT.  To prevent
# loops, we should not respond if the request matches our
# current mode.
#
# WILL/WONT requests indicate the other end would like to
# {en,dis}able a mode.  We are expected to respond with
# DO/DONT.
#
# If we want a particular mode {en,dis}abled then we may start
# negotiation of that mode with a WILL/WONT.
#
# We want the other end to perform echo by default so we will
# DO any request for ECHO and DONT all other requests.
#

sub mux_input {
	my ($self, $mux, $fh, $input) = @_;
	my ($client);

	print "Payload::mux_input [$self] mux<$mux> fh<$fh> input<$$input>\n"
		if ($main::debug);

	while ($$input ne "") {
		# Ordinary text.
		if ($$input =~ s/^([^$TN_IAC]+)//) {
			# Data coming in from the payload, this needs to go to
			# all of the clients.
			$self->cout->junctionInput($self, $1);
			next;
		}

		# IAC,SB,...,SE
		if ($$input =~ s/^$TN_IAC$TN_SB([^$TN_SE]+)$TN_SE//) {
			print "SB\n" if ($main::debug);
			next;
		}
		# IAC,[DO|DONT|WILL|WONT],<what>
		if ($$input =~ s/^$TN_IAC$TN_DO(.)//) {
			my $c = unpack("C", $1);
			print "DO<$c:$1>\n" if ($main::debug);
			# We are DONT on all options so WONT all requests.
			$self->junctionInput($self, "$TN_IAC$TN_WONT$1");
			next;
		}
		if ($$input =~ s/^$TN_IAC$TN_DONT(.)//) {
			my $c = unpack("C", $1);
			print "DONT<$c:$1>\n" if ($main::debug);
			# We are already DONT on all options, no reply.
			next;
		}
		if ($$input =~ s/^$TN_IAC$TN_WILL(.)//) {
			my $c = unpack("C", $1);
			print "WILL<$c:$1>\n" if ($main::debug);

			my $reply = $TN_DONT;
			if ($1 == $TNOPT_ECHO || $1 == $TNOPT_SGA) {
				$reply = $TN_DO;
			}
			$self->junctionInput($self, "$TN_IAC$reply$1");
			next;
		}
		if ($$input =~ s/^$TN_IAC$TN_WONT(.)//) {
			my $c = unpack("C", $1);
			print "WONT<$c:$1>\n" if ($main::debug);
			$self->junctionInput($self, "$TN_IAC$TN_DONT$1");
			next;
		}
		# IAC,<option>
		if ($$input =~ s/^$TN_IAC([^$TN_SB$TN_DO$TN_DONT$TN_WILL$TN_WONT])//) {
			print "OPTION<$1>\n" if ($main::debug);
			next;
		}

		# Incomplete ...
		if ($$input =~ /^$TN_IAC/) {
			return;
		}
	}
}
sub junctionInput {
	my ($self, $from, $data) = @_;
	my ($fh);

	print "Payload::junctionInput [$self] from<$from> data<$data>\n"
		if ($main::debug);

	##$self->{'mux'}->write($self->{'wfh'}, $data);
	# If we are connected ...
	if ($self->{'wfh'}) {
		$fh = $self->{'wfh'};
		print $fh $data;
	} else {
		$from->junctionInput($self, "<<<NOT CONNECTED>>>\n");
	}
}

sub mux_eof {
	my ($self, $mux, $fh) = @_;
	my ($client);

	print "Payload::mux_eof [$self] mux<$mux> fh<$fh>\n" if ($main::debug);

	# Check for a restartable connection.
	if ($self->can("restart")) {
		my ($timeout) = $self->retry_timeout();
		
		$self->cout->junctionInput($self, 
			"<<<PAYLOAD LOST ... retrying in $timeout secs>>>\n");

		# Schedule a timeout to trigger a reconnect.
		Callback->new($mux, $self, $timeout);

	} else {
		$self->cout->junctionEOF($self);
		$self->cin->junctionRemove($self);

		Payload->lost($self->name, $self);
	}

	# Close down the payload ...
	$mux->close($self->{'rfh'});
	##$mux->remove($self->{'wfh'});
}

sub mux_close {
	my ($self, $mux, $fh) = @_;

	$self->connected(0);

	#close($self->{'rfh'});
	close($self->{'wfh'});
	undef $self->{'rfh'};
	undef $self->{'wfh'};

	if ($self->{'pid'}) {
		# Kill the process group for this pid.
		kill 1, 0 - $self->{'pid'};
		undef $self->{'pid'};
	}
}

sub callback_timeout {
	my ($self) = @_;

	print "Payload::callback_timeout [$self]\n" if ($main::debug);

	if ($self->enabled) {
		$self->cout->junctionInput($self, "<<<PAYLOAD RESTART>>>\n");
		$self->openPayload();
	} else {
		$self->cout->junctionInput($self, "<<<PAYLOAD DISABLED>>>\n");
	}
}

sub closePayload {
	my ($self) = @_;

	if ($self->connected) {
		$self->cout->junctionInput($self, "<<<PAYLOAD CLOSED>>>\n");
		
		# Close down the payload ...
		$self->mux->close($self->{'rfh'});
	}
	if ($self->enabled) {
		$self->enabled(0);
		return 1;
	} else {
		return 0;
	}
}
sub openPayload {
	my ($self) = @_;

	$self->enabled(1);
	if (!$self->connected) {
		if ($self->can("restart")) {
			$self->restart();

			return 1;
		}
	}
	return 0;
}

sub helpAdd {
	my ($self, $cmd, $msg) = @_;

	push(@{$self->{'help'}}, [ $cmd, $msg ]);
}

sub commandHelp {
	my ($self) = @_;
	my @entries = (
		[ 'break', 'send a break sequence' ]
	);
	
	if (defined $self->{'help'}) {
		( @entries, @{$self->{'help'}} );

	} else {
		@entries;
	}
}
sub commandAdd {
	my ($self, $cmd, @a) = @_;

	$self->{'cmd'}->{$cmd} = [ @a ];
}
sub commandExec {
	my ($self, $client, $cmd, $a) = @_;
	my ($exe);

	print "Payload::commandExec [$self] client<$client> cmd<$cmd> a<$a>\n"
		if ($main::debug);

	$exe = $self->{'cmd'}->{$cmd};

	if ($cmd eq "break") {
		# Send a telnet break ...
		$self->junctionInput($self, "$TN_IAC$TN_BREAK");
		return;

	} elsif ($cmd eq "close") {
		if (!$self->enabled && !$self->connected) {
			$client->junctionInput($self,
				"console already closed\n");

		} elsif ($self->closePayload()) {
			$self->cout->junctionInput($self, "(" . $client->id .
				") triggered a console close\n");

		} else {
			$client->junctionInput($self, "ERROR: close failed\n");
		}
		return;

	} elsif ($cmd eq "open") {
		if ($self->connected) {
			$client->junctionInput($self, "console already open\n");

		} elsif ($self->openPayload()) {
			$self->cout->junctionInput($self, "(" . $client->id .
				") triggered a console open\n");

		} else {
			$client->junctionInput($self, "open failed\n");
		}
		return;
	} 

	# Ensure we error if we have no command.
	if (!$exe) {
		$client->junctionInput($self, "Command not recognised\n");
		return;
	} 

	my ($msg, $run) = @{$exe};
	if ($msg ne '') {
		$self->cout->junctionInput($self, "(" . $client->id .
			") $msg\n");
	}

	local(*IN, *OUT, *ERR);
	my ($cmd, @args) = split(m/'(.*?)'|"(.*?)"|\s(.*?)\s/g, $run . " $a");
	my @opts;
	for (my $i=0; $i < @args; $i++) {
		next if not $args[$i];
		push(@opts, $args[$i]);
	}
	my $pid = IPC::Open3::open3(*IN, *OUT, *ERR, $cmd, @opts);
	close(*IN{IO});

	# XXX: this should not be happening here.
	$self->mux->add(*OUT{IO});
	my $data = ClientData->new($self->mux, *OUT{IO});
	$data->{'id'} = "cmd:$cmd stdout";

	$data->payload($self);
	$data->cout($self->cout);

	# XXX: this should not be happening here.
	$self->mux->add(*ERR{IO});
	my $data = ClientData->new($self->mux, *ERR{IO});
	$data->{'id'} = "cmd:$cmd stderr";

	$data->payload($self);
	$data->cout($client);
}

sub DESTROY {
	my ($self) = @_;

	print "Payload::DESTROY [$self]\n" if ($main::debug);
}

#
# PAYLOAD APPLICATION: handles forking off a command as a payload.
#
package PayloadApplication;
use base 'Payload';

sub initialise {
	my ($self, $cmd) = @_;
	my ($pid, $wfh, $rfh);

	print "PayloadApplication::initialise [$self] cmd<$cmd>" 
	if ($main::debug);

	$self->SUPER::initialise();

	# XXX: we cannot use the write buffering offered by the mux package
	# without suffering a read error from the PWR file handle, there
	# is no a way to add a write-only channel.

	$self->{'args'} = $cmd;

	# Start the payload ...
	$pid = IPC::Open3::open3($wfh, $rfh, 0, "setsid " . $cmd);

	$self->{'rfh'} = $rfh;
	$self->{'wfh'} = $wfh;
	$self->{'pid'} = $pid;

	$self->mux->add($rfh);
	##$mux->add($wfh);

	$self->mux->set_callback_object($self, $rfh);
	##$mux->set_callback_object($self, $wfh);

	print "SHARE PAYLOAD: $self $wfh/$rfh (to $cmd) [fd=" .
		fileno($wfh) . "/" . fileno($rfh) . "]\n" if ($main::debug);
	print "payload '$cmd' on fd=" . fileno($wfh) . "/" .
		fileno($rfh) . "\n";
	
	$self;
}

sub restart {
	my ($self) = @_;

	$self->initialise($self->{'args'});
}

#
# PAYLOAD SOCKET: handles a network socket as payload.
#
package PayloadSocket;
use base 'Payload';

sub initialise {
	my ($self, $addr) = @_;
	my ($payload);

	print "PayloadSocket::initialise [$self] addr<$addr>\n"
		if ($main::debug);

	$self->SUPER::initialise();

	$self->{'args'} = $addr;

	# Create a listening socket and add it to the multiplexor.
	my $payload = new IO::Socket::INET(PeerAddr => $addr);
	if (!$payload) {
		$self->connected(0);
		if ($self->can("restart")) {
			my ($timeout) = $self->retry_timeout();

			$self->cout->junctionInput($self, 
				"<<<PAYLOAD ERROR ($!) ... retrying in $timeout secs>>>\n");
			# Schedule a timeout to trigger a reconnect.
			Callback->new($self->mux, $self, $timeout);
		} else {
			$self->cout->junctionEOF($self);
			$self->cin->junctionRemove($self);

			Payload->lost($self->name, $self);
		}

	} else {
		$self->{'rfh'} = $payload;
		$self->{'wfh'} = $payload;

		print "SHARE PAYLOAD: $self $payload (to $addr) [fd=" .
			fileno($payload) . "]\n" if ($main::debug);
		print "payload '$addr' on fd=" . fileno($payload) . "\n";
		$self->mux->add($payload);

		$self->mux->set_callback_object($self, $payload);
	}

	print "SHARE PAYLOAD: $self $payload ... done\n" if ($main::debug);

	$self;
}

sub restart {
	my ($self) = @_;

	$self->initialise($self->{'args'});
}

#
# CLIENT: general client object, represents a remote client channel
#
package Client;

sub new {
	my ($class, $mux, $fh) = @_;
	my $self = bless { 'mux' => $mux,
			   'fh'  => $fh }, $class;

	print "Client::new [$self] mux<$mux> fh<$fh>\n"
		if ($main::debug);

	$self->initialise();

	$self;
}

sub clone {
	my ($class, $from) = @_;

	my $self = bless { %{$from} }, $class;

	print "Client::clone [$self] from<$from>\n" if ($main::debug);

	$self->initialise();

	$self;
}

# Data accessors.
sub mux {
	my $self = shift;
	if (@_) { $self->{'mux'} = shift }
	return $self->{'mux'};
}
sub payload {
	my $self = shift;
	if (@_) { $self->{'payload'} = shift }
	return $self->{'payload'};
}
sub fh {
	my $self = shift;
	if (@_) { $self->{'fh'} = shift }
	return $self->{'fh'};
}
sub id {
	my $self = shift;
	if (@_) { $self->{'id'} = shift }
	return $self->{'id'};
}
sub announce {
	my $self = shift;
	if (@_) { $self->{'announce'} = shift }
	return $self->{'announce'};
}
sub cout {
	my $self = shift;
	if (@_) { $self->{'cout'} = shift }
	return $self->{'cout'};
}
sub cin {
	my $self = shift;
	if (@_) {
		$self->{'cin'}->junctionRemove($self) if ($self->{'cin'});
		$self->{'cin'} = shift;
		$self->{'cin'}->junctionAdd($self) if ($self->{'cin'} != undef);
	}
	return $self->{'cin'};
}

sub initialise {
	my ($self) = @_;

	print "Client::initialise [$self]\n" if ($main::debug);

	$self->mux->set_callback_object($self, $self->fh);
}

sub junctionInput {
	my ($self, $from, $data) = @_;

	print "Client::junctionInput [$self] data<$data>\n" if ($main::debug);

	$self->mux->write($self->fh, $data);
}
sub junctionEOF {
	my ($self, $from, $data) = @_;

	print "Client::junctionEOF [$self] data<$data>\n" if ($main::debug);

	$self->shutdown();
}

sub mux_eof {
	my ($self, $mux, $fh, $input) = @_;

	print "Client::mux_eof [$self] mux<$mux> fh<$fh> input<$input>\n"
		if ($main::debug);

	# Handle any pending input, then remove myself from the clients list.
	$self->mux_input($mux, $fh, $input);
	$self->cin(undef);
	$self->cout(undef);

	# Tell the multiplexor we no longer are using this channel.
	$mux->shutdown($fh, 1);
}
sub mux_close {
	my ($self, $mux, $fn) = @_;

	print "Client::close [$self]\n" if ($main::debug);

	if ($self->announce) {
		$self->announce->junctionInput($self, "(" . $self->id .
			") disconnected\n");
	}
	print "$self->{'id'} disconnected\n";
}

sub shutdown {
	my ($self) = @_;

	print "Client::shutdown [$self]\n" if ($main::debug);

	# Close myself down and tell the payload.
	$self->mux->shutdown($self->fh, 2);
}
sub DESTROY {
	my ($self) = @_;

	print "Client::DESTROY [$self]\n" if ($main::debug);
}

#
# CLIENT CMD: represents a client whilst in command mode, when we have commited
# to connecting this will pass the client connection off to a ClientData
# object.
#
package ClientCmd;
use base 'Client';

sub mux_input {
	my ($self, $mux, $fh, $input) = @_;

	print "Client::shutdown [$self] mux<$mux> fh<$fh> input<$$input>\n"
		if ($main::debug);

        while ($$input =~ s/^(.*?)\n//) {
		my ($cmd, $args) = split(' ', $1, 2);
		my (%args) = Conmux::decodeArgs($args);

		my $reply = {
			'status' => 'ENOSYS unknown command',
		};

		# XXX: check authentication if required and reject the
		# command out of hand - leak _nothing_.
		if (!defined $args{'id'}) {
			$reply->{'status'} = 'EACCES identifier required';
			goto reply;
		}
		# They are who they say they are, record who that is.
		$self->{'id'} = $args{'id'};
		
		if ($cmd eq "CONNECT") {
			# Switch over to data mode, hand this connection off
			# to a data client instance, I am done.
			my ($data, $to, $in, $out);
			$data = ClientData->clone($self);

			$to = $args{'to'};
			if (!$to) {
				$reply->{'status'} = "EINVAL CONNECT " .
					" requires 'to' specifier";
				goto reply;
			}
			my $payload = Payload->lookup($to);
			if (!defined $payload) {
				$reply->{'status'} = "EINVAL '$to' not a " .
					" valid destination specifier";
				goto reply;
			}

			$reply->{'status'} = 'OK';

			# Get the payload title and pass that back.
			$reply->{'title'} = $payload->title . ' [channel ' .
				$payload->state() . ']';
			$reply->{'state'} = $payload->state();
			# Get connected clients and pass back as the motd
			for my $cl (keys(%{$payload->cout})) {
				$reply->{'motd'} .= '(' . $payload->cout->{$cl}->id;
				$reply->{'motd'} .= ") is already connected\n";
			}

			$data->payload($payload);
			$args{'type'} = 'client' if (!$args{'type'});
			if ($args{'type'} eq 'status') {
				$data->cout($payload->cout);
			} elsif ($args{'type'} eq 'client') {
				if (!$args{'hide'}) {
					$data->announce($payload->cout);
					$payload->cout->junctionInput(
						$self,  "(" . $self->id .
						") connected\n");
				}
				$data->cin($payload->cout);
				$data->cout($payload->cin);
			} else {
				$reply->{'status'} = "EINVAL '$args{'type'}' " .
					"not a valid destination type";
				goto reply;
			}

			print "$self->{'id'} connected to $to/$args{'type'}\n";

			$self->junctionInput($self, 
				Conmux::encodeArgs($reply) .  "\n");

			# Don't handle any more input - its not going to be
			# for us.
			last;
		}

	reply:
		# We're done, send back our response to this.
		$self->junctionInput($self, Conmux::encodeArgs($reply) . "\n");
	}
}

#
# CLIENT DATA: handles a client connection when in data mode, attaches
# the client connection to the relevant junction.
#
package ClientData;
use base 'Client';

my @help = (
	[ 'msg', 'send a message to all connected clients' ],
	[ 'quit', 'disconnect from the console' ],
);
sub mux_input {
	my ($self, $mux, $fh, $input) = @_;

	print "ClientData::mux_input [$self] mux<$mux> fh<$fh> input<$$input>\n"
		if ($main::debug);

	while ($$input ne "") {
		if ($self->{'cmd'} eq '') {
			# Check for an incomplete escape ... wait for more.
			if ($$input =~ /^~$/s) {
				return;
			}
			if ($$input =~ s/^~\$//s) {
				$self->{'cmd'} = '>';
				my $title = $self->payload->title;
				$self->junctionInput($self, "\r\nCommand($title)> ");
				next;
			}
			# Its not an escape ... pass it on.
			# Ship anything before that cannot be the escape.
			if ($$input =~ s/^(.[^~]*)(~|$)/\2/s) {
				# Data coming in from the client, send it to
				# the payload.
				$self->cout->junctionInput($self, $1);
			}
		} else {
			# Consume characters upto a newline, echo them back
			# to the client as we go.
			while ($$input =~ s/^([^\r\n])//) {
				my $c = $1;
				if ($c eq "\b" || $c eq "\x7f") {
					if (length($self->{'cmd'}) > 1) {
						$c = "\b \b";
						substr($self->{'cmd'},
							-1, 1, '');
					} else {
						$c = '';
					}
				} else {
					$self->{'cmd'} .= $c;
				}
				$self->junctionInput($self, $c);
			}
			# If we arn't at a newline, then wait for more input.
			if ($$input !~ s/^[\r\n]+//) {
				return;
			}

			$self->junctionInput($self, "\n");

			my ($cmd, $a) = split(' ', substr($self->{'cmd'},
				1), 2);
			$self->{'cmd'} = '';

			if ($cmd eq '') {

			} elsif ($cmd eq 'help') {
				my @cmds = $self->payload->commandHelp();

				my $ent;
				my $help = "Conmux commands:\n";
				for $ent (@cmds, @help) {
					$help .= sprintf("  %-20s %s\n",
						$ent->[0], $ent->[1]);
				}
				$self->junctionInput($self, $help);

			} elsif ($cmd eq 'quit') {
				$self->shutdown();

			} elsif ($cmd eq 'msg') {
				$self->cin->junctionInput($self,
					"($self->{'id'}) $a\n");

			# Not a client command ... pass it to the payload.
			} else {
				$self->payload->commandExec($self, $cmd, $a);
			}
		}
	}
}

#
# LIBRARY: split a string honouring quoting.
#
package main;
sub parse($) {
    my ($str) = @_;

    my ($pos, @args, $argc, $quote, $real, $c, $inc);

    $inc = 0;
    @args = ();
    $argc = 0;
    $quote = 0;
    $real = 0;


    $pos = 0;
    while (substr($str, $pos, 1) eq " ") {
	$pos++;
    }
    for (; $pos < length($str); $pos++) {
        $c = substr($str, $pos, 1);
        if ($quote != 2 && $c eq '\\') {
            $real = 1;
            $pos++;
            $c = substr($str, $pos, 1);
        } else {
            $real = 0;
        }

        if ($quote != 2 && $c eq '"' && !$real) {
            $quote ^= 1;
        } elsif ($quote != 1 && $c eq "'" && !$real) {
            $quote ^= 2;
        } elsif ($c eq " " && $quote == 0 && !$real) {
	    while (substr($str, $pos, 1) eq " ") {
		$pos++;
	    }
	    $pos--;
	    $argc++;
        } else {
	    if ($inc) {
		$inc = 0;
		$argc++;
	    }
            $args[$argc] .= $c;
        }
    }

    @args;
}

#
# MAIN: makes the IO multiplexor, junction, listener and payload and stitches
# them all together.
#
package main;

# Usage checks.
if ($#ARGV != 0 && $#ARGV != 3) {
	print STDERR "Usage: $P <config file>\n";
	print STDERR "       $P <local port> <title> socket <host>:<port>\n";
	print STDERR "       $P <local port> <title> cmd <cmd>\n";
	exit 1
}
my @conf;
if ($#ARGV == 3) {
	my ($lport, $title, $what, $arg) = @ARGV;
	@conf = (
		"listener '$lport'",
		"'$what' console '$title' '$arg'"
	);
} else {
	my ($cf) = @ARGV;
	open(CONF, '<', $cf) || die "$P: $cf: open failed - $!\n";
	@conf = <CONF>;
	close(CONF);
}

# Make a new multiplexer.
my $mux  = new IO::Multiplex;

my ($line, $seg, $listener, $payload);
$line = '';
for $seg (@conf) {
	# Handle comments, blank lines and line continuation.
	chomp($seg); $seg =~ s/^\s+//;
	next if ($seg =~ /^#/);
	$line .= $seg;
	if ($line =~ m/\\$/) {
		chop($line);
		next;
	}
	next if (/^\s+$/);

	my ($cmd, @a) = parse($line);
	$line = '';

	if ($cmd eq "listener") {
		if ($#a != 0) {
			warn "$P: Usage: listener <port>\n" .
			     "$P:        $line\n";
			next;
		}

		my ($lport) = @a;
		my ($rhost, $rname);

		# port
		if ($lport =~ m@^\d+$@) {
			# Already in the right format.
			
		# registry/service
		} elsif ($lport =~ m@(.*)/(.*)@) {
			($rhost, $rname, $lport) = ($1, $2, 0);

		# service
		} else {
			($rhost, $rname, $lport) = ('-', $lport, 0);
		}

		# Create the client listener socket.
		$listener = ListenerSocket->new($mux, $lport);

		# Register us with the registry.
		if ($rhost) {
			Conmux::Registry::add($rhost, $rname, $listener->address);
		}

	} elsif ($cmd eq 'socket') {
		if ($#a != 2) {
			warn "$P: Usage: socket <name> <title> <host:port>\n" .
			     "$P:        $line\n";
			next;
		}
		my ($name, $title, $sock) = @a;

		# Create the payload.
		$payload = PayloadSocket->new($name, $title, $mux, $sock);

	} elsif ($cmd eq 'application') {
		if ($#a != 2) {
			warn "$P: Usage: application <name> <title> <host:port>\n" .
			     "$P:        $line\n";
			next;
		}
		my ($name, $title, $app) = @a;

		$payload = PayloadApplication->new($name, $title, $mux, $app);

	} elsif ($cmd eq 'command') {
		if ($#a != 2) {
			warn "$P: Usage: command <name> <msg> <cmd>\n" .
			     "$P:        $line\n";
			next;
		}
		my ($name, $msg, $cmd) = @a;

		$payload->commandAdd($name, $msg, $cmd);
	
	} elsif ($cmd eq 'help') {
		if ($#a != 1) {
			warn "$P: Usage: $cmd <name> <msg>\n" .
			     "$P:        $line\n";
			next;
		}
		my ($name, $msg) = @a;

		$payload->helpAdd($name, $msg);
	} else {
		warn "$P: $cmd: unknown configuration command\n";
	}
}

# Hand over to the multiplexor.
do {
	eval { $mux->loop; };
	warn "$@";
} while ($@ =~ /^Use of freed value in iteration/);
die "ERROR: $@\n";