Merge branch 'python23' - python 2.3 compatibility

* python23:
  Oops, missed another << operator to replace with _shl().
  socket.SHUT_RD and socket.SHUT_WR don't exist in python 2.3.
  compat/ssubprocess.py: some python versions don't have os.closerange().
  _nb_clean: don't catch EPIPE after all.
  Fix busy-waiting in two situations:
  Factor out common mainloop code between client and server.
  Implement our own left-shift operator to shut up python 2.3 warnings.
  Don't use set() since it's not in python 2.3.
  import and use subprocess.py from python 2.6.
  Remove list comprehensions for python 2.3 compatibility.

README.md

@@ -1,14 +1,10 @@
 sshuttle: where transparent proxy meets VPN meets ssh
 =====================================================
 
-I just spent an afternoon working on a new kind of VPN.  You can get
-the first release, <a href="http://github.com/apenwarr/sshuttle">sshuttle
-0.10, on github</a>.
-
 As far as I know, sshuttle is the only program that solves the following
 common case:
 
- - Your client machine (or router) is Linux.
+ - Your client machine (or router) is Linux, FreeBSD, or MacOS.
 
  - You have access to a remote network via ssh.
 
@@ -53,10 +49,8 @@ This is how you use it:
 -----------------------
 
  - <tt>git clone git://github.com/apenwarr/sshuttle</tt>
-    on your client and server machines. The server can be
-    any ssh server with python available; the client must
-    be Linux with iptables, and you'll need root or sudo
-    access.
+    on your client machine. You'll need root or sudo
+    access, and python needs to be installed.
 
  - <tt>./sshuttle -r username@sshserver 0.0.0.0/0 -vv</tt>
 

client.py

@@ -1,8 +1,10 @@
-import struct, socket, select, subprocess, errno, re
+import struct, socket, select, errno, re
+import compat.ssubprocess as ssubprocess
 import helpers, ssnet, ssh
 from ssnet import SockWrapper, Handler, Proxy, Mux, MuxWrapper
 from helpers import *
 
+
 def original_dst(sock):
     try:
         SO_ORIGINAL_DST = 80
@@ -45,7 +47,7 @@ class FirewallClient:
         e = None
         for argv in argv_tries:
             try:
-                self.p = subprocess.Popen(argv, stdout=s1, preexec_fn=setup)
+                self.p = ssubprocess.Popen(argv, stdout=s1, preexec_fn=setup)
                 e = None
                 break
             except OSError, e:
@@ -92,7 +94,7 @@ class FirewallClient:
             raise Fatal('cleanup: %r returned %d' % (self.argv, rv))
 
 
-def _main(listener, fw, use_server, remotename, seed_hosts, auto_nets):
+def _main(listener, fw, use_server, remotename, python, seed_hosts, auto_nets):
     handlers = []
     if use_server:
         if helpers.verbose >= 1:
@@ -100,7 +102,13 @@ def _main(listener, fw, use_server, remotename, seed_hosts, auto_nets):
         else:
             helpers.logprefix = 'client: '
         debug1('connecting to server...\n')
-        (serverproc, serversock) = ssh.connect(remotename)
+        try:
+            (serverproc, serversock) = ssh.connect(remotename, python)
+        except socket.error, e:
+            if e.errno == errno.EPIPE:
+                raise Fatal("failed to establish ssh session")
+            else:
+                raise
         mux = Mux(serversock, serversock)
         handlers.append(mux)
 
@@ -169,26 +177,13 @@ def _main(listener, fw, use_server, remotename, seed_hosts, auto_nets):
             if rv:
                 raise Fatal('server died with error code %d' % rv)
         
-        r = set()
-        w = set()
-        x = set()
-        handlers = filter(lambda s: s.ok, handlers)
-        for s in handlers:
-            s.pre_select(r,w,x)
-        debug2('Waiting: %d[%d,%d,%d]...\n' 
-            % (len(handlers), len(r), len(w), len(x)))
-        (r,w,x) = select.select(r,w,x)
-        #log('r=%r w=%r x=%r\n' % (r,w,x))
-        ready = set(r) | set(w) | set(x)
-        for s in handlers:
-            if s.socks & ready:
-                s.callback()
+        ssnet.runonce(handlers, mux)
         if use_server:
             mux.callback()
             mux.check_fullness()
 
 
-def main(listenip, use_server, remotename, seed_hosts, auto_nets,
+def main(listenip, use_server, remotename, python, seed_hosts, auto_nets,
          subnets_include, subnets_exclude):
     debug1('Starting sshuttle proxy.\n')
     listener = socket.socket()
@@ -220,6 +215,6 @@ def main(listenip, use_server, remotename, seed_hosts, auto_nets,
     
     try:
         return _main(listener, fw, use_server, remotename,
-                     seed_hosts, auto_nets)
+                     python, seed_hosts, auto_nets)
     finally:
         fw.done()

firewall.py

@@ -1,11 +1,12 @@
-import subprocess, re, errno
+import re, errno
+import compat.ssubprocess as ssubprocess
 import helpers
 from helpers import *
 
 
 def ipt_chain_exists(name):
     argv = ['iptables', '-t', 'nat', '-nL']
-    p = subprocess.Popen(argv, stdout = subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout = ssubprocess.PIPE)
     for line in p.stdout:
         if line.startswith('Chain %s ' % name):
             return True
@@ -17,7 +18,7 @@ def ipt_chain_exists(name):
 def ipt(*args):
     argv = ['iptables', '-t', 'nat'] + list(args)
     debug1('>> %s\n' % ' '.join(argv))
-    rv = subprocess.call(argv)
+    rv = ssubprocess.call(argv)
     if rv:
         raise Fatal('%r returned %d' % (argv, rv))
 
@@ -64,20 +65,25 @@ def do_iptables(port, subnets):
 
 def ipfw_rule_exists(n):
     argv = ['ipfw', 'list']
-    p = subprocess.Popen(argv, stdout = subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout = ssubprocess.PIPE)
+    found = False
     for line in p.stdout:
         if line.startswith('%05d ' % n):
-            if line.find('ipttl 42') < 0 and line.find('established') < 0:
+            if not ('ipttl 42 setup keep-state' in line
+                    or ('skipto %d' % (n+1)) in line
+                    or 'check-state' in line):
+                log('non-sshuttle ipfw rule: %r\n' % line.strip())
                 raise Fatal('non-sshuttle ipfw rule #%d already exists!' % n)
-            return True
+            found = True
     rv = p.wait()
     if rv:
         raise Fatal('%r returned %d' % (argv, rv))
+    return found
 
 
 def sysctl_get(name):
     argv = ['sysctl', '-n', name]
-    p = subprocess.Popen(argv, stdout = subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout = ssubprocess.PIPE)
     line = p.stdout.readline()
     rv = p.wait()
     if rv:
@@ -91,7 +97,7 @@ def sysctl_get(name):
 def _sysctl_set(name, val):
     argv = ['sysctl', '-w', '%s=%s' % (name, val)]
     debug1('>> %s\n' % ' '.join(argv))
-    rv = subprocess.call(argv, stdout = open('/dev/null', 'w'))
+    rv = ssubprocess.call(argv, stdout = open('/dev/null', 'w'))
 
 
 _oldctls = []
@@ -105,7 +111,7 @@ def sysctl_set(name, val):
 def ipfw(*args):
     argv = ['ipfw', '-q'] + list(args)
     debug1('>> %s\n' % ' '.join(argv))
-    rv = subprocess.call(argv)
+    rv = ssubprocess.call(argv)
     if rv:
         raise Fatal('%r returned %d' % (argv, rv))
 
@@ -124,10 +130,10 @@ def do_ipfw(port, subnets):
 
     if subnets:
         sysctl_set('net.inet.ip.fw.enable', 1)
-        sysctl_set('net.inet.ip.forwarding', 1)
+        sysctl_set('net.inet.ip.scopedroute', 0)
 
-        ipfw('add', sport, 'accept', 'ip',
-             'from', 'any', 'to', 'any', 'established')
+        ipfw('add', sport, 'check-state', 'ip',
+             'from', 'any', 'to', 'any')
         
         # create new subnet entries
         for swidth,sexclude,snet in sorted(subnets, reverse=True):
@@ -139,7 +145,7 @@ def do_ipfw(port, subnets):
                 ipfw('add', sport, 'fwd', '127.0.0.1,%d' % port,
                      'log', 'tcp',
                      'from', 'any', 'to', '%s/%s' % (snet,swidth),
-                     'not', 'ipttl', '42')
+                     'not', 'ipttl', '42', 'keep-state', 'setup')
 
 
 def program_exists(name):

helpers.py

@@ -28,3 +28,10 @@ def debug3(s):
 
 class Fatal(Exception):
     pass
+
+
+def list_contains_any(l, sub):
+    for i in sub:
+        if i in l:
+            return True
+    return False

hostwatch.py

@@ -1,5 +1,6 @@
-import subprocess, time, socket, re, select
+import time, socket, re, select, errno
 if not globals().get('skip_imports'):
+    import compat.ssubprocess as ssubprocess
     import helpers
     from helpers import *
 
@@ -108,7 +109,7 @@ def _check_netstat():
     debug2(' > netstat\n')
     argv = ['netstat', '-n']
     try:
-        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=null)
+        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
         content = p.stdout.read()
         p.wait()
     except OSError, e:
@@ -128,7 +129,7 @@ def _check_smb(hostname):
     argv = ['smbclient', '-U', '%', '-L', hostname]
     debug2(' > smb: %s\n' % hostname)
     try:
-        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=null)
+        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
         lines = p.stdout.readlines()
         p.wait()
     except OSError, e:
@@ -185,7 +186,7 @@ def _check_nmb(hostname, is_workgroup, is_master):
     argv = ['nmblookup'] + ['-M']*is_master + ['--', hostname]
     debug2(' > n%d%d: %s\n' % (is_workgroup, is_master, hostname))
     try:
-        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=null)
+        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
         lines = p.stdout.readlines()
         rv = p.wait()
     except OSError, e:

main.py

@@ -49,17 +49,18 @@ sshuttle [-l [ip:]port] [-r [username@]sshserver[:port]] <subnets...>
 sshuttle --firewall <port> <subnets...>
 sshuttle --server
 --
-l,listen=  transproxy to this ip address and port number [default=0]
+l,listen=  transproxy to this ip address and port number [0.0.0.0:0]
 H,auto-hosts scan for remote hostnames and update local /etc/hosts
 N,auto-nets  automatically determine subnets to route
+python= specify the name/path of the python interpreter on the remote server [python]
 r,remote=  ssh hostname (and optional username) of remote sshuttle server
 x,exclude= exclude this subnet (can be used more than once)
 v,verbose  increase debug message verbosity
 seed-hosts= with -H, use these hostnames for initial scan (comma-separated)
 noserver   don't use a separate server process (mostly for debugging)
-server     [internal use only]
-firewall   [internal use only]
-hostwatch  [internal use only]
+server     (internal use only)
+firewall   (internal use only)
+hostwatch  (internal use only)
 """
 o = options.Options('sshuttle', optspec)
 (opt, flags, extra) = o.parse(sys.argv[1:])
@@ -99,6 +100,7 @@ try:
         sys.exit(client.main(parse_ipport(opt.listen or '0.0.0.0:0'),
                              not opt.noserver,
                              remotename,
+                             (opt.python or "python"),
                              sh,
                              opt.auto_nets,
                              parse_subnets(includes),

options.py

@@ -1,30 +1,94 @@
-import sys, textwrap, getopt, re
+"""Command-line options parser.
+With the help of an options spec string, easily parse command-line options.
+"""
+import sys, os, textwrap, getopt, re, struct
 
 class OptDict:
     def __init__(self):
         self._opts = {}
 
     def __setitem__(self, k, v):
+        if k.startswith('no-') or k.startswith('no_'):
+            k = k[3:]
+            v = not v
         self._opts[k] = v
-        
+
     def __getitem__(self, k):
+        if k.startswith('no-') or k.startswith('no_'):
+            return not self._opts[k[3:]]
         return self._opts[k]
 
     def __getattr__(self, k):
         return self[k]
 
 
+def _default_onabort(msg):
+    sys.exit(97)
+
+
+def _intify(v):
+    try:
+        vv = int(v or '')
+        if str(vv) == v:
+            return vv
+    except ValueError:
+        pass
+    return v
+
+
+def _atoi(v):
+    try:
+        return int(v or 0)
+    except ValueError:
+        return 0
+
+
+def _remove_negative_kv(k, v):
+    if k.startswith('no-') or k.startswith('no_'):
+        return k[3:], not v
+    return k,v
+
+def _remove_negative_k(k):
+    return _remove_negative_kv(k, None)[0]
+
+
+def _tty_width():
+    s = struct.pack("HHHH", 0, 0, 0, 0)
+    try:
+        import fcntl, termios
+        s = fcntl.ioctl(sys.stderr.fileno(), termios.TIOCGWINSZ, s)
+    except (IOError, ImportError):
+        return _atoi(os.environ.get('WIDTH')) or 70
+    (ysize,xsize,ypix,xpix) = struct.unpack('HHHH', s)
+    return xsize
+
+
 class Options:
-    def __init__(self, exe, optspec, optfunc=getopt.gnu_getopt):
+    """Option parser.
+    When constructed, two strings are mandatory. The first one is the command
+    name showed before error messages. The second one is a string called an
+    optspec that specifies the synopsis and option flags and their description.
+    For more information about optspecs, consult the bup-options(1) man page.
+
+    Two optional arguments specify an alternative parsing function and an
+    alternative behaviour on abort (after having output the usage string).
+
+    By default, the parser function is getopt.gnu_getopt, and the abort
+    behaviour is to exit the program.
+    """
+    def __init__(self, exe, optspec, optfunc=getopt.gnu_getopt,
+                 onabort=_default_onabort):
         self.exe = exe
         self.optspec = optspec
+        self._onabort = onabort
         self.optfunc = optfunc
         self._aliases = {}
         self._shortopts = 'h?'
         self._longopts = ['help']
         self._hasparms = {}
+        self._defaults = {}
         self._usagestr = self._gen_usage()
-        
+
     def _gen_usage(self):
         out = []
         lines = self.optspec.strip().split('\n')
@@ -36,10 +100,13 @@ class Options:
             out.append('%s: %s\n' % (first_syn and 'usage' or '   or', l))
             first_syn = False
         out.append('\n')
+        last_was_option = False
         while lines:
             l = lines.pop()
             if l.startswith(' '):
-                out.append('\n%s\n' % l.lstrip())
+                out.append('%s%s\n' % (last_was_option and '\n' or '',
+                                       l.lstrip()))
+                last_was_option = False
             elif l:
                 (flags, extra) = l.split(' ', 1)
                 extra = extra.strip()
@@ -48,18 +115,24 @@ class Options:
                     has_parm = 1
                 else:
                     has_parm = 0
+                g = re.search(r'\[([^\]]*)\]$', extra)
+                if g:
+                    defval = g.group(1)
+                else:
+                    defval = None
                 flagl = flags.split(',')
                 flagl_nice = []
                 for f in flagl:
-                    f_nice = re.sub(r'\W', '_', f)
-                    self._aliases[f] = flagl[0]
-                    self._aliases[f_nice] = flagl[0]
+                    f,dvi = _remove_negative_kv(f, _intify(defval))
+                    self._aliases[f] = _remove_negative_k(flagl[0])
                     self._hasparms[f] = has_parm
+                    self._defaults[f] = dvi
                     if len(f) == 1:
                         self._shortopts += f + (has_parm and ':' or '')
                         flagl_nice.append('-' + f)
                     else:
-                        assert(not f.startswith('no-')) # supported implicitly
+                        f_nice = re.sub(r'\W', '_', f)
+                        self._aliases[f_nice] = _remove_negative_k(flagl[0])
                         self._longopts.append(f + (has_parm and '=' or ''))
                         self._longopts.append('no-' + f)
                         flagl_nice.append('--' + f)
@@ -67,52 +140,62 @@ class Options:
                 if has_parm:
                     flags_nice += ' ...'
                 prefix = '    %-20s  ' % flags_nice
-                argtext = '\n'.join(textwrap.wrap(extra, width=70,
+                argtext = '\n'.join(textwrap.wrap(extra, width=_tty_width(),
                                                 initial_indent=prefix,
                                                 subsequent_indent=' '*28))
                 out.append(argtext + '\n')
+                last_was_option = True
             else:
                 out.append('\n')
+                last_was_option = False
         return ''.join(out).rstrip() + '\n'
-    
-    def usage(self):
+
+    def usage(self, msg=""):
+        """Print usage string to stderr and abort."""
         sys.stderr.write(self._usagestr)
-        sys.exit(97)
+        e = self._onabort and self._onabort(msg) or None
+        if e:
+            raise e
 
     def fatal(self, s):
-        sys.stderr.write('error: %s\n' % s)
-        return self.usage()
-        
+        """Print an error message to stderr and abort with usage string."""
+        msg = 'error: %s\n' % s
+        sys.stderr.write(msg)
+        return self.usage(msg)
+
     def parse(self, args):
+        """Parse a list of arguments and return (options, flags, extra).
+
+        In the returned tuple, "options" is an OptDict with known options,
+        "flags" is a list of option flags that were used on the command-line,
+        and "extra" is a list of positional arguments.
+        """
         try:
             (flags,extra) = self.optfunc(args, self._shortopts, self._longopts)
         except getopt.GetoptError, e:
             self.fatal(e)
 
         opt = OptDict()
-        for f in self._aliases.values():
-            opt[f] = None
+
+        for k,v in self._defaults.iteritems():
+            k = self._aliases[k]
+            opt[k] = v
+
         for (k,v) in flags:
-            while k.startswith('-'):
-                k = k[1:]
-            if k in ['h', '?', 'help']:
+            k = k.lstrip('-')
+            if k in ('h', '?', 'help'):
                 self.usage()
             if k.startswith('no-'):
                 k = self._aliases[k[3:]]
-                opt[k] = None
+                v = 0
             else:
                 k = self._aliases[k]
                 if not self._hasparms[k]:
                     assert(v == '')
-                    opt[k] = (opt._opts.get(k) or 0) + 1
+                    v = (opt._opts.get(k) or 0) + 1
                 else:
-                    try:
-                        vv = int(v)
-                        if str(vv) == v:
-                            v = vv
-                    except ValueError:
-                        pass
-                    opt[k] = v
-        for (f1,f2) in self._aliases.items():
-            opt[f1] = opt[f2]
+                    v = _intify(v)
+            opt[k] = v
+        for (f1,f2) in self._aliases.iteritems():
+            opt[f1] = opt._opts.get(f2)
         return (opt,flags,extra)

server.py

@@ -1,6 +1,7 @@
-import re, struct, socket, select, subprocess, traceback
+import re, struct, socket, select, traceback
 if not globals().get('skip_imports'):
     import ssnet, helpers, hostwatch
+    import compat.ssubprocess as ssubprocess
     from ssnet import SockWrapper, Handler, Proxy, Mux, MuxWrapper
     from helpers import *
 
@@ -36,14 +37,18 @@ def _maskbits(netmask):
     if not netmask:
         return 32
     for i in range(32):
-        if netmask[0] & (1<<i):
+        if netmask[0] & _shl(1, i):
             return 32-i
     return 0
+    
+    
+def _shl(n, bits):
+    return n * int(2**bits)
 
 
 def _list_routes():
     argv = ['netstat', '-rn']
-    p = subprocess.Popen(argv, stdout=subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE)
     routes = []
     for line in p.stdout:
         cols = re.split(r'\s+', line)
@@ -53,11 +58,12 @@ def _list_routes():
         maskw = _ipmatch(cols[2])  # linux only
         mask = _maskbits(maskw)   # returns 32 if maskw is null
         width = min(ipw[1], mask)
-        ip = ipw[0] & (((1<<width)-1) << (32-width))
+        ip = ipw[0] & _shl(_shl(1, width) - 1, 32-width)
         routes.append((socket.inet_ntoa(struct.pack('!I', ip)), width))
     rv = p.wait()
     if rv != 0:
-        raise Fatal('%r returned %d' % (argv, rv))
+        log('WARNING: %r returned %d\n' % (argv, rv))
+        log('WARNING: That prevents --auto-nets from working.\n')
     return routes
 
 
@@ -121,8 +127,9 @@ def main():
               socket.fromfd(sys.stdout.fileno(),
                             socket.AF_INET, socket.SOCK_STREAM))
     handlers.append(mux)
-    routepkt = ''.join('%s,%d\n' % r
-                       for r in routes)
+    routepkt = ''
+    for r in routes:
+        routepkt += '%s,%d\n' % r
     mux.send(0, ssnet.CMD_ROUTES, routepkt)
 
     hw = Hostwatch()
@@ -155,21 +162,6 @@ def main():
             if rpid:
                 raise Fatal('hostwatch exited unexpectedly: code 0x%04x\n' % rv)
         
-        r = set()
-        w = set()
-        x = set()
-        handlers = filter(lambda s: s.ok, handlers)
-        for s in handlers:
-            s.pre_select(r,w,x)
-        debug2('Waiting: %d[%d,%d,%d] (fullness=%d/%d)...\n' 
-               % (len(handlers), len(r), len(w), len(x),
-                  mux.fullness, mux.too_full))
-        (r,w,x) = select.select(r,w,x)
-        #log('r=%r w=%r x=%r\n' % (r,w,x))
-        ready = set(r) | set(w) | set(x)
-        for s in handlers:
-            #debug2('check: %r: %r\n' % (s, s.socks & ready))
-            if s.socks & ready:
-                s.callback()
+        ssnet.runonce(handlers, mux)
         mux.check_fullness()
         mux.callback()

ssh.py

@@ -1,4 +1,5 @@
-import sys, os, re, subprocess, socket, zlib
+import sys, os, re, socket, zlib
+import compat.ssubprocess as ssubprocess
 import helpers
 from helpers import *
 
@@ -14,12 +15,13 @@ def readfile(name):
 
 
 def empackage(z, filename):
+    (path,basename) = os.path.split(filename)
     content = z.compress(readfile(filename))
     content += z.flush(zlib.Z_SYNC_FLUSH)
-    return '%s\n%d\n%s' % (filename,len(content), content)
+    return '%s\n%d\n%s' % (basename,len(content), content)
 
 
-def connect(rhostport):
+def connect(rhostport, python):
     main_exe = sys.argv[0]
     l = (rhostport or '').split(':', 1)
     rhost = l[0]
@@ -33,6 +35,7 @@ def connect(rhostport):
     z = zlib.compressobj(1)
     content = readfile('assembler.py')
     content2 = (empackage(z, 'helpers.py') +
+                empackage(z, 'compat/ssubprocess.py') +
                 empackage(z, 'ssnet.py') +
                 empackage(z, 'hostwatch.py') +
                 empackage(z, 'server.py') +
@@ -48,18 +51,17 @@ def connect(rhostport):
 
         
     if not rhost:
-        argv = ['python', '-c', pyscript]
+        argv = [python, '-c', pyscript]
     else:
-        argv = ['ssh'] + portl + [rhost, '--', "python -c '%s'" % pyscript]
+        argv = ['ssh'] + portl + [rhost, '--', "'%s' -c '%s'" % (python, pyscript)]
     (s1,s2) = socket.socketpair()
     def setup():
         # runs in the child process
         s2.close()
-        os.setsid()
     s1a,s1b = os.dup(s1.fileno()), os.dup(s1.fileno())
     s1.close()
     debug2('executing: %r\n' % argv)
-    p = subprocess.Popen(argv, stdin=s1a, stdout=s1b, preexec_fn=setup,
+    p = ssubprocess.Popen(argv, stdin=s1a, stdout=s1b, preexec_fn=setup,
                          close_fds=True)
     os.close(s1a)
     os.close(s1b)

ssnet.py

@@ -1,6 +1,12 @@
 import struct, socket, errno, select
 if not globals().get('skip_imports'):
     from helpers import *
+    
+# these don't exist in the socket module in python 2.3!
+SHUT_RD = 0
+SHUT_WR = 1
+SHUT_RDWR = 2
+
 
 HDR_LEN = 8
 
@@ -31,6 +37,22 @@ cmd_to_name = {
     
 
 
+def _add(l, elem):
+    if not elem in l:
+        l.append(elem)
+
+
+def _fds(l):
+    out = []
+    for i in l:
+        try:
+            out.append(i.fileno())
+        except AttributeError:
+            out.append(i)
+    out.sort()
+    return out
+
+
 def _nb_clean(func, *args):
     try:
         return func(*args)
@@ -38,6 +60,7 @@ def _nb_clean(func, *args):
         if e.errno not in (errno.EWOULDBLOCK, errno.EAGAIN):
             raise
         else:
+            debug3('%s: err was: %s\n' % (func.__name__, e))
             return None
 
 
@@ -69,21 +92,30 @@ class SockWrapper:
             debug1('%r: error was: %r\n' % (self, self.exc))
 
     def __repr__(self):
-        return 'SW:%s' % (self.peername,)
+        if self.rsock == self.wsock:
+            fds = '#%d' % self.rsock.fileno()
+        else:
+            fds = '#%d,%d' % (self.rsock.fileno(), self.wsock.fileno())
+        return 'SW%s:%s' % (fds, self.peername)
 
     def seterr(self, e):
         if not self.exc:
             self.exc = e
 
     def try_connect(self):
+        if self.connect_to and self.shut_write:
+            self.noread()
+            self.connect_to = None
         if not self.connect_to:
             return  # already connected
         self.rsock.setblocking(False)
+        debug3('%r: trying connect to %r\n' % (self, self.connect_to))
         try:
             self.rsock.connect(self.connect_to)
             # connected successfully (Linux)
             self.connect_to = None
         except socket.error, e:
+            debug3('%r: connect result: %r\n' % (self, e))
             if e.args[0] in [errno.EINPROGRESS, errno.EALREADY]:
                 pass  # not connected yet
             elif e.args[0] == errno.EISCONN:
@@ -102,14 +134,14 @@ class SockWrapper:
         if not self.shut_read:
             debug2('%r: done reading\n' % self)
             self.shut_read = True
-            #self.rsock.shutdown(socket.SHUT_RD)  # doesn't do anything anyway
+            #self.rsock.shutdown(SHUT_RD)  # doesn't do anything anyway
         
     def nowrite(self):
         if not self.shut_write:
             debug2('%r: done writing\n' % self)
             self.shut_write = True
             try:
-                self.wsock.shutdown(socket.SHUT_WR)
+                self.wsock.shutdown(SHUT_WR)
             except socket.error, e:
                 self.seterr(e)
 
@@ -159,7 +191,7 @@ class SockWrapper:
             wrote = outwrap.write(self.buf[0])
             self.buf[0] = self.buf[0][wrote:]
         while self.buf and not self.buf[0]:
-            self.buf[0:1] = []
+            self.buf.pop(0)
         if not self.buf and self.shut_read:
             outwrap.nowrite()
 
@@ -167,12 +199,13 @@ class SockWrapper:
 class Handler:
     def __init__(self, socks = None, callback = None):
         self.ok = True
-        self.socks = set(socks or [])
+        self.socks = socks or []
         if callback:
             self.callback = callback
 
     def pre_select(self, r, w, x):
-        r |= self.socks
+        for i in self.socks:
+            _add(r, i)
 
     def callback(self):
         log('--no callback defined-- %r\n' % self)
@@ -181,7 +214,7 @@ class Handler:
             v = s.recv(4096)
             if not v:
                 log('--closed-- %r\n' % self)
-                self.socks = set()
+                self.socks = []
                 self.ok = False
 
 
@@ -194,20 +227,20 @@ class Proxy(Handler):
 
     def pre_select(self, r, w, x):
         if self.wrap1.connect_to:
-            w.add(self.wrap1.rsock)
+            _add(w, self.wrap1.rsock)
         elif self.wrap1.buf:
             if not self.wrap2.too_full():
-                w.add(self.wrap2.wsock)
+                _add(w, self.wrap2.wsock)
         elif not self.wrap1.shut_read:
-            r.add(self.wrap1.rsock)
+            _add(r, self.wrap1.rsock)
 
         if self.wrap2.connect_to:
-            w.add(self.wrap2.rsock)
+            _add(w, self.wrap2.rsock)
         elif self.wrap2.buf:
             if not self.wrap1.too_full():
-                w.add(self.wrap1.wsock)
+                _add(w, self.wrap1.wsock)
         elif not self.wrap2.shut_read:
-            r.add(self.wrap2.rsock)
+            _add(r, self.wrap2.rsock)
 
     def callback(self):
         self.wrap1.try_connect()
@@ -216,6 +249,12 @@ class Proxy(Handler):
         self.wrap2.fill()
         self.wrap1.copy_to(self.wrap2)
         self.wrap2.copy_to(self.wrap1)
+        if self.wrap1.buf and self.wrap2.shut_write:
+            self.wrap1.buf = []
+            self.wrap1.noread()
+        if self.wrap2.buf and self.wrap1.shut_write:
+            self.wrap2.buf = []
+            self.wrap2.noread()
         if (self.wrap1.shut_read and self.wrap2.shut_read and
             not self.wrap1.buf and not self.wrap2.buf):
             self.ok = False
@@ -247,7 +286,10 @@ class Mux(Handler):
                 return self.chani
 
     def amount_queued(self):
-        return sum(len(b) for b in self.outbuf)
+        total = 0
+        for b in self.outbuf:
+            total += len(b)
+        return total
             
     def check_fullness(self):
         if self.fullness > 32768:
@@ -308,7 +350,7 @@ class Mux(Handler):
         self.wsock.setblocking(False)
         if self.outbuf and self.outbuf[0]:
             wrote = _nb_clean(os.write, self.wsock.fileno(), self.outbuf[0])
-            debug2('mux wrote: %d/%d\n' % (wrote, len(self.outbuf[0])))
+            debug2('mux wrote: %r/%d\n' % (wrote, len(self.outbuf[0])))
             if wrote:
                 self.outbuf[0] = self.outbuf[0][wrote:]
         while self.outbuf and not self.outbuf[0]:
@@ -346,9 +388,9 @@ class Mux(Handler):
                 break
 
     def pre_select(self, r, w, x):
-        r.add(self.rsock)
+        _add(r, self.rsock)
         if self.outbuf:
-            w.add(self.wsock)
+            _add(w, self.wsock)
 
     def callback(self):
         (r,w,x) = select.select([self.rsock], [self.wsock], [], 0)
@@ -420,3 +462,28 @@ def connect_dst(ip, port):
     return SockWrapper(outsock, outsock,
                        connect_to = (ip,port),
                        peername = '%s:%d' % (ip,port))
+
+
+def runonce(handlers, mux):
+    r = []
+    w = []
+    x = []
+    handlers = filter(lambda s: s.ok, handlers)
+    for s in handlers:
+        s.pre_select(r,w,x)
+    debug2('Waiting: %d r=%r w=%r x=%r (fullness=%d/%d)\n' 
+            % (len(handlers), _fds(r), _fds(w), _fds(x),
+               mux.fullness, mux.too_full))
+    (r,w,x) = select.select(r,w,x)
+    debug2('  Ready: %d r=%r w=%r x=%r\n' 
+        % (len(handlers), _fds(r), _fds(w), _fds(x)))
+    ready = r+w+x
+    did = {}
+    for h in handlers:
+        for s in h.socks:
+            if s in ready:
+                h.callback()
+                did[s] = 1
+    for s in ready:
+        if not s in did:
+            raise Fatal('socket %r was not used by any handler' % s)